test nested servers

This commit is contained in:
Jeremiah Lowin 2025-05-09 17:27:22 -04:00
commit a0939e57fe
3 changed files with 58 additions and 15 deletions

View file

@ -83,7 +83,7 @@ asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
asyncio_default_test_loop_scope = "session"
filterwarnings = []
timeout = 5
timeout = 3
[tool.pyright]
include = ["src", "tests"]

View file

@ -110,7 +110,7 @@ def setup_auth_middleware_and_routes(
def create_base_app(
routes: list[Route | Mount],
middleware: list[Middleware],
debug: bool,
debug: bool = False,
lifespan: Callable | None = None,
) -> Starlette:
"""Create a base Starlette app with common middleware and routes.
@ -127,17 +127,12 @@ def create_base_app(
# Always add RequestContextMiddleware as the outermost middleware
middleware.append(Middleware(RequestContextMiddleware))
# Create the app
app_kwargs = {
"debug": debug,
"routes": routes,
"middleware": middleware,
}
if lifespan:
app_kwargs["lifespan"] = lifespan
return Starlette(**app_kwargs)
return Starlette(
routes=routes,
middleware=middleware,
debug=debug,
lifespan=lifespan,
)
def create_sse_app(
@ -224,7 +219,11 @@ def create_sse_app(
routes.extend(cast(list[Route | Mount], additional_routes))
# Create and return the app
return create_base_app(routes, middleware, debug)
return create_base_app(
routes=routes,
middleware=middleware,
debug=debug,
)
def create_streamable_http_app(
@ -305,4 +304,9 @@ def create_streamable_http_app(
yield
# Create and return the app with lifespan
return create_base_app(routes, middleware, debug, lifespan)
return create_base_app(
routes=routes,
middleware=middleware,
debug=debug,
lifespan=lifespan,
)

View file

@ -5,6 +5,8 @@ from collections.abc import Generator
import pytest
import uvicorn
from mcp.types import TextResourceContents
from starlette.applications import Starlette
from starlette.routing import Mount
from fastmcp.client import Client
from fastmcp.client.transports import StreamableHttpTransport
@ -100,3 +102,40 @@ async def test_http_headers(streamable_http_server: str):
json_result = json.loads(raw_result[0].text)
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
def run_nested_server(host: str, port: int) -> None:
try:
mcp_app = fastmcp_server().streamable_http_app()
mount = Starlette(routes=[Mount("/nest-inner", app=mcp_app)])
mount2 = Starlette(
routes=[Mount("/nest-outer", app=mount)],
lifespan=mcp_app.router.lifespan_context,
)
server = uvicorn.Server(
config=uvicorn.Config(
app=mount2,
host=host,
port=port,
log_level="error",
lifespan="on",
)
)
server.run()
except Exception as e:
print(f"Server error: {e}")
sys.exit(1)
sys.exit(0)
async def test_nested_streamable_http_server_resolves_correctly():
# tests patch for
# https://github.com/modelcontextprotocol/python-sdk/pull/659
with run_server_in_process(run_nested_server) as url:
async with Client(
transport=StreamableHttpTransport(f"{url}/nest-outer/nest-inner/mcp")
) as client:
result = await client.ping()
assert result is True