fix: support fileno-less stdio streams

Author: pranjalbhatia710

src/mcp/server/stdio.py

@@ -20,7 +20,7 @@ async def run_server():
 import os
 import sys
 from contextlib import asynccontextmanager
-from io import TextIOWrapper
+from io import TextIOWrapper, UnsupportedOperation
 
 import anyio
 import anyio.lowlevel
@@ -30,6 +30,33 @@ async def run_server():
 from mcp.shared.message import SessionMessage
 
 
+def _wrap_stdin() -> tuple[anyio.AsyncFile[str], bool]:
+    """Wrap stdin as UTF-8 text without closing process stdio on exit."""
+    try:
+        stdin_fd = os.dup(sys.stdin.fileno())
+    except (AttributeError, OSError, UnsupportedOperation):
+        # Some tests and embedders replace sys.stdin with fileno-less in-memory
+        # streams. Keep supporting that shape by falling back to the existing
+        # buffer-wrapping behavior.
+        return anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding="utf-8", errors="replace")), False
+
+    stdin_buffer = os.fdopen(stdin_fd, "rb", closefd=True)
+    return anyio.wrap_file(TextIOWrapper(stdin_buffer, encoding="utf-8", errors="replace")), True
+
+
+def _wrap_stdout() -> tuple[anyio.AsyncFile[str], bool]:
+    """Wrap stdout as UTF-8 text without closing process stdio on exit."""
+    try:
+        stdout_fd = os.dup(sys.stdout.fileno())
+    except (AttributeError, OSError, UnsupportedOperation):
+        # Match the fileno-less stdin fallback for in-memory test streams and
+        # embedders that provide file-like stdout objects.
+        return anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8")), False
+
+    stdout_buffer = os.fdopen(stdout_fd, "wb", closefd=True)
+    return anyio.wrap_file(TextIOWrapper(stdout_buffer, encoding="utf-8")), True
+
+
 @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
@@ -42,15 +69,9 @@ async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.
     close_stdin = False
     close_stdout = False
     if not stdin:
-        stdin_fd = os.dup(sys.stdin.fileno())
-        stdin_buffer = os.fdopen(stdin_fd, "rb", closefd=True)
-        stdin = anyio.wrap_file(TextIOWrapper(stdin_buffer, encoding="utf-8", errors="replace"))
-        close_stdin = True
+        stdin, close_stdin = _wrap_stdin()
     if not stdout:
-        stdout_fd = os.dup(sys.stdout.fileno())
-        stdout_buffer = os.fdopen(stdout_fd, "wb", closefd=True)
-        stdout = anyio.wrap_file(TextIOWrapper(stdout_buffer, encoding="utf-8"))
-        close_stdout = True
+        stdout, close_stdout = _wrap_stdout()
 
     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

@@ -64,6 +64,27 @@ async def test_stdio_server():
     assert received_responses[1] == JSONRPCResponse(jsonrpc="2.0", id=4, result={})
 
 
+@pytest.mark.anyio
+async def test_stdio_server_supports_fileno_less_standard_streams(monkeypatch: pytest.MonkeyPatch):
+    """The default path supports in-memory stdio replacements without fileno()."""
+    request = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+    raw_stdin = io.BytesIO(request.model_dump_json(by_alias=True, exclude_none=True).encode() + b"\n")
+    raw_stdout = io.BytesIO()
+
+    test_stdin = TextIOWrapper(raw_stdin, encoding="utf-8")
+    test_stdout = TextIOWrapper(raw_stdout, encoding="utf-8")
+    monkeypatch.setattr(sys, "stdin", test_stdin)
+    monkeypatch.setattr(sys, "stdout", test_stdout)
+
+    with anyio.fail_after(5):
+        async with stdio_server() as (read_stream, write_stream):
+            await write_stream.aclose()
+            async with read_stream:  # pragma: no branch
+                message = await read_stream.receive()
+                assert isinstance(message, SessionMessage)
+                assert message.message == request
+
+
 @pytest.mark.anyio
 async def test_stdio_server_invalid_utf8():
     """Non-UTF-8 bytes on stdin must not crash the server.