Spaces:
Running
Running
Jeremiah Lowin commited on
Commit ·
b237957
1
Parent(s): 329333a
Switch tests to use http
Browse files
tests/server/test_http_dependencies.py
CHANGED
|
@@ -7,7 +7,7 @@ import uvicorn
|
|
| 7 |
from mcp.types import TextContent, TextResourceContents
|
| 8 |
|
| 9 |
from fastmcp.client import Client
|
| 10 |
-
from fastmcp.client.transports import
|
| 11 |
from fastmcp.server.dependencies import get_http_request
|
| 12 |
from fastmcp.server.server import FastMCP
|
| 13 |
from fastmcp.utilities.tests import run_server_in_process
|
|
@@ -43,9 +43,15 @@ def fastmcp_server():
|
|
| 43 |
|
| 44 |
def run_server(host: str, port: int) -> None:
|
| 45 |
try:
|
| 46 |
-
app = fastmcp_server().
|
| 47 |
server = uvicorn.Server(
|
| 48 |
-
config=uvicorn.Config(
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 49 |
)
|
| 50 |
server.run()
|
| 51 |
except Exception as e:
|
|
@@ -57,13 +63,13 @@ def run_server(host: str, port: int) -> None:
|
|
| 57 |
@pytest.fixture(autouse=True, scope="module")
|
| 58 |
def sse_server() -> Generator[str, None, None]:
|
| 59 |
with run_server_in_process(run_server) as url:
|
| 60 |
-
yield f"{url}/
|
| 61 |
|
| 62 |
|
| 63 |
async def test_http_headers_resource(sse_server: str):
|
| 64 |
"""Test getting HTTP headers from the server."""
|
| 65 |
async with Client(
|
| 66 |
-
transport=
|
| 67 |
) as client:
|
| 68 |
raw_result = await client.read_resource("request://headers")
|
| 69 |
assert isinstance(raw_result[0], TextResourceContents)
|
|
@@ -75,7 +81,7 @@ async def test_http_headers_resource(sse_server: str):
|
|
| 75 |
async def test_http_headers_tool(sse_server: str):
|
| 76 |
"""Test getting HTTP headers from the server."""
|
| 77 |
async with Client(
|
| 78 |
-
transport=
|
| 79 |
) as client:
|
| 80 |
result = await client.call_tool("get_headers_tool")
|
| 81 |
assert isinstance(result[0], TextContent)
|
|
@@ -87,7 +93,7 @@ async def test_http_headers_tool(sse_server: str):
|
|
| 87 |
async def test_http_headers_prompt(sse_server: str):
|
| 88 |
"""Test getting HTTP headers from the server."""
|
| 89 |
async with Client(
|
| 90 |
-
transport=
|
| 91 |
) as client:
|
| 92 |
result = await client.get_prompt("get_headers_prompt")
|
| 93 |
assert isinstance(result.messages[0].content, TextContent)
|
|
|
|
| 7 |
from mcp.types import TextContent, TextResourceContents
|
| 8 |
|
| 9 |
from fastmcp.client import Client
|
| 10 |
+
from fastmcp.client.transports import StreamableHttpTransport
|
| 11 |
from fastmcp.server.dependencies import get_http_request
|
| 12 |
from fastmcp.server.server import FastMCP
|
| 13 |
from fastmcp.utilities.tests import run_server_in_process
|
|
|
|
| 43 |
|
| 44 |
def run_server(host: str, port: int) -> None:
|
| 45 |
try:
|
| 46 |
+
app = fastmcp_server().streamable_http_app()
|
| 47 |
server = uvicorn.Server(
|
| 48 |
+
config=uvicorn.Config(
|
| 49 |
+
app=app,
|
| 50 |
+
host=host,
|
| 51 |
+
port=port,
|
| 52 |
+
log_level="error",
|
| 53 |
+
lifespan="on",
|
| 54 |
+
)
|
| 55 |
)
|
| 56 |
server.run()
|
| 57 |
except Exception as e:
|
|
|
|
| 63 |
@pytest.fixture(autouse=True, scope="module")
|
| 64 |
def sse_server() -> Generator[str, None, None]:
|
| 65 |
with run_server_in_process(run_server) as url:
|
| 66 |
+
yield f"{url}/mcp"
|
| 67 |
|
| 68 |
|
| 69 |
async def test_http_headers_resource(sse_server: str):
|
| 70 |
"""Test getting HTTP headers from the server."""
|
| 71 |
async with Client(
|
| 72 |
+
transport=StreamableHttpTransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
|
| 73 |
) as client:
|
| 74 |
raw_result = await client.read_resource("request://headers")
|
| 75 |
assert isinstance(raw_result[0], TextResourceContents)
|
|
|
|
| 81 |
async def test_http_headers_tool(sse_server: str):
|
| 82 |
"""Test getting HTTP headers from the server."""
|
| 83 |
async with Client(
|
| 84 |
+
transport=StreamableHttpTransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
|
| 85 |
) as client:
|
| 86 |
result = await client.call_tool("get_headers_tool")
|
| 87 |
assert isinstance(result[0], TextContent)
|
|
|
|
| 93 |
async def test_http_headers_prompt(sse_server: str):
|
| 94 |
"""Test getting HTTP headers from the server."""
|
| 95 |
async with Client(
|
| 96 |
+
transport=StreamableHttpTransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
|
| 97 |
) as client:
|
| 98 |
result = await client.get_prompt("get_headers_prompt")
|
| 99 |
assert isinstance(result.messages[0].content, TextContent)
|