Skip to content

Commit 454ba6d

Browse files
committed
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).
1 parent 56af447 commit 454ba6d

3 files changed

Lines changed: 78 additions & 27 deletions

File tree

‎src/mcp/server/mcpserver/server.py‎

Lines changed: 19 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@
5252
from starlette.requests import Request
5353
from starlette.responses import Response
5454
from starlette.routing import Mount, Route
55-
from starlette.types import Receive, Scope, Send
55+
from starlette.types import ASGIApp, Receive, Scope, Send
5656

5757
from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
5858
from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
@@ -138,6 +138,17 @@ class Settings(BaseModel, Generic[LifespanResultT]):
138138
)
139139

140140

141+
class _RawASGIEndpoint:
142+
"""Passes a raw ASGI callable to `Route`, which treats plain functions as request/response
143+
endpoints and would send a second response after the handler already sent one (#883)."""
144+
145+
def __init__(self, app: ASGIApp) -> None:
146+
self._app = app
147+
148+
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
149+
await self._app(scope, receive, send)
150+
151+
141152
def lifespan_wrapper(
142153
app: MCPServer[LifespanResultT],
143154
lifespan: Callable[[MCPServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]],
@@ -1153,14 +1164,13 @@ def sse_app(
11531164
message_path, security_settings=transport_security, max_request_body_size=max_request_body_size
11541165
)
11551166

1156-
async def handle_sse(scope: Scope, receive: Receive, send: Send): # pragma: no cover
1157-
# Add client ID from auth context into request context if available
1158-
1167+
async def handle_sse(scope: Scope, receive: Receive, send: Send) -> None:
1168+
# connect_sse sends the complete SSE response; sending anything after it would
1169+
# deliver a second http.response.start, which crashes BaseHTTPMiddleware (#883).
11591170
async with sse.connect_sse(scope, receive, send) as streams:
11601171
await self._lowlevel_server.run(
11611172
streams[0], streams[1], self._lowlevel_server.create_initialization_options()
11621173
)
1163-
return Response()
11641174

11651175
# Create routes
11661176
routes: list[Route | Mount] = []
@@ -1213,7 +1223,9 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send): # pragma: no
12131223
routes.append(
12141224
Route(
12151225
sse_path,
1216-
endpoint=RequireAuthMiddleware(handle_sse, required_scopes, resource_metadata_url),
1226+
endpoint=RequireAuthMiddleware(
1227+
_RawASGIEndpoint(handle_sse), required_scopes, resource_metadata_url
1228+
),
12171229
methods=["GET"],
12181230
)
12191231
)
@@ -1225,15 +1237,10 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send): # pragma: no
12251237
)
12261238
else:
12271239
# Auth is disabled, no need for RequireAuthMiddleware
1228-
# Since handle_sse is an ASGI app, we need to create a compatible endpoint
1229-
async def sse_endpoint(request: Request) -> Response: # pragma: no cover
1230-
# Convert the Starlette request to ASGI parameters
1231-
return await handle_sse(request.scope, request.receive, request._send) # type: ignore[reportPrivateUsage]
1232-
12331240
routes.append(
12341241
Route(
12351242
sse_path,
1236-
endpoint=sse_endpoint,
1243+
endpoint=_RawASGIEndpoint(handle_sse),
12371244
methods=["GET"],
12381245
)
12391246
)

‎src/mcp/server/sse.py‎

