fix(streamable-http): reject GET/DELETE without session-id before allocating

Before this patch, any request to the stateful streamable_http_manager
without an MCP-Session-Id header entered the "new session" branch and:

  1. allocated a StreamableHTTPServerTransport
  2. registered it in `_server_instances`
  3. spawned a background `run_server` task waiting on `serve_loop` /
     `app.run()` for messages that never come

The subsequent transport-level rejection (400 "Missing session ID" for
GET/DELETE, 406 "Not Acceptable" for GET with missing Accept header)
returned the correct HTTP response but did not tear the allocated
session or its task down. The `finally` cleanup in the background task
only fires when the loop completes, and without the (opt-in,
off-by-default) `session_idle_timeout` the task blocks forever.

Under a real deployment we hit this via a Docker healthcheck polling
`GET /mcp` every 30s on a FastMCP-based server: ~2 sessions/min leaked,
~5.3 MiB/day RAM growth, 28 800 accumulated sessions with 0 teardown
events over 10 days.

The fix reorders `_handle_stateful_request` so that:

  * security validation runs first (preserves the 421 DNS-rebinding
    behaviour tested in test_streamable_http_security_get_request),
  * GET and DELETE with no session-id return 400 "Missing session ID"
    at the manager layer without touching `_server_instances` and
    without spawning any task,
  * POST without session-id continues to initialize a new session
    exactly as before,
  * PUT/PATCH/OPTIONS with no session-id continue to reach the
    transport and get the existing 405 "Method Not Allowed".

A regression test asserts both counters (`_server_instances`,
`_task_group._tasks`) stay at zero after 300 bad requests.

Author: raphaelOhana

src/mcp/server/streamable_http_manager.py

@@ -21,7 +21,7 @@
     EventStore,
     StreamableHTTPServerTransport,
 )
-from mcp.server.transport_security import TransportSecuritySettings
+from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings
 from mcp.types import INVALID_REQUEST, ErrorData, JSONRPCError
 
 logger = logging.getLogger(__name__)
