From d97d137099013555ca0c93eb79cec7e677764529 Mon Sep 17 00:00:00 2001 From: sunhouy Date: Thu, 5 Nov 2026 11:12:47 +0800 Subject: [PATCH] fix(client): end the read stream when stdout hits EOF on a live server `stdout_reader` closes the read stream inside `async with read_stream_writer`, but its `finally` runs `_drain_stdout` first, and before this change that drain sat in a shielded `while True: await process.stdout.receive()`. When the server closes stdout without exiting, the drain waits on the still-open pipes of a live process that will never write again, so the writer is never closed, the session never sees the end of the stream, and every waiting request waits forever. A crashed wrapper or a closed fd reproduces it without any mistake on the caller's side. Terminate the process before draining unless shutdown already asked for it. Shutdown keeps its own stop sequence, and the drain keeps its reason to exist (a server flushing its last output must not block on a full pipe), it just runs against a process that is now closing its pipes. That also makes the read side symmetric with `stdin_writer`, which already closes the read stream on a failed write so the session ends instead of hanging. --- src/mcp/client/stdio.py | 10 +++++++--- tests/client/test_stdio.py | 27 ++++++++++++++++++++++++++- 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/src/mcp/client/stdio.py b/src/mcp/client/stdio.py index 6a3ad12111..c9ca4238a9 100644 --- a/src/mcp/client/stdio.py +++ b/src/mcp/client/stdio.py @@ -136,7 +136,7 @@ async def stdio_client( read_stream_writer, read_stream = anyio.create_memory_object_stream[SessionMessage | Exception](0) write_stream, write_stream_reader = anyio.create_memory_object_stream[SessionMessage](0) - shutting_down = False + shutting_down_event = anyio.Event() writer_done = anyio.Event() async def stdout_reader() -> None: @@ -157,13 +157,17 @@ async def stdout_reader() -> None: except (anyio.ClosedResourceError, anyio.BrokenResourceError): return # the session is gone; only the drain below remains finally: + # The server is usually gone by now; if it is not, its pipes stay + # open and the drain below would wait on it forever. + if not shutting_down_event.is_set(): + await _terminate_process_tree(process) await _drain_stdout(process) except anyio.ClosedResourceError: pass # our own shutdown closed the stdout stream under the read except (anyio.BrokenResourceError, ConnectionError): # Teardown noise during shutdown, a real failure otherwise; either way # the session sees clean closure when the read stream closes. - if not shutting_down: + if not shutting_down_event.is_set(): logger.exception("Reading from the MCP server's stdout failed mid-session") async def stdin_writer() -> None: @@ -204,7 +208,7 @@ async def shutdown() -> None: try: yield read_stream, write_stream finally: - shutting_down = True + shutting_down_event.set() # Shutdown must finish even under caller cancellation, or the server # process would leak; every wait inside is bounded. (Native # task.cancel() and the fallback's worker threads can still defeat it.) diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py index b64eb01902..97749d9c4c 100644 --- a/tests/client/test_stdio.py +++ b/tests/client/test_stdio.py @@ -323,6 +323,29 @@ async def test_invalid_json_from_the_server_surfaces_as_an_in_stream_exception( assert await _next_message(read_stream) == ping +@pytest.mark.anyio +async def test_stdout_eof_while_the_server_keeps_running_ends_the_read_stream( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """EOF on stdout closes the session even though the process never exited. + + A crashed wrapper or a closed fd leaves the server alive with its stdin and stderr + still open, so nothing else ends the connection; the read stream must close before + the reader parks in the drain, which waits on a live server's pipes forever. + """ + process = FakeProcess() + terminated = install_fake_process(monkeypatch, process) + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS) as (read_stream, _): + process.close_stdout() + with pytest.raises(anyio.EndOfStream): + await read_stream.receive() + + # The server is still around, so it is killed instead of left behind. + assert terminated == [process] + + @pytest.mark.anyio async def test_a_server_that_dies_before_responding_fails_initialize_with_connection_closed( monkeypatch: pytest.MonkeyPatch, @@ -723,6 +746,8 @@ async def test_a_failed_write_to_a_live_server_closes_the_read_stream_instead_of 1. A send fails with EIO while the server is alive; the read stream ends. 2. Output the server produces afterwards is still drained, so it cannot wedge on a full pipe. + 3. The server outlived the session and gets the same termination shutdown would + have given it. """ ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping") pong = JSONRPCResponse(jsonrpc="2.0", id=1, result={}) @@ -744,7 +769,7 @@ async def test_a_failed_write_to_a_live_server_closes_the_read_stream_instead_of assert process.pending_stdout_chunks() == 0 assert process.written == [] - assert terminated == [] + assert terminated == [process] @pytest.mark.anyio