Fix concurrent requests sharing a JSON-RPC id hijacking each other's responses

_handle_post_request unconditionally overwrote self._request_streams[request_id],
so a second concurrent POST reusing an in-flight request's id would silently
take over that request's stream and both callers would race for a single
response. Reject the second request with a 409 duplicate-request-id error
instead of overwriting the existing in-flight entry, for both the JSON and
SSE response modes.

Fixes #3137

Author: tirthfx

src/mcp/server/streamable_http.py

@@ -550,6 +550,13 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
             request_id = str(message.id)
 
             if self.is_json_response_enabled:
+                if request_id in self._request_streams:
+                    response = self._create_error_response(
+                        "Duplicate request id: a request with this id is already in flight on this session.",
+                        HTTPStatus.CONFLICT,
+                    )
+                    await response(scope, receive, send)
+                    return
                 self._request_streams[request_id] = anyio.create_memory_object_stream[EventMessage](
                     REQUEST_STREAM_BUFFER_SIZE
                 )
@@ -570,7 +577,7 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                             response_message = event_message.message
                             break
                         # For notifications and requests, keep waiting
-                        else:  # pragma: no cover
+                        else:
                             logger.debug(f"received: {event_message.message.method}")
 
                     # At this point we should have a response
@@ -597,6 +604,14 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 finally:
                     await self._clean_up_memory_streams(request_id)
             else:
+                if request_id in self._request_streams:
+                    response = self._create_error_response(
+                        "Duplicate request id: a request with this id is already in flight on this session.",
+                        HTTPStatus.CONFLICT,
+                    )
+                    await response(scope, receive, send)
+                    return
+
                 # 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.

tests/shared/test_streamable_http.py

@@ -809,6 +809,169 @@ 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; the first still completes."""
+    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
+        negotiated_version = extract_protocol_version_from_sse(init_response)
+        session_id = init_response.headers.get(MCP_SESSION_ID_HEADER)
+        assert session_id is not None
+
+        headers = {
+            "Accept": "application/json, text/event-stream",
+            "Content-Type": "application/json",
+            MCP_SESSION_ID_HEADER: session_id,
+            MCP_PROTOCOL_VERSION_HEADER: negotiated_version,
+        }
+        shared_id = "shared-request-id"
+
+        first_request_blocked = anyio.Event()
+        first_response_lines: list[str] = []
+
+        async def run_first_request() -> None:
+            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
+                async for line in first_response.aiter_lines():  # pragma: no branch
+                    first_response_lines.append(line)
+                    if line.startswith("data: ") and "First notification before lock" in line:
+                        first_request_blocked.set()
+                    if line.startswith("data: ") and '"result"' in line:
+                        break
+
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(run_first_request)
+
+            with anyio.fail_after(5):
+                await first_request_blocked.wait()
+
+            # The first request is still in flight (blocked on the lock), holding `shared_id`.
+            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
+            assert "Duplicate request id" in duplicate_response.text
+
+            # Release the lock so the first request can finish.
+            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
+
+        final_data_line = next(line for line in reversed(first_response_lines) if line.startswith("data: "))
+        final_payload = json.loads(final_data_line.removeprefix("data: "))
+        assert final_payload["id"] == shared_id
+        assert final_payload["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; the first still completes."""
+    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
+        session_id = init_response.headers.get(MCP_SESSION_ID_HEADER)
+        assert session_id is not None
+        negotiated_version = init_response.json()["result"]["protocolVersion"]
+
+        headers = {
+            "Accept": "application/json, text/event-stream",
+            "Content-Type": "application/json",
+            MCP_SESSION_ID_HEADER: session_id,
+            MCP_PROTOCOL_VERSION_HEADER: negotiated_version,
+        }
+        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)
+
+            # The first request registers its request stream and blocks on the lock before
+            # anything else in this task group can run.
+            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
+            assert "Duplicate request id" in duplicate_response.text
+
+            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_get_validation(basic_app: Starlette) -> None:
     """A GET without an Accept header covering text/event-stream is rejected with 406."""