Hold the stream claim for in-place fallbacks too

A transport that falls back to serving the real stdin or stdout in
place (incomplete descriptor table, or a claim that failed with
OSError) now keeps its fd in the claimed set for its lifetime, so a
second concurrent stdio_server() is refused in every shape rather than
only after a successful diversion. The sentinel now means a transport
owns the stream, not that the fd is currently diverted.

Author: maxisbey

src/mcp/server/stdio.py

@@ -27,6 +27,7 @@ async def run_server():
 from mcp.shared._context_streams import create_context_streams
 from mcp.shared.message import SessionMessage
 
+# fds whose real stream a serving transport owns, whether diverted or served in place.
 _claimed: set[int] = set()
 _claimed_lock = threading.Lock()
 
@@ -67,12 +68,19 @@ def _claim_fd(
     Raises:
         RuntimeError: fd is already claimed by another transport in this process.
     """
-    if not _is_backed_by_fd(stream, fd) or not _std_descriptors_open():
+    if not _is_backed_by_fd(stream, fd):
         return stream.buffer, None
     with _claimed_lock:
         if fd in _claimed:
             raise RuntimeError(f"another stdio_server() in this process has already claimed fd {fd}")
         _claimed.add(fd)
+
+    def unclaim() -> None:
+        with _claimed_lock:
+            _claimed.discard(fd)
+
+    if not _std_descriptors_open():
+        return stream.buffer, unclaim
     private_fd = None
     try:
         private_fd = os.dup(fd)
@@ -84,21 +92,17 @@ def _claim_fd(
         if sys.platform == "win32":  # pragma: no cover
             rebind_std_handle_to_fd(fd)
     except OSError:
-        # Undo before unclaiming: fd stays claimed for as long as it is diverted.
         if private_fd is not None:
             _restore_fd(fd, private_fd)
             os.close(private_fd)
-        with _claimed_lock:
-            _claimed.discard(fd)
-        return stream.buffer, None
+        return stream.buffer, unclaim
 
     def restore() -> None:
         # Drain text buffered during the claim (a stray print) to the diversion.
         with suppress(OSError, ValueError):
             stream.flush()
         _restore_fd(fd, private_fd)
-        with _claimed_lock:
-            _claimed.discard(fd)
+        unclaim()
 
     # closefd=False: a blocked worker thread may still read this descriptor after exit.
     return os.fdopen(private_fd, mode, closefd=False), restore

tests/server/test_stdio.py

@@ -256,6 +256,10 @@ def failing_dup2(fd: int, fd2: int, inheritable: bool = True) -> int:
                 async with read_stream:  # pragma: no branch
                     # Isolation was skipped: fd 0 is still the protocol pipe.
                     assert os.path.sameopenfile(0, in_r)
+                    # In-place transports still own the stream: a second one is refused.
+                    with pytest.raises(RuntimeError, match="already claimed fd 0"):
+                        async with stdio_server():
+                            pytest.fail("unreachable")  # pragma: no cover
                     # The spent injector passes calls through untouched.
                     os.close(os.dup(0))
                     received = await read_stream.receive()