ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-22 13:34:17 +02:00
Code Issues Projects Releases Packages Wiki Activity

Switch tests to use http

Browse source
This commit is contained in:
Jeremiah Lowin 2025-05-07 21:06:00 -04:00
commit 4355112cb3
1 changed files with 13 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

20
tests/server/test_http_dependencies.py
View file

@ -7,7 +7,7 @@ import uvicorn
from mcp.types import TextContent, TextResourceContents
from fastmcp.client import Client
from fastmcp.client.transports import SSETransport
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.server.dependencies import get_http_request
from fastmcp.server.server import FastMCP
from fastmcp.utilities.tests import run_server_in_process
@ -43,9 +43,15 @@ def fastmcp_server():
def run_server(host: str, port: int) -> None:
try:
app = fastmcp_server().sse_app()
app = fastmcp_server().streamable_http_app()
server = uvicorn.Server(
config=uvicorn.Config(app=app, host=host, port=port, log_level="error")
config=uvicorn.Config(
app=app,
host=host,
port=port,
log_level="error",
lifespan="on",
)
)
server.run()
except Exception as e:
@ -57,13 +63,13 @@ def run_server(host: str, port: int) -> None:
@pytest.fixture(autouse=True, scope="module")
def sse_server() -> Generator[str, None, None]:
with run_server_in_process(run_server) as url:
yield f"{url}/sse"
yield f"{url}/mcp"
async def test_http_headers_resource(sse_server: str):
"""Test getting HTTP headers from the server."""
async with Client(
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
transport=StreamableHttpTransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
) as client:
raw_result = await client.read_resource("request://headers")
assert isinstance(raw_result[0], TextResourceContents)
@ -75,7 +81,7 @@ async def test_http_headers_resource(sse_server: str):
async def test_http_headers_tool(sse_server: str):
"""Test getting HTTP headers from the server."""
async with Client(
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
transport=StreamableHttpTransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
) as client:
result = await client.call_tool("get_headers_tool")
assert isinstance(result[0], TextContent)
@ -87,7 +93,7 @@ async def test_http_headers_tool(sse_server: str):
async def test_http_headers_prompt(sse_server: str):
"""Test getting HTTP headers from the server."""
async with Client(
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
transport=StreamableHttpTransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
) as client:
result = await client.get_prompt("get_headers_prompt")
assert isinstance(result.messages[0].content, TextContent)