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