Let stdio_server exit while the peer holds stdin open

The stdin reader ran each readline as an uncancellable AnyIO worker-thread
call, so leaving stdio_server() -- by returning, by SIGINT, or after a
broken stdout -- waited for a line or EOF that a peer keeping stdin open
never sends. Even once abandoned, a non-daemon worker blocks interpreter
exit, and a read parked in sys.stdin's BufferedReader makes finalization
abort with `_enter_buffered_busy`.

Read each line in a daemon thread awaited through an abandonable future,
read stdin through its raw (lock-free) layer, and cancel the reader when
the caller leaves the context.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: weijie-tan3

src/mcp/server/stdio.py

@@ -11,17 +11,20 @@ async def run_server():
     ```
 """
 
+import io
 import os
 import sys
 import threading
 from collections.abc import Callable
+from concurrent.futures import Future, InvalidStateError
 from contextlib import asynccontextmanager, suppress
 from dataclasses import dataclass
 from io import TextIOWrapper
-from typing import BinaryIO, Literal, TextIO
+from typing import IO, BinaryIO, Literal, TextIO, cast
 
 import anyio
 import anyio.lowlevel
+import anyio.to_thread
 import mcp_types as types
 
 from mcp.os.win32.utilities import rebind_std_handle_to_fd
@@ -158,6 +161,44 @@ def release() -> None:
     return os.fdopen(private_fd, mode, closefd=False), release
 
 
+def _lock_free_reader(buffer: BinaryIO) -> BinaryIO:
+    """The raw layer beneath a buffered stdin, when it has one.
+
+    A reader thread can still be parked in a read at interpreter exit. Through a
+    BufferedReader that read holds the buffer's lock, and finalization aborts with
+    `_enter_buffered_busy` when it closes sys.stdin; the raw layer takes no lock.
+    """
+    raw = getattr(buffer, "raw", None)
+    # TextIOWrapper reads a raw stream through read(), which RawIOBase provides.
+    return cast(BinaryIO, raw) if isinstance(raw, io.RawIOBase) else buffer
+
+
+def _readline_into(file: IO[str], future: "Future[str]") -> None:
+    try:
+        line = file.readline()
+    except Exception as exc:  # thread entry point: every failure is handed to the awaiting task
+        with suppress(InvalidStateError):
+            future.set_exception(exc)
+    else:
+        with suppress(InvalidStateError):
+            future.set_result(line)
+
+
+async def _readline(file: IO[str]) -> str:
+    """Read one line without letting a blocked read outlive cancellation or hold up exit.
+
+    The read runs in a daemon thread, so neither cancellation nor interpreter exit
+    waits for input that may never arrive (a peer that keeps stdin open, SIGINT, or a
+    closed stdout). Cancelling the future releases the AnyIO worker waiting on it.
+    """
+    future: Future[str] = Future()
+    threading.Thread(target=_readline_into, args=(file, future), name="mcp-stdio-reader", daemon=True).start()
+    try:
+        return await anyio.to_thread.run_sync(future.result, abandon_on_cancel=True)
+    finally:
+        future.cancel()
+
+
 @asynccontextmanager
 async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.AsyncFile[str] | None = None):
     """Serve MCP over the process's stdin and stdout.
@@ -173,27 +214,32 @@ async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.
     try:
         if not stdin:
             stdin_buffer, restore_stdin = _claim_fd(0, sys.stdin, "rb", _open_stdin_diversion)
-            stdin = anyio.wrap_file(_UnownedTextWrapper(stdin_buffer, encoding="utf-8", errors="replace"))
+            stdin = anyio.wrap_file(
+                _UnownedTextWrapper(_lock_free_reader(stdin_buffer), encoding="utf-8", errors="replace")
+            )
         if not stdout:
             stdout_buffer, restore_stdout = _claim_fd(1, sys.stdout, "wb", _open_stdout_diversion)
             stdout = anyio.wrap_file(_UnownedTextWrapper(stdout_buffer, encoding="utf-8"))
 
         read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
         write_stream, write_stream_reader = create_context_streams[SessionMessage](0)
 
+        stdin_reader_scope = anyio.CancelScope()
+
         async def stdin_reader():
             try:
                 async with read_stream_writer:
-                    async for line in stdin:
-                        try:
-                            message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
-                        except Exception as exc:
-                            await read_stream_writer.send(exc)
-                            continue
-
-                        session_message = SessionMessage(message)
-                        await read_stream_writer.send(session_message)
-            except anyio.ClosedResourceError:  # pragma: no cover
+                    with stdin_reader_scope:
+                        while line := await _readline(stdin.wrapped):
+                            try:
+                                message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
+                            except Exception as exc:
+                                await read_stream_writer.send(exc)
+                                continue
+
+                            session_message = SessionMessage(message)
+                            await read_stream_writer.send(session_message)
+            except anyio.ClosedResourceError:  # pragma: lax no cover
                 await anyio.lowlevel.checkpoint()
 
         async def stdout_writer():
@@ -203,13 +249,17 @@ async def stdout_writer():
                         json = session_message.message.model_dump_json(by_alias=True, exclude_unset=True)
                         await stdout.write(json + "\n")
                         await stdout.flush()
-            except anyio.ClosedResourceError:  # pragma: no cover
+            except anyio.ClosedResourceError:  # pragma: lax no cover
                 await anyio.lowlevel.checkpoint()
 
         async with anyio.create_task_group() as tg:
             tg.start_soon(stdin_reader)
             tg.start_soon(stdout_writer)
-            yield read_stream, write_stream
+            try:
+                yield read_stream, write_stream
+            finally:
+                # The caller is done; a peer holding stdin open must not keep the transport alive.
+                stdin_reader_scope.cancel()
     finally:
         if restore_stdout is not None:
             restore_stdout()

tests/server/test_stdio.py

@@ -8,6 +8,7 @@
 from io import TextIOWrapper
 
 import anyio
+import anyio.lowlevel
 import anyio.to_thread
 import pytest
 from mcp_types import (
@@ -315,6 +316,63 @@ def flaky_dup2(fd: int, fd2: int, inheritable: bool = True) -> int:
                 pytest.fail("unreachable")  # pragma: no cover
 
 
+@contextmanager
+def _unblock_a_hung_stdin_read_after(seconds: float, write_fd: int) -> Iterator[None]:
+    """Writes a stray line after `seconds`, so an uncancellable stdin read fails the test instead of hanging it."""
+    watchdog = threading.Timer(seconds, os.write, (write_fd, b"\n"))
+    watchdog.start()
+    try:
+        yield
+    finally:
+        watchdog.cancel()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("leaving", ["returns", "is_cancelled"])
+async def test_stdio_server_exits_while_the_peer_holds_stdin_open(
+    leaving: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Leaving the transport does not wait for a stdin line or EOF that may never come.
+
+    Regression: the stdin read was an uncancellable worker-thread call, so returning
+    from the context, a closed stdout, or SIGINT hung until the peer sent more input.
+    """
+    with _pipe_planted_on_fd0(monkeypatch) as (_, in_w):
+        monkeypatch.setattr(sys, "stdout", TextIOWrapper(io.BytesIO(), encoding="utf-8"))
+        try:
+            with _unblock_a_hung_stdin_read_after(10, in_w), anyio.fail_after(5):
+                with anyio.CancelScope() as scope:
+                    async with stdio_server() as (read_stream, write_stream):
+                        await anyio.wait_all_tasks_blocked()  # the reader is now parked on stdin
+                        read_stream.close()
+                        await write_stream.aclose()
+                        if leaving == "is_cancelled":
+                            scope.cancel()
+                            await anyio.sleep_forever()
+                assert scope.cancelled_caught == (leaving == "is_cancelled")
+                # A hang that outlived the deadline surfaces here as TimeoutError.
+                await anyio.lowlevel.checkpoint()
+        finally:
+            os.close(in_w)
+
+
+@pytest.mark.anyio
+async def test_stdio_server_surfaces_a_failing_stdin_read() -> None:
+    """An error reading stdin propagates out of the transport rather than being swallowed."""
+
+    class _FailingStdin(io.StringIO):
+        def readline(self, size: int | None = -1, /) -> str:
+            raise OSError("stdin read failed")
+
+    stdin = anyio.AsyncFile(_FailingStdin())
+    stdout = anyio.AsyncFile(io.StringIO())
+    with anyio.fail_after(5), pytest.RaisesGroup(pytest.RaisesExc(OSError, match="stdin read failed")):
+        # Coverage mis-traces nested `async with` exit arcs on Python 3.11+.
+        async with stdio_server(stdin=stdin, stdout=stdout) as (read_stream, write_stream):  # pragma: no branch
+            async with read_stream, write_stream:  # pragma: no branch
+                await anyio.sleep_forever()
+
+
 @pytest.mark.anyio
 async def test_stdio_server_takes_stdout_off_the_descriptor_table_while_serving(
     monkeypatch: pytest.MonkeyPatch,

tests/transports/stdio/test_lifecycle.py

@@ -11,12 +11,13 @@
 import anyio
 import anyio.abc
 import pytest
-from mcp_types import TextContent
+from mcp_types import JSONRPCRequest, TextContent
 
 from mcp.client import stdio
 from mcp.client.client import Client
 from mcp.client.stdio import StdioServerParameters, stdio_client
 from mcp.os.win32.utilities import FallbackProcess
+from mcp.shared.message import SessionMessage
 from tests.transports.stdio._liveness import (
     accept_alive,
     assert_stream_closed,
@@ -258,3 +259,58 @@ def run_noisy_child() -> str:
     assert isinstance(content, TextContent)
     assert content.text == "0"
     assert "this is not json" in server_stderr
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("stdin_claim", ["isolated", "in_place"])
+async def test_a_server_that_stops_serving_while_stdin_stays_open_exits_cleanly(
+    stdin_claim: str,
+    tmp_path: Path,
+    spawned_processes: list[anyio.abc.Process | FallbackProcess],
+    terminate_calls: list[anyio.abc.Process | FallbackProcess],
+) -> None:
+    """Leaving stdio_server() while the client holds stdin open lets the interpreter exit.
+
+    The stdin read still parked at exit must neither block shutdown (a hang until the
+    client closes stdin) nor hold sys.stdin's buffer lock (a `_enter_buffered_busy`
+    abort during finalization). `in_place` forces the degraded path that reads sys.stdin.
+    """
+    degrade = dedent(
+        """
+        import mcp.server.stdio
+
+        def failing_dup_above_std(fd):
+            raise OSError("forced degrade")
+
+        mcp.server.stdio._dup_above_std = failing_dup_above_std
+        """
+    )
+    server = (degrade if stdin_claim == "in_place" else "") + dedent(
+        """
+        import anyio
+        from mcp.server.stdio import stdio_server
+
+        async def main():
+            async with stdio_server() as (read_stream, write_stream):
+                await read_stream.receive()
+                await write_stream.aclose()
+
+        anyio.run(main)
+        """
+    )
+    params = StdioServerParameters(command=sys.executable, args=["-c", server])
+
+    with (tmp_path / "server-stderr.txt").open("w+", encoding="utf-8") as errlog:
+        # Allow one cold interpreter start on loaded CI.
+        with anyio.fail_after(10.0):
+            async with stdio_client(params, errlog=errlog) as (read_stream, write_stream):
+                await write_stream.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")))
+                # Ends at the server's stdout EOF, i.e. once its process has exited.
+                async for _ in read_stream:
+                    pass  # pragma: no cover
+        errlog.seek(0)
+        server_stderr = errlog.read()
+
+    assert "Fatal Python error" not in server_stderr
+    assert spawned_processes[0].returncode == 0
+    assert terminate_calls == []