fix(server): type-preserve streamable HTTP request stream keys

Route per-request streams with i:/s:-prefixed keys so integer and
string JSON-RPC ids stay distinct, and so a string id cannot share
the standalone GET slot. Mention notifications/cancelled in the 409
conflict message. Cover both cases with regression tests.

Author: arimu1

src/mcp/server/streamable_http.py

@@ -57,9 +57,32 @@
 CONTENT_TYPE_JSON = "application/json"
 CONTENT_TYPE_SSE = "text/event-stream"
 
-# Special key for the standalone GET stream
+# Type aliases
+StreamId = str
+EventId = str
+# An SSE event-dict as accepted by sse-starlette (`event`, `data`, `id`, `retry`).
+SSEEvent = dict[str, Any]
+
+# Special key for the standalone GET stream. Request routing keys never equal
+# this value (see `_request_stream_key`), so a JSON-RPC string id of
+# `"_GET_stream"` cannot lock out or share the standalone GET slot.
 GET_STREAM_KEY = "_GET_stream"
 
+
+def _request_stream_key(request_id: RequestId) -> StreamId:
+    """Type-preserving routing key for per-request streams and the event store.
+
+    JSON-RPC treats integer ``1`` and string ``\"1\"`` as distinct ids. Using
+    ``str(id)`` collapses them (so an in-flight numeric id falsely 409s a
+    string id, or cross-wires responses) and also lets a string id equal to
+    ``GET_STREAM_KEY`` share the standalone GET slot.
+    """
+    # RequestId is int | str; exclude bool (a subclass of int) defensively.
+    if type(request_id) is int:
+        return f"i:{request_id}"
+    return f"s:{request_id}"
+
+
 # Buffer for the per-request `_request_streams` so the serial `message_router`
 # can deposit a response and move on instead of head-of-line blocking the
 # whole session on a lazily-started `sse_writer`. See #1764.
@@ -76,12 +99,6 @@
 # Pattern ensures entire string contains only valid characters by using ^ and $ anchors
 SESSION_ID_PATTERN = re.compile(r"^[\x21-\x7E]+$")
 
