2 2
import os
3 3
from dotenv import load_dotenv
4 4
from mcp.client.sse import sse_client
5 -
from mcp import ClientSession
5 +
from mcp.client.stdio import stdio_client
6 +
from mcp.client.streamable_http import streamablehttp_client
7 +
from mcp import ClientSession, StdioServerParameters
6 8 7 9
load_dotenv()
8 10 9 11
DEFAULT_GRAFANA_URL = "http://localhost:3000"
10 -
DEFAULT_MCP_URL = "http://localhost:8000/sse"
12 +
DEFAULT_MCP_URL = "http://localhost:8000"
13 +
DEFAULT_MCP_TRANSPORT = "sse"
11 14 12 15
models = ["gpt-4o", "claude-3-5-sonnet-20240620"]
13 16 17 + 14 18
@pytest.fixture
15 19
def anyio_backend():
16 20
return "asyncio"
17 21 18 22 23 +
@pytest.fixture
24 +
def mcp_transport():
25 +
return os.environ.get("MCP_TRANSPORT", DEFAULT_MCP_TRANSPORT)
26 + 27 + 19 28
@pytest.fixture
20 29
def mcp_url():
21 30
return os.environ.get("MCP_GRAFANA_URL", DEFAULT_MCP_URL)
22 31 32 + 33 +
@pytest.fixture
34 +
def grafana_env():
35 +
env = {"GRAFANA_URL": os.environ.get("GRAFANA_URL", DEFAULT_GRAFANA_URL)}
36 +
if key := os.environ.get("GRAFANA_API_KEY"):
37 +
env["GRAFANA_API_KEY"] = key
38 +
return env
39 + 40 + 23 41
@pytest.fixture
24 42
def grafana_headers():
25 43
headers = {
@@ -29,23 +47,35 @@ def grafana_headers():
29 47
headers["X-Grafana-API-Key"] = key
30 48
return headers
31 49 32 -
@pytest.fixture
33 -
async def mcp_client(mcp_url, grafana_headers):
34 -
async with sse_client(mcp_url, headers=grafana_headers) as (read, write):
35 -
async with ClientSession(read, write) as session:
36 -
await session.initialize()
37 -
yield session
38 - 39 -
@pytest.fixture
40 -
def mcp_url():
41 -
return os.environ.get("MCP_GRAFANA_URL", DEFAULT_MCP_URL)
42 50 43 51
@pytest.fixture
44 -
async def mcp_client(mcp_url, grafana_headers):
45 -
async with sse_client(mcp_url, headers=grafana_headers) as (
46 -
read,
47 -
write,
48 -
):
49 -
async with ClientSession(read, write) as session:
50 -
await session.initialize()
51 -
yield session
52 +
async def mcp_client(mcp_transport, mcp_url, grafana_env, grafana_headers):
53 +
if mcp_transport == "stdio":
54 +
params = StdioServerParameters(
55 +
command=os.environ.get("MCP_GRAFANA_PATH", "../dist/mcp-grafana"),
56 +
args=["--debug", "--log-level", "debug"],
57 +
env=grafana_env,
58 +
)
59 +
async with stdio_client(params) as (read, write):
60 +
async with ClientSession(read, write) as session:
61 +
await session.initialize()
62 +
yield session
63 +
elif mcp_transport == "sse":
64 +
url = f"{mcp_url}/sse"
65 +
async with sse_client(url, headers=grafana_headers) as (read, write):
66 +
async with ClientSession(read, write) as session:
67 +
await session.initialize()
68 +
yield session
69 +
elif mcp_transport == "streamable-http":
70 +
# Use HTTP client for streamable-http transport
71 +
url = f"{mcp_url}/mcp"
72 +
async with streamablehttp_client(url, headers=grafana_headers) as (
73 +
read,
74 +
write,
75 +
_,
76 +
):
77 +
async with ClientSession(read, write) as session:
78 +
await session.initialize()
79 +
yield session
80 +
else:
81 +
raise ValueError(f"Unsupported transport: {mcp_transport}")
RetroSearch is an open source project built by @garambo | Open a GitHub Issue
Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo
HTML:
3.2
| Encoding:
UTF-8
| Version:
0.7.4