Skip to content

Commit 58bff01

Browse files
committed
Serve both SSE routes through one request/response endpoint
1 parent 43f85b3 commit 58bff01

1 file changed

Lines changed: 21 additions & 27 deletions

File tree

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

Lines changed: 21 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -51,8 +51,7 @@
5151
from starlette.middleware.authentication import AuthenticationMiddleware
5252
from starlette.requests import Request
5353
from starlette.responses import Response
54-
from starlette.routing import Mount, Route
55-
from starlette.types import ASGIApp, Receive, Scope, Send
54+
from starlette.routing import Mount, Route, request_response
5655

5756
from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
5857
from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
@@ -1178,21 +1177,25 @@ def sse_app(
11781177

11791178
security = TransportSecurityMiddleware(transport_security)
11801179

1181-
async def handle_sse(scope: Scope, receive: Receive, send: Send) -> None:
1182-
if scope["method"] == "HEAD":
1183-
response = await security.validate_request(Request(scope, receive))
1184-
if response is None:
1185-
# GET's headers without its never-ending stream; a Content-Length would claim an empty one.
1186-
response = Response(
1187-
media_type="text/event-stream", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}
1188-
)
1189-
del response.headers["content-length"]
1190-
await response(scope, receive, send)
1191-
return
1192-
async with sse.connect_sse(scope, receive, send) as streams: # pragma: no cover
1180+
async def sse_endpoint(request: Request) -> Response:
1181+
if request.method == "HEAD":
1182+
if error_response := await security.validate_request(request):
1183+
return error_response
1184+
# GET's headers without its never-ending stream; a Content-Length would claim an empty one.
1185+
response = Response(
1186+
media_type="text/event-stream", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}
1187+
)
1188+
del response.headers["content-length"]
1189+
return response
1190+
async with sse.connect_sse( # pragma: no cover
1191+
request.scope,
1192+
request.receive,
1193+
request._send, # type: ignore[reportPrivateUsage]
1194+
) as streams:
11931195
await self._lowlevel_server.run(
11941196
streams[0], streams[1], self._lowlevel_server.create_initialization_options()
11951197
)
1198+
return Response() # pragma: no cover
11961199

11971200
# Create routes
11981201
routes: list[Route | Mount] = []
@@ -1250,7 +1253,9 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send) -> None:
12501253
routes.append(
12511254
Route(
12521255
sse_path,
1253-
endpoint=RequireAuthMiddleware(handle_sse, required_scopes, resource_metadata_url),
1256+
endpoint=RequireAuthMiddleware(
1257+
request_response(sse_endpoint), required_scopes, resource_metadata_url
1258+
),
12541259
methods=["GET"],
12551260
)
12561261
)
@@ -1265,7 +1270,7 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send) -> None:
12651270
routes.append(
12661271
Route(
12671272
sse_path,
1268-
endpoint=_ASGIEndpoint(handle_sse),
1273+
endpoint=sse_endpoint,
12691274
methods=["GET"],
12701275
)
12711276
)
@@ -1429,14 +1434,3 @@ def require_client_extension(ctx: ServerRequestContext[Any, Any], identifier: st
14291434
message=f"Client did not declare required extension {identifier!r}",
14301435
data=data.model_dump(by_alias=True, mode="json", exclude_none=True),
14311436
)
1432-
1433-
1434-
class _ASGIEndpoint:
1435-
"""Hands a `Route` a raw ASGI app; a plain function would be wrapped as a request/response endpoint,
1436-
which sends a second response after the SSE stream has already completed."""
1437-
1438-
def __init__(self, app: ASGIApp) -> None:
1439-
self.app = app
1440-
1441-
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
1442-
await self.app(scope, receive, send)

0 commit comments

Comments
 (0)