-# Type aliases
-StreamId = str
-EventId = str
-# An SSE event-dict as accepted by sse-starlette (`event`, `data`, `id`, `retry`).
-SSEEvent = dict[str, Any]
-
 
 def check_accept_headers(request: Request) -> tuple[bool, bool]:
     """Return (has_json, has_sse) for the request's Accept header, with RFC 7231 wildcard handling.
@@ -598,16 +615,22 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 else request.headers.get(MCP_PROTOCOL_VERSION_HEADER, DEFAULT_NEGOTIATED_VERSION)
             )
 
-            request_id = str(message.id)
-
-            # `_request_streams` is keyed by the bare JSON-RPC id. A second concurrent
-            # POST reusing an in-flight id would overwrite that slot and cross-wire
-            # responses (one caller receives the other's payload; the other hangs).
-            # Reject the collision with 409 rather than silently re-routing. Ids may
-            # still be reused after the earlier request has completed.
+            # Type-preserving routing key so int 1 and str "1" stay distinct, and so
+            # a string id never collides with GET_STREAM_KEY. The membership test and
+            # the assignment below must not have an await between them: with
+            # resumability on, EventStore.store_event is user code and any await
+            # re-opens a check-then-act window (#3137 / #3163).
+            request_id = _request_stream_key(message.id)
+
+            # A second concurrent POST reusing an in-flight id would overwrite that
+            # slot and cross-wire responses (one caller receives the other's payload;
+            # the other hangs). Reject with 409 rather than silently re-routing. Ids
+            # may still be reused after the earlier request has completed; clients
+            # that retry after a timeout should send notifications/cancelled first.
             if request_id in self._request_streams:
                 response = self._create_error_response(
-                    f"Conflict: Request id {request_id} is already in flight for this session",
+                    f"Conflict: Request id {message.id!r} is already in flight for this "
+                    f"session; send notifications/cancelled before reusing the id",
                     HTTPStatus.CONFLICT,
                 )
                 await response(scope, receive, send)
@@ -1039,7 +1062,7 @@ async def message_router():
                         # Null-id errors (e.g., parse errors) fall through to
                         # the GET stream since they can't be correlated.
                         if isinstance(message, JSONRPCResponse | JSONRPCError) and message.id is not None:
-                            target_request_id = str(message.id)
+                            target_request_id = _request_stream_key(message.id)
                         # Extract related_request_id from meta if it exists
                         elif (
                             session_message.metadata is not None
@@ -1056,7 +1079,7 @@ async def message_router():
                                 # storing or queueing rather than park it (#1764).
                                 logger.debug(f"Dropped message related to request {related_request_id} in JSON mode")
                                 continue
-                            target_request_id = str(related_request_id)
+                            target_request_id = _request_stream_key(related_request_id)
 
                         request_stream_id = target_request_id if target_request_id is not None else GET_STREAM_KEY
 

tests/shared/test_streamable_http.py

@@ -1011,6 +1011,155 @@ async def test_request_id_reuse_after_completion_allowed(basic_app: Starlette) -
             assert body["result"]["content"][0]["text"] == "Called test_tool"
 
 
+@pytest.mark.anyio
+async def test_numeric_and_string_request_ids_do_not_collide_while_in_flight(basic_app: Starlette) -> None:
+    """Integer id 1 and string id "1" are distinct JSON-RPC ids and may run concurrently.
+
+    Routing used to key streams with str(id), so an in-flight numeric 1 falsely
+    409'd a string "1" (or cross-wired responses). Raw HTTP is required to control
+    the JSON-RPC id types on the wire.
+    """
+    async with make_client(basic_app) as client:
+        init_response = await client.post(
+            "/mcp",
+            headers={
+                "Accept": "application/json, text/event-stream",
+                "Content-Type": "application/json",
+            },
+            json=INIT_REQUEST,
+        )
+        assert init_response.status_code == 200
+        headers = {
+            "Accept": "application/json, text/event-stream",
+            "Content-Type": "application/json",
+            MCP_SESSION_ID_HEADER: init_response.headers[MCP_SESSION_ID_HEADER],
+            MCP_PROTOCOL_VERSION_HEADER: extract_protocol_version_from_sse(init_response),
+        }
+
+        async with client.stream(
+            "POST",
+            "/mcp",
+            headers=headers,
+            json={
+                "jsonrpc": "2.0",
+                "method": "tools/call",
+                "params": {"name": "wait_for_lock_with_notification", "arguments": {}},
+                "id": 1,
+            },
+        ) as first_response:
+            assert first_response.status_code == 200
+            lines = first_response.aiter_lines()
+            with anyio.fail_after(5):
+                notification = await next_sse_data(lines)
+            assert notification["params"]["data"] == "First notification before lock"
+
+            string_id_response = await client.post(
+                "/mcp",
+                headers=headers,
+                json={
+                    "jsonrpc": "2.0",
+                    "method": "tools/call",
+                    "params": {"name": "test_tool", "arguments": {}},
+                    "id": "1",
+                },
+            )
+            assert string_id_response.status_code == 200
+            string_body = first_sse_data(string_id_response)
+            assert string_body["id"] == "1"
+            assert string_body["result"]["content"][0]["text"] == "Called test_tool"
+
+            release_response = await client.post(
+                "/mcp",
+                headers=headers,
+                json={
+                    "jsonrpc": "2.0",
+                    "method": "tools/call",
+                    "params": {"name": "release_lock", "arguments": {}},
+                    "id": "release-lock-type-key",
+                },
+            )
+            assert release_response.status_code == 200
+
+            with anyio.fail_after(5):
+                second_notification = await next_sse_data(lines)
+                final = await next_sse_data(lines)
+            assert second_notification["params"]["data"] == "Second notification after lock"
+            assert final["id"] == 1
+            assert final["result"]["content"][0]["text"] == "Completed"
+
+
+@pytest.mark.anyio
+async def test_request_id_matching_get_stream_key_does_not_block_standalone_get(basic_app: Starlette) -> None:
+    """A POST with string id \"_GET_stream\" must not occupy the standalone GET slot.
+
+    Pre-fix, both used the bare string as the routing key, so a client could lock
+    itself out of GET (or have GET overwrite its POST slot).
+    """
+    async with make_client(basic_app) as client:
+        init_response = await client.post(
+            "/mcp",
+            headers={
+                "Accept": "application/json, text/event-stream",
+                "Content-Type": "application/json",
+            },
+            json=INIT_REQUEST,
+        )
+        assert init_response.status_code == 200
+        session_id = init_response.headers[MCP_SESSION_ID_HEADER]
+        negotiated_version = extract_protocol_version_from_sse(init_response)
+        headers = {
+            "Accept": "application/json, text/event-stream",
+            "Content-Type": "application/json",
+            MCP_SESSION_ID_HEADER: session_id,
+            MCP_PROTOCOL_VERSION_HEADER: negotiated_version,
+        }
+
+        async with client.stream(
+            "POST",
+            "/mcp",
+            headers=headers,
+            json={
+                "jsonrpc": "2.0",
+                "method": "tools/call",
+                "params": {"name": "wait_for_lock_with_notification", "arguments": {}},
+                "id": "_GET_stream",
+            },
+        ) as post_response:
+            assert post_response.status_code == 200
+            lines = post_response.aiter_lines()
+            with anyio.fail_after(5):
+                notification = await next_sse_data(lines)
+            assert notification["params"]["data"] == "First notification before lock"
+
+            get_headers = {
+                "Accept": "text/event-stream",
+                MCP_SESSION_ID_HEADER: session_id,
+                MCP_PROTOCOL_VERSION_HEADER: negotiated_version,
+            }
+            async with client.stream("GET", "/mcp", headers=get_headers) as get_response:
+                assert get_response.status_code == 200
+                assert get_response.headers.get("Content-Type") == "text/event-stream"
+
+            release_response = await client.post(
+                "/mcp",
+                headers=headers,
+                json={
+                    "jsonrpc": "2.0",
+                    "method": "tools/call",
+                    "params": {"name": "release_lock", "arguments": {}},
+                    "id": "release-after-get-key",
+                },
+            )
+            assert release_response.status_code == 200
+
+            with anyio.fail_after(5):
+                second_notification = await next_sse_data(lines)
+                final = await next_sse_data(lines)
+            assert second_notification["params"]["data"] == "Second notification after lock"
+            assert final["id"] == "_GET_stream"
+            assert final["result"]["content"][0]["text"] == "Completed"
+
+
 @pytest.mark.anyio
 async def test_duplicate_request_id_rejected_during_priming_event_store() -> None:
     """The duplicate-id guard holds while the event store persists the priming event.
@@ -1029,7 +1178,8 @@ def __init__(self) -> None:
             self.release = anyio.Event()
 
         async def store_event(self, stream_id: StreamId, message: types.JSONRPCMessage | None) -> EventId:
-            if stream_id == "gated-1" and message is None:
+            # Routing keys are type-tagged (`s:` for string JSON-RPC ids).
+            if stream_id == "s:gated-1" and message is None:
                 self.entered.set()
                 await self.release.wait()
             return await super().store_event(stream_id, message)