fix(server): discard a session whose establishing request was refused

src/mcp/server/streamable_http.py

@@ -793,7 +793,7 @@ async def _handle_delete_request(self, request: Request, send: Send) -> None:
             await response(request.scope, request.receive, send)
             return
 
-        if not await self._validate_request_headers(request, send):  # pragma: no cover
+        if not await self._validate_request_headers(request, send):
             return
 
         await self.terminate()

src/mcp/server/streamable_http_manager.py

@@ -355,8 +355,30 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                 # Start the server task
                 await self._task_group.start(run_server)
 
+                # The session above is provisional: every validation (Host, Accept,
+                # Content-Type, JSON parse, JSON-RPC shape, "Missing session ID")
+                # lives inside `handle_request`, so it runs only now. If the request
+                # that was meant to establish this session is refused, the session
+                # must not survive it -- otherwise a rejected request leaves live
+                # state behind and hands the caller a usable session id.
+                establishing_status: int | None = None
+
+                async def send_tracking_status(message: Message) -> None:
+                    nonlocal establishing_status
+                    if message["type"] == "http.response.start":
+                        establishing_status = message["status"]
+                    await send(message)
+
                 # Handle the HTTP request and return the response
-                await http_transport.handle_request(scope, receive, send)
+                await http_transport.handle_request(scope, receive, send_tracking_status)
+
+                if establishing_status is not None and establishing_status >= 400:
+                    logger.debug(
+                        f"Discarding session {new_session_id}: establishing request returned {establishing_status}"
+                    )
+                    self._server_instances.pop(new_session_id, None)
+                    self._session_owners.pop(new_session_id, None)
+                    await http_transport.terminate()
         else:
             # Unknown or expired session ID - return 404 per MCP spec
             # TODO(L62): Align error code once spec clarifies

tests/server/test_streamable_http_manager.py

@@ -3,13 +3,14 @@
 import json
 import logging
 from collections.abc import Iterator
-from typing import Any
+from typing import Any, Final
 from unittest.mock import AsyncMock, patch
 
 import anyio
 import httpx2
 import pytest
 from mcp_types import INVALID_REQUEST, ListToolsResult, PaginatedRequestParams
+from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS
 from starlette.types import Message, Receive, Scope, Send
 
 from mcp import Client
@@ -24,6 +25,19 @@
     StreamableHTTPSessionManager,
 )
 
+_INITIALIZE_BODY: Final[bytes] = json.dumps(
+    {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": "initialize",
+        "params": {
+            "protocolVersion": HANDSHAKE_PROTOCOL_VERSIONS[-1],
+            "capabilities": {},
+            "clientInfo": {"name": "test-client", "version": "1.0"},
+        },
+    }
+).encode()
+
 
 @pytest.mark.anyio
 async def test_run_can_only_be_called_once():
