fix(client/stdio): let callers clean up multiple transports

independently on asyncio (#577)

`stdio_client` wrapped its reader / writer in
`anyio.create_task_group()` inside the async context manager body. The
task group binds its cancel scope to whichever task entered the
generator, and anyio enforces a strict LIFO stack on those scopes
within a single task. If a caller opened two transports from the same
task and then closed them in the order they were opened (FIFO) — which
is what happens in multi-client orchestrators, dependency-injection
containers and pytest fixtures — teardown crashed with:

    RuntimeError: Attempted to exit a cancel scope that isn't the
    current task's current cancel scope

On the asyncio backend we can spawn the reader / writer with
`asyncio.create_task` instead. The background tasks then live outside
any caller-bound cancel scope, so each transport becomes
self-contained and the cleanup order of two transports no longer
matters. Teardown keeps the documented shutdown sequence (close stdin,
wait for the process with `PROCESS_TERMINATION_TIMEOUT`, escalate to
`_terminate_process_tree`, then close the memory streams) in a shared
`_cleanup_process_and_streams` helper so the two branches can't drift.

On structured-concurrency backends (trio) an anyio task group is still
required — trio intentionally forbids orphan tasks, and cross-task
cleanup is fundamentally incompatible with its model. Those callers
still have to clean up LIFO. The backend dispatch goes through
`sniffio.current_async_library()`.

Regression tests (all asyncio-only):

- `test_stdio_client_supports_fifo_cleanup_on_asyncio` — the exact
  scenario from #577. Verified that it FAILS on the pre-fix code
  with the original RuntimeError, and passes on the fixed code.
- `test_stdio_client_supports_lifo_cleanup_on_asyncio` — historical
  LIFO path must still work unchanged.
- `test_stdio_client_shared_exit_stack_fifo_on_asyncio` — a single
  AsyncExitStack around two transports (already LIFO under the hood,
  but worth pinning so future refactors cannot silently break it).

All 13 stdio tests pass locally (10 pre-existing + 3 new, 1 Unix-only
sigint test skipped on Windows).

Fixes #577

Author: sakenuGOD

src/mcp/client/stdio.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import os
 import sys
@@ -7,6 +8,7 @@
 
 import anyio
 import anyio.lowlevel
+import sniffio
 from anyio.abc import Process
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from anyio.streams.text import TextReceiveStream
@@ -177,38 +179,70 @@ async def stdin_writer():
         except anyio.ClosedResourceError:  # pragma: no cover
             await anyio.lowlevel.checkpoint()
 
-    async with anyio.create_task_group() as tg, process:
-        tg.start_soon(stdout_reader)
-        tg.start_soon(stdin_writer)
+    async def _cleanup_process_and_streams() -> None:
+        # MCP spec: stdio shutdown sequence
+        # 1. Close input stream to server
+        # 2. Wait for server to exit, or send SIGTERM if it doesn't exit in time
+        # 3. Send SIGKILL if still not exited
+        if process.stdin:  # pragma: no branch
+            try:
+                await process.stdin.aclose()
+            except Exception:  # pragma: no cover
+                # stdin might already be closed, which is fine
+                pass
+
         try:
-            yield read_stream, write_stream
-        finally:
-            # MCP spec: stdio shutdown sequence
-            # 1. Close input stream to server
-            # 2. Wait for server to exit, or send SIGTERM if it doesn't exit in time
-            # 3. Send SIGKILL if still not exited
-            if process.stdin:  # pragma: no branch
-                try:
-                    await process.stdin.aclose()
-                except Exception:  # pragma: no cover
-                    # stdin might already be closed, which is fine
-                    pass
+            # Give the process time to exit gracefully after stdin closes
+            with anyio.fail_after(PROCESS_TERMINATION_TIMEOUT):
+                await process.wait()
+        except TimeoutError:
+            # Process didn't exit from stdin closure, use platform-specific termination
+            # which handles SIGTERM -> SIGKILL escalation
+            await _terminate_process_tree(process)
+        except ProcessLookupError:  # pragma: no cover
+            # Process already exited, which is fine
+            pass
+        await read_stream.aclose()
+        await write_stream.aclose()
+        await read_stream_writer.aclose()
+        await write_stream_reader.aclose()
 
+    # On asyncio we spawn the reader / writer with asyncio.create_task rather
+    # than an anyio task group, so their cancel scopes are not bound to the
+    # caller's task. That is what lets callers clean up multiple transports
+    # in arbitrary order — see #577. On structured-concurrency backends
+    # (trio), we keep the task group: orphan tasks are disallowed there by
+    # design, and cross-task cleanup is fundamentally incompatible with
+    # that model, so callers on trio still have to clean up LIFO.
+    if sniffio.current_async_library() == "asyncio":
+        async with process:
+            stdout_task = asyncio.create_task(stdout_reader())
+            stdin_task = asyncio.create_task(stdin_writer())
             try:
-                # Give the process time to exit gracefully after stdin closes
-                with anyio.fail_after(PROCESS_TERMINATION_TIMEOUT):
-                    await process.wait()
-            except TimeoutError:
-                # Process didn't exit from stdin closure, use platform-specific termination
-                # which handles SIGTERM -> SIGKILL escalation
-                await _terminate_process_tree(process)
-            except ProcessLookupError:  # pragma: no cover
-                # Process already exited, which is fine
-                pass
-            await read_stream.aclose()
-            await write_stream.aclose()
-            await read_stream_writer.aclose()
-            await write_stream_reader.aclose()
+                yield read_stream, write_stream
+            finally:
+                try:
+                    await _cleanup_process_and_streams()
+                finally:
+                    for task in (stdout_task, stdin_task):
+                        if not task.done():
+                            task.cancel()
+                    for task in (stdout_task, stdin_task):
+                        try:
+                            await task
+                        except BaseException:  # noqa: BLE001, S110
+                            # Reader/writer cancellation or late I/O errors
+                            # surfaced after the streams were closed: those
+                            # are expected during teardown.
+                            pass
+    else:
+        async with anyio.create_task_group() as tg, process:
+            tg.start_soon(stdout_reader)
+            tg.start_soon(stdin_writer)
+            try:
+                yield read_stream, write_stream
+            finally:
+                await _cleanup_process_and_streams()
 
 
 def _get_executable_command(command: str) -> str:

tests/client/test_stdio.py

@@ -558,3 +558,92 @@ def sigterm_handler(signum, frame):
         f"stdio_client cleanup took {elapsed:.1f} seconds for stdin-ignoring process. "
         f"Expected between 2-4 seconds (2s stdin timeout + termination time)."
     )
