fix: use os.dup to avoid closing real stdin/stdout in stdio_server

src/mcp/server/stdio.py

@@ -17,6 +17,8 @@ async def run_server():
     ```
 """
 
+import io
+import os
 import sys
 from contextlib import asynccontextmanager
 from io import TextIOWrapper
@@ -34,14 +36,26 @@ async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.
     """Server transport for stdio: this communicates with an MCP client by reading
     from the current process' stdin and writing to stdout.
     """
-    # Purposely not using context managers for these, as we don't want to close
-    # standard process handles. Encoding of stdin/stdout as text streams on
-    # python is platform-dependent (Windows is particularly problematic), so we
-    # re-wrap the underlying binary stream to ensure UTF-8.
+    # Use os.dup() to duplicate file descriptors so that closing the wrappers
+    # doesn't close the real stdin/stdout. This allows the caller to continue
+    # using stdin/stdout after the server exits.
     if not stdin:
-        stdin = anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding="utf-8", errors="replace"))
+        try:
+            stdin_fd = os.dup(sys.stdin.fileno())
+            stdin_bin = os.fdopen(stdin_fd, "rb", closefd=True)
+            stdin = anyio.wrap_file(TextIOWrapper(stdin_bin, encoding="utf-8", errors="replace"))
+        except (io.UnsupportedOperation, ValueError):
+            # Fallback for environments where fileno() is not available
+            # (e.g., BytesIO-backed streams in tests)
+            stdin = anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding="utf-8", errors="replace"))
     if not stdout:
-        stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8"))
+        try:
+            stdout_fd = os.dup(sys.stdout.fileno())
+            stdout_bin = os.fdopen(stdout_fd, "wb", closefd=True)
+            stdout = anyio.wrap_file(TextIOWrapper(stdout_bin, encoding="utf-8"))
+        except (io.UnsupportedOperation, ValueError):
+            # Fallback for environments where fileno() is not available
+            stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8"))
 
     read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
     write_stream, write_stream_reader = create_context_streams[SessionMessage](0)

tests/server/test_stdio.py

@@ -92,3 +92,21 @@ async def test_stdio_server_invalid_utf8(monkeypatch: pytest.MonkeyPatch):
                 second = await read_stream.receive()
                 assert isinstance(second, SessionMessage)
                 assert second.message == valid
+
+
+@pytest.mark.anyio
+async def test_stdio_server_uses_os_dup():
+    """Verify stdio_server uses os.dup to avoid closing real stdin/stdout.
+
+    This verifies the fix for https://github.com/modelcontextprotocol/python-sdk/issues/1933
+    where wrapping sys.stdin.buffer/sys.stdout.buffer directly caused the real
+    file descriptors to be closed when the context manager exited.
+    """
+    import inspect
+
+    from mcp.server import stdio as stdio_module
+
+    source = inspect.getsource(stdio_module.stdio_server)
+    # The fix uses os.dup to duplicate file descriptors
+    assert "os.dup" in source
+    assert "os.fdopen" in source