Answer HEAD on the SSE endpoint with 405 instead of opening a stream

Author: Kludex

src/mcp/server/mcpserver/server.py

@@ -51,8 +51,7 @@
 from starlette.middleware.authentication import AuthenticationMiddleware
 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.routing import Mount, Route, request_response
 
 from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
 from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
@@ -1172,14 +1171,19 @@ 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 with sse.connect_sse(scope, receive, send) as streams:
+        async def sse_endpoint(request: Request) -> Response:
+            if request.method == "HEAD":
+                # The SSE stream never ends, so a HEAD answered with it would never complete.
+                return Response(status_code=405, headers={"Allow": "GET"})
+            async with sse.connect_sse(  # pragma: no cover
+                request.scope,
+                request.receive,
+                request._send,  # type: ignore[reportPrivateUsage]
+            ) as streams:
                 await self._lowlevel_server.run(
                     streams[0], streams[1], self._lowlevel_server.create_initialization_options()
                 )
-            return Response()
+            return Response()  # pragma: no cover
 
         # Create routes
         routes: list[Route | Mount] = []
@@ -1237,7 +1241,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(
+                        request_response(sse_endpoint), required_scopes, resource_metadata_url
+                    ),
                     methods=["GET"],
                 )
             )
@@ -1249,11 +1255,6 @@ 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,

src/mcp/server/sse.py

@@ -15,6 +15,9 @@
 
     # Define handler functions
     async def handle_sse(request):
+        # Starlette also routes HEAD here; the stream never ends, so a HEAD would never complete
+        if request.method == "HEAD":
+            return Response(status_code=405, headers={"Allow": "GET"})
         async with sse.connect_sse(
             request.scope, request.receive, request._send
         ) as streams:

tests/interaction/_requirements.py

@@ -2705,6 +2705,15 @@ def __post_init__(self) -> None:
         transports=("sse",),
         note="Only observable over the legacy SSE transport.",
     ),
+    "transport:sse:head": Requirement(
+        source="issue:#3736",
+        behavior=(
+            "A HEAD request to the SSE endpoint of `MCPServer.sse_app()` is answered 405 with `Allow: GET` "
+            "without opening a session or a stream, so the connection stays usable."
+        ),
+        transports=("sse",),
+        note="Only observable over the legacy SSE transport.",
+    ),
     "transport:stdio": Requirement(
         source=f"{SPEC_BASE_URL}/basic/transports#stdio",
         behavior=(

tests/interaction/transports/test_sse.py

@@ -18,6 +18,8 @@
 from mcp.client.client import Client
 from mcp.client.sse import sse_client
 from mcp.server import Server
+from mcp.server.mcpserver import MCPServer
+from mcp.server.transport_security import TransportSecuritySettings
 from tests.interaction._connect import BASE_URL, build_sse_app
 from tests.interaction._requirements import requirement
 from tests.interaction.transports._bridge import StreamingASGITransport
@@ -88,3 +90,15 @@ async def test_post_for_an_unknown_session_is_rejected() -> None:
             "/messages/", params={"session_id": uuid4().hex}, json={"jsonrpc": "2.0", "method": "ping", "id": 1}
         )
     assert (response.status_code, response.text) == snapshot((404, "Could not find session"))
+
+
+@requirement("transport:sse:head")
+async def test_head_on_the_sse_endpoint_is_answered_without_opening_a_stream() -> None:
+    """`MCPServer.sse_app()` answers HEAD at once instead of streaming forever."""
+    app = MCPServer("legacy").sse_app(
+        transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False)
+    )
+    async with httpx2.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http:
+        with anyio.fail_after(5):
+            response = await http.head("/sse")
+    assert (response.status_code, response.headers["allow"]) == snapshot((405, "GET"))