From 454ba6de7c773d932363442424c207ae9a1903f7 Mon Sep 17 00:00:00 2001 From: Justin Wilson Date: Mon, 24 Aug 2026 17:37:18 -0500 Subject: [PATCH] fix(server): register SSE GET as raw ASGI so BaseHTTPMiddleware does not see a second response start Starlette wraps function endpoints in request_response. After EventSourceResponse has already finalized the SSE stream, that wrapper sent a second http.response.start (empty Response), which BaseHTTPMiddleware asserts against on client disconnect (#883). --- src/mcp/server/mcpserver/server.py | 31 +++++++++++------- src/mcp/server/sse.py | 29 +++++++++-------- tests/server/mcpserver/test_server.py | 45 +++++++++++++++++++++++++++ 3 files changed, 78 insertions(+), 27 deletions(-) diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py index 93bef1655a..b8cb970c90 100644 --- a/src/mcp/server/mcpserver/server.py +++ b/src/mcp/server/mcpserver/server.py @@ -52,7 +52,7 @@ from starlette.requests import Request from starlette.responses import Response from starlette.routing import Mount, Route -from starlette.types import Receive, Scope, Send +from starlette.types import ASGIApp, Receive, Scope, Send from mcp.server.auth.middleware.auth_context import AuthContextMiddleware from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware @@ -138,6 +138,17 @@ class Settings(BaseModel, Generic[LifespanResultT]): ) +class _RawASGIEndpoint: + """Passes a raw ASGI callable to `Route`, which treats plain functions as request/response + endpoints and would send a second response after the handler already sent one (#883).""" + + def __init__(self, app: ASGIApp) -> None: + self._app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + await self._app(scope, receive, send) + + def lifespan_wrapper( app: MCPServer[LifespanResultT], lifespan: Callable[[MCPServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]], @@ -1153,14 +1164,13 @@ def sse_app( message_path, security_settings=transport_security, max_request_body_size=max_request_body_size ) - async def handle_sse(scope: Scope, receive: Receive, send: Send): # pragma: no cover - # Add client ID from auth context into request context if available - + async def handle_sse(scope: Scope, receive: Receive, send: Send) -> None: + # connect_sse sends the complete SSE response; sending anything after it would + # deliver a second http.response.start, which crashes BaseHTTPMiddleware (#883). async with sse.connect_sse(scope, receive, send) as streams: await self._lowlevel_server.run( streams[0], streams[1], self._lowlevel_server.create_initialization_options() ) - return Response() # Create routes routes: list[Route | Mount] = [] @@ -1213,7 +1223,9 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send): # pragma: no routes.append( Route( sse_path, - endpoint=RequireAuthMiddleware(handle_sse, required_scopes, resource_metadata_url), + endpoint=RequireAuthMiddleware( + _RawASGIEndpoint(handle_sse), required_scopes, resource_metadata_url + ), methods=["GET"], ) ) @@ -1225,15 +1237,10 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send): # pragma: no ) else: # Auth is disabled, no need for RequireAuthMiddleware - # Since handle_sse is an ASGI app, we need to create a compatible endpoint - async def sse_endpoint(request: Request) -> Response: # pragma: no cover - # Convert the Starlette request to ASGI parameters - return await handle_sse(request.scope, request.receive, request._send) # type: ignore[reportPrivateUsage] - routes.append( Route( sse_path, - endpoint=sse_endpoint, + endpoint=_RawASGIEndpoint(handle_sse), methods=["GET"], ) ) diff --git a/src/mcp/server/sse.py b/src/mcp/server/sse.py index d71ef25004..e85ab661ef 100644 --- a/src/mcp/server/sse.py +++ b/src/mcp/server/sse.py @@ -7,31 +7,30 @@ # Create an SSE transport at an endpoint sse = SseServerTransport("/messages/") + # Define a raw ASGI handler for the SSE connection + class HandleSSE: + async def __call__(self, scope, receive, send): + async with sse.connect_sse(scope, receive, send) as streams: + await app.run( + streams[0], streams[1], app.create_initialization_options() + ) + # Create Starlette routes for SSE and message handling routes = [ - Route("/sse", endpoint=handle_sse, methods=["GET"]), + Route("/sse", endpoint=HandleSSE(), methods=["GET"]), Mount("/messages/", app=sse.handle_post_message), ] - # Define handler functions - async def handle_sse(request): - async with sse.connect_sse( - request.scope, request.receive, request._send - ) as streams: - await app.run( - streams[0], streams[1], app.create_initialization_options() - ) - # Return empty response to avoid NoneType error - return Response() - # Create and run Starlette app starlette_app = Starlette(routes=routes) uvicorn.run(starlette_app, host="127.0.0.1", port=port) ``` -Note: The handle_sse function must return a Response to avoid a -"TypeError: 'NoneType' object is not callable" error when client disconnects. The example above returns -an empty Response() after the SSE connection ends to fix this. +Note: The SSE handler must be registered as a raw ASGI app (a class instance with +`__call__`, as above), not a plain function. Starlette wraps function endpoints in +`request_response`, which sends a second response after `connect_sse` has already +sent the complete SSE response — crashing middleware such as `BaseHTTPMiddleware` +when the client disconnects. See SseServerTransport class documentation for more details. """ diff --git a/tests/server/mcpserver/test_server.py b/tests/server/mcpserver/test_server.py index c398bc6243..52c0cb60f2 100644 --- a/tests/server/mcpserver/test_server.py +++ b/tests/server/mcpserver/test_server.py @@ -1,5 +1,6 @@ import base64 import logging +from collections.abc import MutableMapping from pathlib import Path from types import SimpleNamespace from typing import Annotated, Any @@ -46,7 +47,12 @@ ) from pydantic import AfterValidator, BaseModel, ValidationError from starlette.applications import Starlette +from starlette.middleware import Middleware +from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint +from starlette.requests import Request +from starlette.responses import Response from starlette.routing import Mount, Route +from starlette.types import Receive, Scope, Send from typing_extensions import NotRequired, TypedDict from mcp.client import Client @@ -74,6 +80,7 @@ from mcp.server.transport_security import TransportSecuritySettings from mcp.shared.exceptions import MCPError from mcp.shared.uri_template import InvalidUriTemplate +from tests.interaction.transports import StreamingASGITransport pytestmark = pytest.mark.anyio @@ -1934,6 +1941,44 @@ async def test_sse_app_applies_the_configured_request_body_limit() -> None: assert response.status_code == 413 +async def test_sse_app_under_base_http_middleware_survives_client_disconnect_without_second_response() -> None: + """A client disconnecting from the SSE stream must not make the app send a second + `http.response.start` — `BaseHTTPMiddleware` crashes on it (issue #883; pins the SDK's + Starlette wiring, not a spec mandate).""" + inner = MCPServer("test").sse_app(host="0.0.0.0") + sent_types: list[str] = [] + + async def recording_inner(scope: Scope, receive: Receive, send: Send) -> None: + async def recording_send(message: MutableMapping[str, Any]) -> None: + sent_types.append(message["type"]) + await send(message) + + await inner(scope, receive, recording_send) + + class PassthroughMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response: + return await call_next(request) + + outer = Starlette(routes=[Mount("/", app=recording_inner)], middleware=[Middleware(PassthroughMiddleware)]) + + with anyio.fail_after(5): + # cancel_on_close=False: closing the client waits for the app's own disconnect handling, + # which is where a second response would be sent. + transport = StreamingASGITransport(outer, cancel_on_close=False) + async with httpx2.AsyncClient(transport=transport, base_url="http://localhost") as http: + async with http.stream("GET", "/sse") as response: + assert response.status_code == 200 + first_event: list[str] = [] + async for line in response.aiter_lines(): + if not line: # blank line terminates the first SSE event + break + first_event.append(line) + assert first_event[0] == "event: endpoint" + # leaving the stream block closes the response: the app sees http.disconnect + + assert sent_types.count("http.response.start") == 1 + + async def test_report_progress_delegates_to_session_report_progress(): """Context.report_progress delegates to ServerSession.report_progress unconditionally.