Add SseServerTransport.head_response so the transport answers HEAD

Author: Kludex

src/mcp/server/mcpserver/server.py

@@ -98,11 +98,7 @@
     StreamableHTTPSessionManager,
 )
 from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, SubscriptionBus
-from mcp.server.transport_security import (
-    DEFAULT_MAX_REQUEST_BODY_SIZE,
-    TransportSecurityMiddleware,
-    TransportSecuritySettings,
-)
+from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
 from mcp.shared.exceptions import MCPError
 from mcp.shared.uri_template import UriTemplate
 
@@ -1175,18 +1171,9 @@ def sse_app(
             message_path, security_settings=transport_security, max_request_body_size=max_request_body_size
         )
 
-        security = TransportSecurityMiddleware(transport_security)
-
         async def sse_endpoint(request: Request) -> Response:
             if request.method == "HEAD":
-                if error_response := await security.validate_request(request):
-                    return error_response
-                # GET's headers without its never-ending stream; a Content-Length would claim an empty one.
-                response = Response(
-                    media_type="text/event-stream", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}
-                )
-                del response.headers["content-length"]
-                return response
+                return await sse.head_response(request)
             async with sse.connect_sse(  # pragma: no cover
                 request.scope,
                 request.receive,

src/mcp/server/sse.py

@@ -15,11 +15,9 @@
 
     # Define handler functions
     async def handle_sse(request):
-        # Starlette also routes HEAD here: answer with GET's headers, not its never-ending stream
+        # Starlette also routes HEAD here, and the stream would never end
         if request.method == "HEAD":
-            response = Response(media_type="text/event-stream", headers={"Cache-Control": "no-store"})
-            del response.headers["content-length"]
-            return response
+            return await sse.head_response(request)
         async with sse.connect_sse(
             request.scope, request.receive, request._send
         ) as streams:
@@ -77,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
@@ -139,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":