fix: prevent stdio_server from closing real stdin/stdout

Duplicate stdin/stdout file descriptors with os.dup() before wrapping
in TextIOWrapper. Without this, closing the wrapper also closes
sys.stdin.buffer / sys.stdout.buffer, breaking any subsequent stdio
operations in the caller process.

Fixes #1933

Author: EvanYao826

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,28 @@ 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.
+    # Duplicate stdin/stdout file descriptors so that closing the wrappers does
+    # not close the underlying process handles.  Without dup(), TextIOWrapper
+    # takes ownership of sys.stdin.buffer / sys.stdout.buffer and closes them on
+    # __exit__, breaking any subsequent stdio operations in the caller (see #1933).
+    #
+    # When stdin/stdout lack real file descriptors (e.g. io.BytesIO in tests),
+    # fall back to wrapping the buffer directly — there are no process handles
+    # to protect in that case.
     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)
+        except io.UnsupportedOperation:
+            stdin_bin = sys.stdin.buffer
+        stdin = anyio.wrap_file(TextIOWrapper(stdin_bin, 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)
+        except io.UnsupportedOperation:
+            stdout_bin = sys.stdout.buffer
+        stdout = anyio.wrap_file(TextIOWrapper(stdout_bin, 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

@@ -2,6 +2,8 @@
 import sys
 from io import TextIOWrapper
 
+from pathlib import Path
+
 import anyio
 import pytest
 
@@ -63,6 +65,54 @@ async def test_stdio_server():
     assert received_responses[1] == JSONRPCResponse(jsonrpc="2.0", id=4, result={})
 
 
+@pytest.mark.anyio
+@pytest.mark.filterwarnings("default:unclosed file:ResourceWarning")
+async def test_stdio_server_does_not_close_sys_stdin_stdout(tmp_path: Path):
+    """Exiting stdio_server must not close the real sys.stdin / sys.stdout.
+
+    Regression test for https://github.com/modelcontextprotocol/python-sdk/issues/1933.
+    Uses real file descriptors via os.pipe() to exercise the os.dup() path.
+    """
+    import os
+
+    valid = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+    payload = valid.model_dump_json(by_alias=True, exclude_none=True).encode() + b"\n"
+
+    # Create a pipe with the MCP message, and a temp file for stdout.
+    read_fd, write_fd = os.pipe()
+    os.write(write_fd, payload)
+    os.close(write_fd)
+
+    stdout_path = tmp_path / "stdout.bin"
+    stdout_fd = os.open(str(stdout_path), os.O_CREAT | os.O_WRONLY | os.O_TRUNC)
+
+    # Save originals and replace with our pipe/file.
+    orig_stdin = sys.stdin
+    orig_stdout = sys.stdout
+    test_stdin = os.fdopen(read_fd, "rb")
+    test_stdout = os.fdopen(stdout_fd, "wb")
+    sys.stdin = test_stdin
+    sys.stdout = test_stdout
+
+    try:
+        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 exiting the server, the original sys.stdin / sys.stdout must
+        # still be usable — the wrappers must NOT have closed them.
+        assert not sys.stdin.closed, "sys.stdin was closed by stdio_server"
+        assert not sys.stdout.closed, "sys.stdout was closed by stdio_server"
+    finally:
+        sys.stdin = orig_stdin
+        sys.stdout = orig_stdout
+        test_stdin.close()
+        test_stdout.close()
+
+
 @pytest.mark.anyio
 async def test_stdio_server_invalid_utf8(monkeypatch: pytest.MonkeyPatch):
     """Non-UTF-8 bytes on stdin must not crash the server.