@@ -146,6 +160,67 @@ async def send(message: Message) -> None:
     assert response_start["status"] == 413
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("method", "headers", "body", "expected_status"),
+    [
+        pytest.param(
+            "POST",
+            [(b"content-type", b"application/json"), (b"accept", b"application/json, text/event-stream")],
+            json.dumps({"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}).encode(),
+            400,
+            id="post-that-is-not-initialize",
+        ),
+        pytest.param(
+            "POST",
+            [(b"content-type", b"application/json"), (b"accept", b"application/json, text/event-stream")],
+            b"{not valid json",
+            400,
+            id="post-with-malformed-body",
+        ),
+        pytest.param(
+            "POST",
+            [(b"content-type", b"application/json"), (b"accept", b"text/plain")],
+            _INITIALIZE_BODY,
+            406,
+            id="post-with-unacceptable-accept",
+        ),
+        pytest.param("GET", [(b"accept", b"text/event-stream")], b"", 400, id="get-without-session"),
+        pytest.param("DELETE", [], b"", 400, id="delete-without-session"),
+    ],
+)
+async def test_refused_request_leaves_no_session_behind(
+    method: str, headers: list[tuple[bytes, bytes]], body: bytes, expected_status: int
+) -> None:
+    """SDK-defined: a request the transport refuses must not leave a registered session behind.
+
+    The session is minted before any validation runs -- Host, Accept, Content-Type, JSON parse,
+    JSON-RPC shape and the "Missing session ID" check all live downstream in the transport -- so a
+    refusal has to undo it. Otherwise a rejected request grows `_server_instances` forever and hands
+    the caller a session id that later requests can still use.
+
+    This is the same property the suite already asserts by name for the 413 path in
+    `test_oversized_content_length_is_rejected_before_body_read_or_session_creation`.
+    """
+    manager = StreamableHTTPSessionManager(app=Server("test-refused-request"))
+    sent_messages: list[Message] = []
+
+    async def mock_send(message: Message) -> None:
+        sent_messages.append(message)
+
+    async def mock_receive() -> Message:
+        return {"type": "http.request", "body": body, "more_body": False}
+
+    scope: Scope = {"type": "http", "method": method, "path": "/mcp", "headers": headers}
+
+    async with manager.run():
+        await manager.handle_request(scope, mock_receive, mock_send)
+
+        response_start = next(msg for msg in sent_messages if msg["type"] == "http.response.start")
+        assert response_start["status"] == expected_status
+        assert manager._server_instances == {}
+
+
 @pytest.mark.anyio
 async def test_client_disconnect_while_streaming_request_body_is_replayed() -> None:
     """SDK-defined: raw ASGI is required to prove a disconnect before body completion reaches the transport."""
@@ -513,35 +588,13 @@ async def test_idle_session_is_reaped(caplog: pytest.LogCaptureFixture, request:
     caplog.set_level(logging.INFO, logger=streamable_http_manager.__name__)
 
     async with manager.run():
-        sent_messages: list[Message] = []
-
-        async def mock_send(message: Message):
-            sent_messages.append(message)
+        # Establish the session with a real `initialize`: a request the transport refuses no
+        # longer leaves a session behind, so there would be nothing for the reaper to reap.
+        session_id = await _open_session(manager, None)
 
-        scope = {
-            "type": "http",
-            "method": "POST",
-            "path": "/mcp",
-            "headers": [(b"content-type", b"application/json")],
-        }
-
-        async def mock_receive():
+        async def mock_receive() -> Message:
             return {"type": "http.request", "body": b"", "more_body": False}
 
-        await manager.handle_request(scope, mock_receive, mock_send)
-
-        session_id = None
-        for msg in sent_messages:  # pragma: no branch
-            if msg["type"] == "http.response.start":  # pragma: no branch
-                for header_name, header_value in msg.get("headers", []):  # pragma: no branch
-                    if header_name.decode().lower() == MCP_SESSION_ID_HEADER.lower():
-                        session_id = header_value.decode()
-                        break
-                if session_id:  # pragma: no branch
-                    break
-
-        assert session_id is not None, "Session ID not found in response headers"
-
         # Wait for the 50ms idle timeout to fire and the session to be unregistered. Re-requesting
         # the session to poll for the 404 would push its idle deadline forward and keep it alive.
         with anyio.fail_after(5):
@@ -613,18 +666,31 @@ def _request_scope(
 
 
 async def _open_session(manager: StreamableHTTPSessionManager, user: AuthenticatedUser | None) -> str:
-    """Create a new session as `user` and return its session ID."""
+    """Create a new session as `user` and return its session ID.
+
+    Establishes the session the way a real client does, with an `initialize` request, because
+    a request the transport refuses no longer leaves a session behind. The reply is an SSE
+    stream, so the body is followed by a disconnect: that ends the stream and lets
+    `handle_request` return, while the session itself lives on in the manager's task group.
+    """
     sent_messages: list[Message] = []
+    body_sent = False
 
     async def mock_send(message: Message) -> None:
         sent_messages.append(message)
 
     async def mock_receive() -> Message:
-        return {"type": "http.request", "body": b"", "more_body": False}
+        nonlocal body_sent
+        if body_sent:
+            return {"type": "http.disconnect"}
+        body_sent = True
+        return {"type": "http.request", "body": _INITIALIZE_BODY, "more_body": False}
 
-    await manager.handle_request(_request_scope(user=user), mock_receive, mock_send)
+    with anyio.fail_after(5):
+        await manager.handle_request(_request_scope(user=user), mock_receive, mock_send)
 
     response_start = next(msg for msg in sent_messages if msg["type"] == "http.response.start")
+    assert response_start["status"] == 200, f"initialize was refused with {response_start['status']}"
     headers = dict(response_start.get("headers", []))
     return headers[MCP_SESSION_ID_HEADER.encode()].decode()
 

tests/server/test_streamable_http_router.py

@@ -1,5 +1,7 @@
 """Regression coverage for the StreamableHTTP per-session response router."""
 
+import logging
+
 import anyio
 import pytest
 from mcp_types import JSONRPCMessage, JSONRPCResponse
@@ -141,3 +143,23 @@ async def test_json_post_answers_500_when_session_terminates_mid_request() -> No
 
     assert post.sent[0]["type"] == "http.response.start"
     assert post.sent[0]["status"] == 500
+
+
+@pytest.mark.anyio
+async def test_router_reports_a_stream_closure_it_did_not_cause(caplog: pytest.LogCaptureFixture) -> None:
+    """A stream closed without terminating the transport is an anomaly, not a clean shutdown.
+
+    `terminate()` closes these streams deliberately and the router says so at debug level.
+    Anything else closing them is unexplained, so the router must surface it at exception
+    level instead of swallowing it as a normal end-of-stream.
+    """
+    transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=True)
+    caplog.set_level(logging.ERROR, logger="mcp.server.streamable_http")
+
+    async with transport.connect():
+        assert not transport.is_terminated
+        # Close the stream the router is iterating without going through terminate().
+        assert transport._write_stream_reader is not None
+        await transport._write_stream_reader.aclose()
+
+    assert "Unexpected closure of read stream in message router" in caplog.text