Lines changed: 14 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -7,31 +7,30 @@
77
# Create an SSE transport at an endpoint
88
sse = SseServerTransport("/messages/")
99
10+
# Define a raw ASGI handler for the SSE connection
11+
class HandleSSE:
12+
async def __call__(self, scope, receive, send):
13+
async with sse.connect_sse(scope, receive, send) as streams:
14+
await app.run(
15+
streams[0], streams[1], app.create_initialization_options()
16+
)
17+
1018
# Create Starlette routes for SSE and message handling
1119
routes = [
12-
Route("/sse", endpoint=handle_sse, methods=["GET"]),
20+
Route("/sse", endpoint=HandleSSE(), methods=["GET"]),
1321
Mount("/messages/", app=sse.handle_post_message),
1422
]
1523
16-
# Define handler functions
17-
async def handle_sse(request):
18-
async with sse.connect_sse(
19-
request.scope, request.receive, request._send
20-
) as streams:
21-
await app.run(
22-
streams[0], streams[1], app.create_initialization_options()
23-
)
24-
# Return empty response to avoid NoneType error
25-
return Response()
26-
2724
# Create and run Starlette app
2825
starlette_app = Starlette(routes=routes)
2926
uvicorn.run(starlette_app, host="127.0.0.1", port=port)
3027
```
3128
32-
Note: The handle_sse function must return a Response to avoid a
33-
"TypeError: 'NoneType' object is not callable" error when client disconnects. The example above returns
34-
an empty Response() after the SSE connection ends to fix this.
29+
Note: The SSE handler must be registered as a raw ASGI app (a class instance with
30+
`__call__`, as above), not a plain function. Starlette wraps function endpoints in
31+
`request_response`, which sends a second response after `connect_sse` has already
32+
sent the complete SSE response — crashing middleware such as `BaseHTTPMiddleware`
33+
when the client disconnects.
3534
3635
See SseServerTransport class documentation for more details.
3736
"""

‎tests/server/mcpserver/test_server.py‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import base64
22
import logging
3+
from collections.abc import MutableMapping
34
from pathlib import Path
45
from types import SimpleNamespace
56
from typing import Annotated, Any
@@ -46,7 +47,12 @@
4647
)
4748
from pydantic import AfterValidator, BaseModel, ValidationError
4849
from starlette.applications import Starlette
50+
from starlette.middleware import Middleware
51+
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
52+
from starlette.requests import Request
53+
from starlette.responses import Response
4954
from starlette.routing import Mount, Route
55+
from starlette.types import Receive, Scope, Send
5056
from typing_extensions import NotRequired, TypedDict
5157

5258
from mcp.client import Client
@@ -74,6 +80,7 @@
7480
from mcp.server.transport_security import TransportSecuritySettings
7581
from mcp.shared.exceptions import MCPError
7682
from mcp.shared.uri_template import InvalidUriTemplate
83+
from tests.interaction.transports import StreamingASGITransport
7784

7885
pytestmark = pytest.mark.anyio
7986

@@ -1934,6 +1941,44 @@ async def test_sse_app_applies_the_configured_request_body_limit() -> None:
19341941
assert response.status_code == 413
19351942

19361943

1944+
async def test_sse_app_under_base_http_middleware_survives_client_disconnect_without_second_response() -> None:
1945+
"""A client disconnecting from the SSE stream must not make the app send a second
1946+
`http.response.start` — `BaseHTTPMiddleware` crashes on it (issue #883; pins the SDK's
1947+
Starlette wiring, not a spec mandate)."""
1948+
inner = MCPServer("test").sse_app(host="0.0.0.0")
1949+
sent_types: list[str] = []
1950+
1951+
async def recording_inner(scope: Scope, receive: Receive, send: Send) -> None:
1952+
async def recording_send(message: MutableMapping[str, Any]) -> None:
1953+
sent_types.append(message["type"])
1954+
await send(message)
1955+
1956+
await inner(scope, receive, recording_send)
1957+
1958+
class PassthroughMiddleware(BaseHTTPMiddleware):
1959+
async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
1960+
return await call_next(request)
1961+
1962+
outer = Starlette(routes=[Mount("/", app=recording_inner)], middleware=[Middleware(PassthroughMiddleware)])
1963+
1964+
with anyio.fail_after(5):
1965+
# cancel_on_close=False: closing the client waits for the app's own disconnect handling,
1966+
# which is where a second response would be sent.
1967+
transport = StreamingASGITransport(outer, cancel_on_close=False)
1968+
async with httpx2.AsyncClient(transport=transport, base_url="http://localhost") as http:
1969+
async with http.stream("GET", "/sse") as response:
1970+
assert response.status_code == 200
1971+
first_event: list[str] = []
1972+
async for line in response.aiter_lines():
1973+
if not line: # blank line terminates the first SSE event
1974+
break
1975+
first_event.append(line)
1976+
assert first_event[0] == "event: endpoint"
1977+
# leaving the stream block closes the response: the app sees http.disconnect
1978+
1979+
assert sent_types.count("http.response.start") == 1
1980+
1981+
19371982
async def test_report_progress_delegates_to_session_report_progress():
19381983
"""Context.report_progress delegates to ServerSession.report_progress unconditionally.
19391984

0 commit comments

Comments
 (0)