fix(server): reject concurrent duplicate JSON-RPC request ids

Stateful streamable HTTP keyed per-request streams by bare request id and
overwrote in-flight entries, cross-wiring concurrent POSTs that reused an id.
Reject the collision with 409 and reserve the slot before the priming await so
resumability cannot reopen the race.

Fixes #3137

Author: arimu1

src/mcp/server/streamable_http.py

@@ -600,6 +600,19 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
 
             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.
+            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",
+                    HTTPStatus.CONFLICT,
+                )
+                await response(scope, receive, send)
+                return
+
             if self.is_json_response_enabled:
                 self._request_streams[request_id] = anyio.create_memory_object_stream[EventMessage](
                     REQUEST_STREAM_BUFFER_SIZE
@@ -631,19 +644,25 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                     await self._clean_up_memory_streams(request_id)
                 await response(scope, receive, send)
             else:
-                # Mint the priming event before any per-request state exists:
-                # `EventStore.store_event` is user code and may raise, in which
-                # case the outer handler returns a 500 with nothing to clean up.
-                # Still strictly precedes dispatch, so storage order == wire order.
-                priming_event = await self._mint_priming_event(request_id, protocol_version)
-
-                sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[SSEEvent](0)
-                self._sse_stream_writers[request_id] = sse_stream_writer
+                # Reserve the routing slot before any await so a concurrent POST
+                # reusing this id cannot pass the guard above while we mint the
+                # priming event (EventStore.store_event is user code and may
+                # await). Release the reservation if priming raises.
                 self._request_streams[request_id] = anyio.create_memory_object_stream[EventMessage](
                     REQUEST_STREAM_BUFFER_SIZE
                 )
                 request_stream_reader = self._request_streams[request_id][1]
 
+                try:
+                    # Still strictly precedes dispatch, so storage order == wire order.
+                    priming_event = await self._mint_priming_event(request_id, protocol_version)
+                except BaseException:
+                    await self._clean_up_memory_streams(request_id)
+                    raise
+
+                sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[SSEEvent](0)
+                self._sse_stream_writers[request_id] = sse_stream_writer
+
                 headers = {
                     "Cache-Control": "no-cache, no-transform",
                     "Connection": "keep-alive",

tests/shared/test_streamable_http.py

@@ -93,6 +93,14 @@ def first_sse_data(response: httpx2.Response) -> dict[str, Any]:
     raise ValueError("No data event in SSE response")  # pragma: no cover
 
 
+async def next_sse_data(lines: AsyncIterator[str]) -> dict[str, Any]:
+    """Return the next SSE `data:` payload from a live line iterator, parsed as JSON."""
+    while True:
+        line = await anext(lines)
+        if line.startswith("data: "):
+            return json.loads(line.removeprefix("data: "))
+
+
 def extract_protocol_version_from_sse(response: httpx2.Response) -> str:
     """Extract the negotiated protocol version from an SSE initialization response."""
     return first_sse_data(response)["result"]["protocolVersion"]
@@ -808,6 +816,293 @@ async def test_get_sse_stream(basic_app: Starlette) -> None:
             assert second_get.status_code == 409
 
 
+@pytest.mark.anyio
+async def test_post_duplicate_request_id_rejected_while_first_still_in_flight(basic_app: Starlette) -> None:
+    """A second POST reusing an in-flight request's JSON-RPC id is rejected with 409; the first still completes.
+
+    Per-request routing is keyed by request id, so overwriting an in-flight slot would
+    cross-wire responses (see #3137). Raw HTTP is required to assert the 409 status code.
+    """
+    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),
+        }
+        shared_id = "shared-request-id"
+
+        async with client.stream(
+            "POST",
+            "/mcp",
+            headers=headers,
+            json={
+                "jsonrpc": "2.0",
+                "method": "tools/call",
+                "params": {"name": "wait_for_lock_with_notification", "arguments": {}},
+                "id": shared_id,
+            },
+        ) as first_response:
+            assert first_response.status_code == 200
+            lines = first_response.aiter_lines()
+            # First notification confirms the request is registered and blocked on the lock.
+            with anyio.fail_after(5):
+                notification = await next_sse_data(lines)
+            assert notification["params"]["data"] == "First notification before lock"
+
+            duplicate_response = await client.post(
+                "/mcp",
+                headers=headers,
+                json={
+                    "jsonrpc": "2.0",
+                    "method": "tools/call",
+                    "params": {"name": "test_tool", "arguments": {}},
+                    "id": shared_id,
+                },
+            )
+            assert duplicate_response.status_code == 409
+            error = duplicate_response.json()["error"]
+            assert error["code"] == INVALID_REQUEST
+            assert "already in flight" in error["message"]
+
+            release_response = await client.post(
+                "/mcp",
+                headers=headers,
+                json={
+                    "jsonrpc": "2.0",
+                    "method": "tools/call",
+                    "params": {"name": "release_lock", "arguments": {}},
+                    "id": "release-lock-1",
+                },
+            )
+            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"] == shared_id
+            assert final["result"]["content"][0]["text"] == "Completed"
+
+
+@pytest.mark.anyio
+async def test_json_response_duplicate_request_id_rejected_while_first_still_in_flight(json_app: Starlette) -> None:
+    """In JSON response mode, a second POST reusing an in-flight id is rejected with 409; the first still completes.
+
+    Same collision as the SSE case, exercised on the JSON branch of `_handle_post_request` (#3137).
+    """
+    async with make_client(json_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: init_response.json()["result"]["protocolVersion"],
+        }
+        shared_id = "shared-json-request-id"
+        first_result: dict[str, Any] = {}
+
+        async def run_first_request() -> None:
+            response = await client.post(
+                "/mcp",
+                headers=headers,
+                json={
+                    "jsonrpc": "2.0",
+                    "method": "tools/call",
+                    "params": {"name": "wait_for_lock_with_notification", "arguments": {}},
+                    "id": shared_id,
+                },
+            )
+            assert response.status_code == 200
+            first_result.update(response.json())
+
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(run_first_request)
+            # First request registers its stream then blocks server-side on the lock;
+            # nothing else can run until that await yields.
+            await anyio.wait_all_tasks_blocked()
+
+            duplicate_response = await client.post(
+                "/mcp",
+                headers=headers,
+                json={
+                    "jsonrpc": "2.0",
+                    "method": "tools/call",
+                    "params": {"name": "test_tool", "arguments": {}},
+                    "id": shared_id,
+                },
+            )
+            assert duplicate_response.status_code == 409
+            error = duplicate_response.json()["error"]
+            assert error["code"] == INVALID_REQUEST
+            assert "already in flight" in error["message"]
+
+            release_response = await client.post(
+                "/mcp",
+                headers=headers,
+                json={
+                    "jsonrpc": "2.0",
+                    "method": "tools/call",
+                    "params": {"name": "release_lock", "arguments": {}},
+                    "id": "release-lock-json-1",
+                },
+            )
+            assert release_response.status_code == 200
+
+        assert first_result["id"] == shared_id
+        assert first_result["result"]["content"][0]["text"] == "Completed"
+
+
+@pytest.mark.anyio
+async def test_request_id_reuse_after_completion_allowed(basic_app: Starlette) -> None:
+    """A request id may be reused once the earlier request with that id has completed.
+
+    Only concurrent reuse is ambiguous to route; sequential reuse is allowed (some
+    clients send a constant id for every request).
+    """
+    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),
+        }
+
+        for _ in range(2):
+            response = await client.post(
+                "/mcp",
+                headers=headers,
+                json={
+                    "jsonrpc": "2.0",
+                    "id": 1,
+                    "method": "tools/call",
+                    "params": {"name": "test_tool", "arguments": {}},
+                },
+            )
+            assert response.status_code == 200
+            body = first_sse_data(response)
+            assert body["id"] == 1
+            assert body["result"]["content"][0]["text"] == "Called test_tool"
+
+
+@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.
+
+    With an event store, the SSE branch awaits `EventStore.store_event` before the
+    response starts. The routing slot is reserved before that await so a concurrent
+    POST reusing the id cannot slip past the guard and overwrite the slot (#3137).
+    """
+
+    class GatedEventStore(SimpleEventStore):
+        """Blocks the priming write for the gated stream until released."""
+
+        def __init__(self) -> None:
+            super().__init__()
+            self.entered = anyio.Event()
+            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:
+                self.entered.set()
+                await self.release.wait()
+            return await super().store_event(stream_id, message)
+
+    def first_message_sse_data(response: httpx2.Response) -> dict[str, Any]:
+        """First non-empty SSE data payload; skips the empty-data priming event."""
+        for line in response.text.splitlines():
+            if line.startswith("data: ") and line.removeprefix("data: ").strip():
+                return json.loads(line.removeprefix("data: "))
+        raise ValueError("No message data event in SSE response")  # pragma: no cover
+
+    store = GatedEventStore()
+    async with running_app(event_store=store) as app:
+        async with make_client(app) as client:
+            # Priming events require protocol >= 2025-11-25.
+            init_request = {
+                "jsonrpc": "2.0",
+                "method": "initialize",
+                "params": {
+                    "clientInfo": {"name": "test-client", "version": "1.0"},
+                    "protocolVersion": types.LATEST_PROTOCOL_VERSION,
+                    "capabilities": {},
+                },
+                "id": "init-1",
+            }
+            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: first_message_sse_data(init_response)["result"]["protocolVersion"],
+            }
+            call: dict[str, Any] = {
+                "jsonrpc": "2.0",
+                "id": "gated-1",
+                "method": "tools/call",
+                "params": {"name": "test_tool", "arguments": {}},
+            }
+            results: dict[str, httpx2.Response] = {}
+
+            async def post_first() -> None:
+                results["first"] = await client.post("/mcp", headers=headers, json=call)
+
+            async with anyio.create_task_group() as tg:
+                tg.start_soon(post_first)
+                with anyio.fail_after(5):
+                    await store.entered.wait()
+
+                # Without early slot reservation this POST would also suspend in the
+                # gated store and hang; fail_after turns that regression into a fast fail.
+                with anyio.fail_after(5):
+                    duplicate = await client.post("/mcp", headers=headers, json=call)
+                assert duplicate.status_code == 409
+                error = duplicate.json()["error"]
+                assert error["code"] == INVALID_REQUEST
+                assert "already in flight" in error["message"]
+
+                store.release.set()
+
+            assert results["first"].status_code == 200
+            body = first_message_sse_data(results["first"])
+            assert body["id"] == "gated-1"
+            assert body["result"]["content"][0]["text"] == "Called test_tool"
+
+
 @pytest.mark.anyio
 async def test_get_validation(basic_app: Starlette) -> None:
     """A GET without an Accept header covering text/event-stream is rejected with 406."""