diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 1a4e9939a4..24a1f097d9 100644 --- a/src/mcp/server/streamable_http.py +++ b/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() diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py index 31f587ee66..725a35e625 100644 --- a/src/mcp/server/streamable_http_manager.py +++ b/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 diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py index 70440d9d03..65c59d8e2c 100644 --- a/tests/server/test_streamable_http_manager.py +++ b/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() diff --git a/tests/server/test_streamable_http_router.py b/tests/server/test_streamable_http_router.py index 07aa063499..a370af6e18 100644 --- a/tests/server/test_streamable_http_router.py +++ b/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