Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 19 additions & 12 deletions src/mcp/server/mcpserver/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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]],
Expand Down Expand Up @@ -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] = []
Expand Down Expand Up @@ -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"],
)
)
Expand All @@ -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"],
)
)
Expand Down
29 changes: 14 additions & 15 deletions src/mcp/server/sse.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
"""
Expand Down
45 changes: 45 additions & 0 deletions tests/server/mcpserver/test_server.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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.

Expand Down
Loading