fix: never mint a request id already used by a completed caller-supplied id

src/mcp/shared/direct_dispatcher.py

@@ -250,14 +250,18 @@ async def _dispatch_request(
                     in_flight_key = coerce_request_id(request_id)
                     if in_flight_key in self._in_flight_ids:
                         raise ValueError(f"request id {request_id!r} is already in flight")
+                    # Advance the mint counter past the supplied key: ids are
+                    # single-use within a session, so a minted id may not
+                    # revisit this key even after the request completes.
+                    if isinstance(in_flight_key, int):
+                        self._next_id = max(self._next_id, in_flight_key)
                 else:
                     # Synthesize an id (the DispatchContext contract reserves None
-                    # for notifications), minting past any key a supplied id
-                    # occupies: the collision error is reserved for the caller
-                    # who actually chose the id.
+                    # for notifications). Cannot collide: the counter is past
+                    # every supplied integer key (advanced above) and every
+                    # previously minted id; the collision error is reserved for
+                    # the caller who actually chose the id.
                     self._next_id += 1
-                    while self._next_id in self._in_flight_ids:
-                        self._next_id += 1
                     request_id = self._next_id
                     in_flight_key = request_id
                 self._in_flight_ids.add(in_flight_key)

src/mcp/shared/dispatcher.py

@@ -87,8 +87,10 @@ class CallOptions(TypedDict, total=False):
     Callers that need to know a request's id before its result arrives (a
     `subscriptions/listen` stream is demultiplexed by it) mint their own ids
     here; string ids that don't parse as integers can never collide with the
-    dispatcher's minted sequence. Per the class contract, dispatchers that
-    predate this key ignore it and mint as usual.
+    dispatcher's minted sequence, and a supplied integer (or numeric-string)
+    key advances that sequence past itself, so a minted id never reuses a
+    supplied one. Per the class contract, dispatchers that predate this key
+    ignore it and mint as usual.
     """
 
     timeout: float

src/mcp/shared/jsonrpc_dispatcher.py

@@ -335,12 +335,16 @@ async def send_raw_request(
             pending_key = coerce_request_id(request_id)
             if pending_key in self._pending:
                 raise ValueError(f"request id {request_id!r} is already in flight")
+            # Advance the mint counter past the supplied key: ids are single-use
+            # within a session, so a minted id may not revisit this key even
+            # after the request completes.
+            if isinstance(pending_key, int):
+                self._next_id = max(self._next_id, pending_key)
         else:
-            # Mint past any key a supplied id occupies: the collision error is
-            # reserved for the caller who actually chose the id.
+            # Cannot collide: the counter is past every supplied integer key
+            # (advanced above) and every previously minted id. The collision
+            # error is reserved for the caller who actually chose the id.
             request_id = self._allocate_id()
-            while request_id in self._pending:
-                request_id = self._allocate_id()
             pending_key = request_id
         out_params = dict(params) if params is not None else {}
         out_meta = dict(out_params.get("_meta") or {})

tests/shared/test_dispatcher.py

@@ -474,11 +474,35 @@ async def parked() -> None:
 
                 tg.start_soon(parked)
                 await entered.wait()
-                # The counter mints 1 and 2, then skips the occupied 3 to 4.
+                # Accepting "3" advanced the counter past its coerced key, so
+                # the mints start at 4.
                 for _ in range(3):
                     await client.send_raw_request("plain", None)
                 release.set()
-            assert [request_id for request_id in seen_ids if request_id != "3"] == [1, 2, 4]
+            assert [request_id for request_id in seen_ids if request_id != "3"] == [4, 5, 6]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("supplied_id", [2, "2"], ids=["int", "numeric-string"])
+async def test_minted_ids_never_reuse_a_completed_supplied_id(pair_factory: PairFactory, supplied_id: RequestId):
+    """Ids are single-use within a session (spec: "The request ID MUST NOT have been
+    previously used by the requestor within the same session"), so the mint counter
+    skips a supplied integer key even after that request completes — a peer that
+    tracks ids per session would see two different requests under one id."""
+    seen_ids: list[RequestId | None] = []
+
+    async def record(
+        ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
+    ) -> dict[str, Any]:
+        seen_ids.append(ctx.request_id)
+        return {}
+
+    async with running_pair(pair_factory, server_on_request=record) as (client, *_):
+        with anyio.fail_after(5):
+            await client.send_raw_request("supplied", None, {"request_id": supplied_id})
+            await client.send_raw_request("plain", None)
+            await client.send_raw_request("plain", None)
+    assert seen_ids == [supplied_id, 3, 4]
 
 
 @pytest.mark.anyio