+
+
+# A stub MCP-ish server: exits cleanly as soon as stdin closes. We only need
+# the stdio_client to be able to stand the transport up; we do not exercise
+# any MCP protocol traffic in the FIFO-cleanup tests below.
+_QUIET_STDIN_STUB = textwrap.dedent(
+    """
+    import sys
+    for _ in sys.stdin:
+        pass
+    """
+)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("anyio_backend", ["asyncio"])
+async def test_stdio_client_supports_fifo_cleanup_on_asyncio(anyio_backend):
+    """
+    Regression for https://github.com/modelcontextprotocol/python-sdk/issues/577.
+
+    Prior to the fix, closing two ``stdio_client`` transports in the order
+    they were opened (FIFO) crashed with::
+
+        RuntimeError: Attempted to exit a cancel scope that isn't the
+        current task's current cancel scope
+
+    because each stdio_client bound a task-group cancel scope to the
+    caller's task, and anyio enforces a strict LIFO stack on those
+    scopes. On asyncio the fix uses ``asyncio.create_task`` for the
+    reader / writer so the transports are independent and the cleanup
+    order no longer matters.
+
+    (Trio intentionally forbids orphan tasks — there is no equivalent
+    fix on that backend, so this test is asyncio-only.)
+    """
+    params = StdioServerParameters(command=sys.executable, args=["-c", _QUIET_STDIN_STUB])
+
+    s1 = AsyncExitStack()
+    s2 = AsyncExitStack()
+
+    await s1.__aenter__()
+    await s2.__aenter__()
+
+    await s1.enter_async_context(stdio_client(params))
+    await s2.enter_async_context(stdio_client(params))
+
+    # Close in FIFO order — the opposite of what anyio's structured
+    # concurrency would normally require.
+    await s1.aclose()
+    await s2.aclose()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("anyio_backend", ["asyncio"])
+async def test_stdio_client_supports_lifo_cleanup_on_asyncio(anyio_backend):
+    """
+    Sanity check for the fix above: the historical LIFO cleanup path
+    must still work unchanged on asyncio.
+    """
+    params = StdioServerParameters(command=sys.executable, args=["-c", _QUIET_STDIN_STUB])
+
+    s1 = AsyncExitStack()
+    s2 = AsyncExitStack()
+    await s1.__aenter__()
+    await s2.__aenter__()
+
+    await s1.enter_async_context(stdio_client(params))
+    await s2.enter_async_context(stdio_client(params))
+
+    # LIFO cleanup — the last-opened transport closes first.
+    await s2.aclose()
+    await s1.aclose()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("anyio_backend", ["asyncio"])
+async def test_stdio_client_shared_exit_stack_fifo_on_asyncio(anyio_backend):
+    """
+    The same story, but through a single ``AsyncExitStack`` that the
+    caller then closes once. ExitStack runs callbacks in LIFO order on
+    its own, so this case already worked — the test pins that behavior
+    so a future refactor of the asyncio branch cannot silently break
+    it.
+    """
+    params = StdioServerParameters(command=sys.executable, args=["-c", _QUIET_STDIN_STUB])
+
+    async with AsyncExitStack() as stack:
+        await stack.enter_async_context(stdio_client(params))
+        await stack.enter_async_context(stdio_client(params))