fix: preserve real stdin/stdout after stdio server exits

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,27 @@ 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.
+    # We duplicate file descriptors when using the real stdin/stdout to avoid
+    # closing the process's standard handles when the TextIOWrapper is closed.
+    # This allows the process to continue using stdio normally after the server exits.
+    # For streams without a fileno() (e.g., in-memory streams in tests), we fall back
+    # to wrapping them directly.
     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 (OSError, io.UnsupportedOperation):
+            # Fallback for streams that don't support fileno() (e.g., BytesIO 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 (OSError, io.UnsupportedOperation):
+            # Fallback for streams that don't support fileno() (e.g., BytesIO in tests)
+            stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8"))
 
     read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
     read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception]

tests/server/test_stdio.py

@@ -1,5 +1,8 @@
 import io
+import os
 import sys
+import tempfile
+import warnings
 from io import TextIOWrapper
 
 import anyio
@@ -92,3 +95,58 @@ 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
+@pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning")
+async def test_stdio_server_does_not_close_real_stdio(monkeypatch: pytest.MonkeyPatch):
+    """Verify that stdio_server does not close the real stdin/stdout.
+
+    Regression test for https://github.com/modelcontextprotocol/python-sdk/issues/1933.
+    When using the default stdin/stdout (i.e., not passing custom streams),
+    the server should duplicate file descriptors so that closing the wrapper
+    does not close sys.stdin/sys.stdout.
+    """
+    # Create temp files to use as stdin/stdout (need real file descriptors)
+    with tempfile.NamedTemporaryFile(delete=False) as tmp_stdin:
+        tmp_stdin.write(b'{"jsonrpc":"2.0","id":1,"method":"ping"}\n')
+        tmp_stdin_path = tmp_stdin.name
+
+    with tempfile.NamedTemporaryFile(delete=False) as tmp_stdout:
+        tmp_stdout_path = tmp_stdout.name
+
+    stdin_wrapper = None
+    stdout_wrapper = None
+
+    try:
+        # Open the files and create wrappers that look like sys.stdin/stdout
+        stdin_file = open(tmp_stdin_path, "rb")
+        stdout_file = open(tmp_stdout_path, "wb")
+
+        stdin_wrapper = TextIOWrapper(stdin_file, encoding="utf-8")
+        stdout_wrapper = TextIOWrapper(stdout_file, encoding="utf-8")
+
+        monkeypatch.setattr(sys, "stdin", stdin_wrapper)
+        monkeypatch.setattr(sys, "stdout", stdout_wrapper)
+
+        # Run the server with default stdin/stdout
+        with anyio.fail_after(5):
+            async with stdio_server() as (read_stream, write_stream):
+                await write_stream.aclose()
+                async with read_stream:
+                    msg = await read_stream.receive()
+                    assert isinstance(msg, SessionMessage)
+
+        # After server exits, verify the original stdin/stdout are still usable
+        # The monkeypatched sys.stdin/stdout should NOT be closed
+        assert not stdin_wrapper.closed, "sys.stdin was closed by stdio_server"
+        assert not stdout_wrapper.closed, "sys.stdout was closed by stdio_server"
+
+    finally:
+        # Clean up
+        if stdin_wrapper and not stdin_wrapper.closed:
+            stdin_wrapper.close()
+        if stdout_wrapper and not stdout_wrapper.closed:
+            stdout_wrapper.close()
+        os.unlink(tmp_stdin_path)
+        os.unlink(tmp_stdout_path)