feat(server): forward optional stdin/stdout to stdio_server

MCPServer.run_stdio_async and run(..., transport="stdio") now accept
optional async text streams, matching mcp.server.stdio.stdio_server.

This supports hosts that redirect process stdout for logging while
keeping a dedicated MCP JSON-RPC stream.

Made-with: Cursor

Author: mangelajo

src/mcp/server/mcpserver/server.py

@@ -238,7 +238,13 @@ def session_manager(self) -> StreamableHTTPSessionManager:
         return self._lowlevel_server.session_manager  # pragma: no cover
 
     @overload
-    def run(self, transport: Literal["stdio"] = ...) -> None: ...
+    def run(
+        self,
+        transport: Literal["stdio"] = ...,
+        *,
+        stdin: anyio.AsyncFile[str] | None = ...,
+        stdout: anyio.AsyncFile[str] | None = ...,
+    ) -> None: ...
 
     @overload
     def run(
@@ -270,12 +276,19 @@ def run(
     def run(
         self,
         transport: Literal["stdio", "sse", "streamable-http"] = "stdio",
+        *,
+        stdin: anyio.AsyncFile[str] | None = None,
+        stdout: anyio.AsyncFile[str] | None = None,
         **kwargs: Any,
     ) -> None:
         """Run the MCP server. Note this is a synchronous function.
 
         Args:
             transport: Transport protocol to use ("stdio", "sse", or "streamable-http")
+            stdin: Optional async text stream for MCP input (stdio transport only).
+                When omitted, uses process stdin. See :func:`mcp.server.stdio.stdio_server`.
+            stdout: Optional async text stream for MCP output (stdio transport only).
+                When omitted, uses process stdout.
             **kwargs: Transport-specific options (see overloads for details)
         """
         TRANSPORTS = Literal["stdio", "sse", "streamable-http"]
@@ -284,7 +297,7 @@ def run(
 
         match transport:
             case "stdio":
-                anyio.run(self.run_stdio_async)
+                anyio.run(lambda: self.run_stdio_async(stdin=stdin, stdout=stdout))
             case "sse":  # pragma: no cover
                 anyio.run(lambda: self.run_sse_async(**kwargs))
             case "streamable-http":  # pragma: no cover
@@ -836,9 +849,25 @@ def decorator(  # pragma: no cover
 
         return decorator  # pragma: no cover
 
-    async def run_stdio_async(self) -> None:
-        """Run the server using stdio transport."""
-        async with stdio_server() as (read_stream, write_stream):
+    async def run_stdio_async(
+        self,
+        *,
+        stdin: anyio.AsyncFile[str] | None = None,
+        stdout: anyio.AsyncFile[str] | None = None,
+    ) -> None:
+        """Run the server using stdio transport.
+
+        Args:
+            stdin: Async text stream to read JSON-RPC lines from. When ``None``,
+                uses the process stdin (see :func:`mcp.server.stdio.stdio_server`).
+            stdout: Async text stream to write JSON-RPC lines to. When ``None``,
+                uses the process stdout.
+
+        Custom streams are useful when the process ``sys.stdout`` / ``sys.stdin``
+        must be redirected (for example so logging or subprocess output does not
+        corrupt the MCP JSON-RPC stream on fd 1).
+        """
+        async with stdio_server(stdin=stdin, stdout=stdout) as (read_stream, write_stream):
             await self._lowlevel_server.run(
                 read_stream,
                 write_stream,

tests/server/mcpserver/test_run_stdio_custom_streams.py

@@ -0,0 +1,44 @@
+"""MCPServer.run_stdio_async forwards optional stdin/stdout to stdio_server."""
+
+from __future__ import annotations
+
+import io
+from contextlib import asynccontextmanager
+from unittest.mock import AsyncMock
+
+import anyio
+import pytest
+
+from mcp.server.mcpserver import MCPServer
+
+
+@pytest.mark.anyio
+async def test_run_stdio_async_passes_streams_to_stdio_server(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, object] = {}
+
+    @asynccontextmanager
+    async def spy_stdio_server(stdin=None, stdout=None):
+        captured["stdin"] = stdin
+        captured["stdout"] = stdout
+        read_stream = AsyncMock()
+        write_stream = AsyncMock()
+        yield read_stream, write_stream
+
+    async def noop_run(*_args, **_kwargs):
+        return None
+
+    monkeypatch.setattr("mcp.server.mcpserver.server.stdio_server", spy_stdio_server)
+
+    server = MCPServer("test-stdio-spy")
+    monkeypatch.setattr(server._lowlevel_server, "run", noop_run)
+    monkeypatch.setattr(server._lowlevel_server, "create_initialization_options", lambda: object())
+
+    sin = io.StringIO()
+    sout = io.StringIO()
+    await server.run_stdio_async(
+        stdin=anyio.AsyncFile(sin),
+        stdout=anyio.AsyncFile(sout),
+    )
+
+    assert captured["stdin"] is not None
+    assert captured["stdout"] is not None