Answer HEAD on the SSE endpoint with GET's headers instead of opening a stream

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,18 @@ 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":
+                return await sse.head_response(request)
+            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 +1240,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 +1254,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, and the stream would never end
+        if request.method == "HEAD":
+            return await sse.head_response(request)
         async with sse.connect_sse(
             request.scope, request.receive, request._send
         ) as streams:
@@ -72,6 +75,9 @@ class SseServerTransport:
         2. handle_post_message() is an ASGI application which receives incoming POST
            requests, which should contain client messages that link to a
            previously-established SSE session.
+
+    head_response() answers a HEAD request to the SSE endpoint, which Starlette
+    routes to the same endpoint as GET.
     """
 
     _endpoint: str
@@ -134,6 +140,27 @@ def __init__(
         self._post_message_app = RequestBodyLimitMiddleware(self._handle_post_message, max_request_body_size)
         logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")
 
+    async def head_response(self, request: Request) -> Response:
+        """Build the response to a HEAD request on the SSE endpoint.
+
+        An SSE stream never ends, so a HEAD served through `connect_sse()` would never complete.
+        This runs the same Host/Origin validation and returns the headers a GET gets, with no body.
+
+        Args:
+            request: The HEAD request.
+
+        Returns:
+            The validation error response, or the SSE headers with no body.
+        """
+        if error_response := await self._security.validate_request(request, is_post=False):
+            return error_response
+        response = Response(
+            media_type="text/event-stream", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}
+        )
+        # A Content-Length would claim an empty stream.
+        del response.headers["content-length"]
+        return response
+
     @asynccontextmanager
     async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
         if scope["type"] != "http":

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 at once with the "
+            "headers a GET would get, no body and no Content-Length, after the same Host/Origin checks as a GET."
+        ),
+        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,35 @@ 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_with_headers_only() -> None:
+    """`MCPServer.sse_app()` answers HEAD at once with GET's headers and no body 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, dict(response.headers), response.content) == snapshot(
+        (
+            200,
+            {
+                "cache-control": "no-store",
+                "x-accel-buffering": "no",
+                "content-type": "text/event-stream; charset=utf-8",
+            },
+            b"",
+        )
+    )
+
+
+@requirement("transport:sse:head")
+async def test_head_on_the_sse_endpoint_is_subject_to_dns_rebinding_protection() -> None:
+    """A HEAD from a Host the transport security settings reject gets the same refusal a GET would."""
+    app = MCPServer("legacy").sse_app(host="127.0.0.1")
+    async with httpx2.AsyncClient(transport=StreamingASGITransport(app), base_url="http://evil.example") as http:
+        with anyio.fail_after(5):
+            response = await http.head("/sse")
+    assert response.status_code == snapshot(421)