@@ -85,6 +85,15 @@ def __init__(
         self.retry_interval = retry_interval
         self.session_idle_timeout = session_idle_timeout
 
+        # Pre-check middleware: the manager runs security validation *before*
+        # deciding whether to allocate a session. Without this, a request that
+        # fails security (DNS rebinding, bad Host) would either be rejected too
+        # late (after a session has already been created and would leak) or with
+        # the wrong status code (400 "Missing session ID" instead of 421
+        # "Invalid Host header"). Using the same middleware the transport uses
+        # keeps the two validation paths consistent.
+        self._security = TransportSecurityMiddleware(security_settings)
+
         # Session tracking (only used if not stateless)
         self._session_creation_lock = anyio.Lock()
         self._server_instances: dict[str, StreamableHTTPServerTransport] = {}
@@ -229,6 +238,16 @@ async def _handle_stateful_request(
             send: ASGI send function
         """
         request = Request(scope, receive)
+
+        # Run security validation FIRST so DNS-rebinding / bad-Host requests
+        # are rejected with 421 (or the transport-level Content-Type 400)
+        # regardless of whether a session existed or not — and so bad-Host
+        # requests can never trigger a session allocation.
+        security_error = await self._security.validate_request(request, is_post=(request.method == "POST"))
+        if security_error is not None:
+            await security_error(scope, receive, send)
+            return
+
         request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)
 
         user = scope.get("user")
@@ -262,6 +281,36 @@ async def _handle_stateful_request(
             return
 
         if request_mcp_session_id is None:
+            # Only POST may initialize a new session (per MCP spec: a session is
+            # opened by the response to the ``initialize`` JSON-RPC request,
+            # which is always a POST). GET and DELETE without a session-id are
+            # protocol errors and must be rejected here — the transport-layer
+            # rejection happens too late to matter: by the time
+            # `StreamableHTTPServerTransport.handle_request()` sees the request,
+            # the session has already been allocated in `_server_instances`
+            # AND a background `run_server` task has been spawned, waiting on a
+            # stream that will never receive anything. Both leak forever unless
+            # `session_idle_timeout` is set (which is opt-in and off by default).
+            # Other unsupported methods (PUT/PATCH/OPTIONS/...) are intentionally
+            # let through to the existing ``_handle_unsupported_request`` path,
+            # which returns 405 — preserved for API stability.
+            if request.method in ("GET", "DELETE"):
+                error_response = JSONRPCError(
+                    jsonrpc="2.0",
+                    id="server-error",
+                    error=ErrorData(
+                        code=INVALID_REQUEST,
+                        message="Bad Request: Missing session ID",
+                    ),
+                )
+                response = Response(
+                    content=error_response.model_dump_json(by_alias=True, exclude_none=True),
+                    status_code=400,
+                    media_type="application/json",
+                )
+                await response(scope, receive, send)
+                return
+
             # New session case
             logger.debug("Creating new transport")
             async with self._session_creation_lock:

tests/server/test_streamable_http_manager.py

@@ -382,6 +382,77 @@ async def capture_send(message: Message):
         assert response_start["status"] == 404
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("method", ["GET", "DELETE"])
+async def test_non_post_without_session_id_does_not_allocate_session(method: str):
+    """Regression test: GET/DELETE without a session-id must return 400 without
+    allocating a session or spawning a background task.
+
+    Before this fix, any request without a session id — including GET/DELETE —
+    entered the "new session" branch of the manager, created a transport,
+    registered it in ``_server_instances``, and launched a ``run_server`` task
+    that would wait forever for messages that never come. The transport-level
+    validation then rejected the request (with 400 or 406), but the allocated
+    session + task were leaked. Under a Docker healthcheck polling /mcp every
+    30 seconds this accumulated ~2 sessions/min indefinitely (~1 GiB/week).
+    """
+    app = Server("test-non-post-no-session")
+    manager = StreamableHTTPSessionManager(app=app)
+
+    async with manager.run():
+        sent_messages: list[Message] = []
+        response_body = b""
+
+        async def mock_send(message: Message):
+            nonlocal response_body
+            sent_messages.append(message)
+            if message["type"] == "http.response.body":
+                response_body += message.get("body", b"")
+
+        scope: Scope = {
+            "type": "http",
+            "method": method,
+            "path": "/mcp",
+            "headers": [
+                (b"accept", b"application/json, text/event-stream"),
+            ],
+        }
+
+        async def mock_receive():  # pragma: no cover
+            return {"type": "http.request", "body": b"", "more_body": False}
+
+        # Snapshot before
+        assert len(manager._server_instances) == 0
+
+        await manager.handle_request(scope, mock_receive, mock_send)
+
+        # Give any accidentally-spawned background task a chance to register
+        await anyio.sleep(0.05)
+
+        # No session, no task
+        assert len(manager._server_instances) == 0, (
+            f"{method} without session-id must not allocate a session — leaked {len(manager._server_instances)}"
+        )
+        assert manager._task_group is not None
+        # anyio TaskGroup internals: no live tasks belonging to run_server
+        assert len(manager._task_group._tasks) == 0, (
+            f"{method} without session-id must not spawn a background task — leaked {len(manager._task_group._tasks)}"
+        )
+
+        # Response should be a well-formed JSON-RPC error at status 400
+        response_start = next(
+            (msg for msg in sent_messages if msg["type"] == "http.response.start"),
+            None,
+        )
+        assert response_start is not None, "Should have sent a response"
+        assert response_start["status"] == 400
+
+        error_data = json.loads(response_body)
+        assert error_data["jsonrpc"] == "2.0"
+        assert error_data["error"]["code"] == INVALID_REQUEST
+        assert "Missing session ID" in error_data["error"]["message"]
+
+
 def test_session_idle_timeout_rejects_non_positive():
     with pytest.raises(ValueError, match="positive number"):
         StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=-1)