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.

Author: sunhouy

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.)

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