Fix stdio_server from closing process stdio handles

src/mcp/server/stdio.py

@@ -20,6 +20,7 @@ async def run_server():
 import sys
 from contextlib import asynccontextmanager
 from io import TextIOWrapper
+from typing import BinaryIO
 
 import anyio
 import anyio.lowlevel
@@ -29,6 +30,27 @@ async def run_server():
 from mcp.shared.message import SessionMessage
 
 
+class _NonClosingTextIOWrapper(TextIOWrapper):
+    """Text wrapper that never closes the underlying binary stream.
+
+    stdio_server should not close the process' real stdin/stdout handles when its
+    background tasks wind down.
+    """
+
+    def close(self) -> None:
+        if self.closed:
+            return
+
+        # Preserve normal flush semantics for writable streams while keeping the
+        # underlying stdio handle alive.
+        if self.writable():
+            self.flush()
+
+
+def _wrap_process_stdio(binary_stream: BinaryIO) -> anyio.AsyncFile[str]:
+    return anyio.wrap_file(_NonClosingTextIOWrapper(binary_stream, encoding="utf-8"))
+
+
 @asynccontextmanager
 async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.AsyncFile[str] | None = None):
     """Server transport for stdio: this communicates with an MCP client by reading
@@ -39,9 +61,9 @@ async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.
     # python is platform-dependent (Windows is particularly problematic), so we
     # re-wrap the underlying binary stream to ensure UTF-8.
     if not stdin:
-        stdin = anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding="utf-8"))
+        stdin = _wrap_process_stdio(sys.stdin.buffer)
     if not stdout:
-        stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8"))
+        stdout = _wrap_process_stdio(sys.stdout.buffer)
 
     read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
     read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception]

tests/server/test_stdio.py

@@ -1,4 +1,5 @@
 import io
+import sys
 
 import anyio
 import pytest
@@ -59,3 +60,31 @@ async def test_stdio_server():
     assert len(received_responses) == 2
     assert received_responses[0] == JSONRPCRequest(jsonrpc="2.0", id=3, method="ping")
     assert received_responses[1] == JSONRPCResponse(jsonrpc="2.0", id=4, result={})
+
+
+@pytest.mark.anyio
+async def test_stdio_server_does_not_close_process_stdio(monkeypatch: pytest.MonkeyPatch):
+    stdin_bytes = io.BytesIO()
+    stdout_bytes = io.BytesIO()
+
+    stdin_text = io.TextIOWrapper(stdin_bytes, encoding="utf-8")
+    stdout_text = io.TextIOWrapper(stdout_bytes, encoding="utf-8")
+
+    stdin_text.write(JSONRPCRequest(jsonrpc="2.0", id=1, method="ping").model_dump_json(by_alias=True) + "\n")
+    stdin_text.seek(0)
+
+    monkeypatch.setattr(sys, "stdin", stdin_text)
+    monkeypatch.setattr(sys, "stdout", stdout_text)
+
+    async with stdio_server() as (read_stream, write_stream):
+        async with read_stream:
+            first = await read_stream.receive()
+            assert isinstance(first, SessionMessage)
+
+        async with write_stream:
+            await write_stream.send(SessionMessage(JSONRPCResponse(jsonrpc="2.0", id=2, result={})))
+
+    # Regression check for #1933: process stdio should still be writable/readable.
+    print("still-open")
+    assert stdin_text.readable()
+    assert stdout_text.writable()