fix(stdio): resolve BrokenResourceError race condition on quick exit

ROOT CAUSE:
The stdio_client async context manager had a race condition when exiting
quickly (before subprocess finished outputting data). The cleanup code in
the finally block closed memory streams while background tasks (stdout_reader
and stdin_writer) were still using them, resulting in BrokenResourceError.

Timeline of the bug:
1. User code exits the async with stdio_client(...) context
2. The finally block executes
3. Streams are closed immediately
4. Background tasks are still running and trying to send/receive data
5. Tasks encounter closed streams → BrokenResourceError

CHANGES:
1. Added BrokenResourceError to exception handlers in both client and server
   - src/mcp/client/stdio.py: Lines 161, 177 (stdout_reader, stdin_writer)
   - src/mcp/server/stdio.py: Lines 67, 77 (stdin_reader, stdout_writer)
   - This allows tasks to exit gracefully if streams close during operation

2. Added task cancellation before stream closure in client transport
   - src/mcp/client/stdio.py: Line 210 (after process cleanup)
   - tg.cancel_scope.cancel() sends cancellation signal to background tasks
   - Tasks receive signal and finish their current operation
   - Then streams are closed (tasks aren't using them anymore)

3. Added regression test
   - tests/client/test_stdio.py: test_stdio_client_quick_exit_race_condition
   - Verifies that quick context exits don't cause ExceptionGroup crashes

IMPACT:
- No more ExceptionGroup crashes when exiting quickly
- Graceful task shutdown with proper cancellation
- Backward compatible - all existing tests pass
- Better resource cleanup - tasks finish before streams close

TECHNICAL NOTES:
- Server transport only needed exception handler changes (not task cancellation)
  because it doesn't manage subprocess lifecycle
- The fix uses defense-in-depth: both proper coordination AND graceful handling
- anyio.BrokenResourceError is raised when operations are attempted on closed
  resources, distinct from ClosedResourceError (resource already closed)

FILES MODIFIED:
- src/mcp/client/stdio.py
- src/mcp/server/stdio.py
- tests/client/test_stdio.py

Author: peter-luminova

src/mcp/client/stdio.py

@@ -158,7 +158,7 @@ async def stdout_reader():
 
                         session_message = SessionMessage(message)
                         await read_stream_writer.send(session_message)
-        except anyio.ClosedResourceError:  # pragma: lax no cover
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError):  # pragma: lax no cover
             await anyio.lowlevel.checkpoint()
 
     async def stdin_writer():
@@ -174,7 +174,7 @@ async def stdin_writer():
                             errors=server.encoding_error_handler,
                         )
                     )
-        except anyio.ClosedResourceError:  # pragma: no cover
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError):  # pragma: no cover
             await anyio.lowlevel.checkpoint()
 
     async with anyio.create_task_group() as tg, process:
@@ -205,6 +205,11 @@ async def stdin_writer():
             except ProcessLookupError:  # pragma: no cover
                 # Process already exited, which is fine
                 pass
+
+            # Cancel background tasks before closing streams to prevent race condition
+            # where tasks try to use closed streams (BrokenResourceError)
+            tg.cancel_scope.cancel()
+
             await read_stream.aclose()
             await write_stream.aclose()
             await read_stream_writer.aclose()

src/mcp/server/stdio.py

@@ -64,7 +64,7 @@ async def stdin_reader():
 
                     session_message = SessionMessage(message)
                     await read_stream_writer.send(session_message)
-        except anyio.ClosedResourceError:  # pragma: no cover
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError):  # pragma: no cover
             await anyio.lowlevel.checkpoint()
 
     async def stdout_writer():
@@ -74,7 +74,7 @@ async def stdout_writer():
                     json = session_message.message.model_dump_json(by_alias=True, exclude_none=True)
                     await stdout.write(json + "\n")
                     await stdout.flush()
-        except anyio.ClosedResourceError:  # pragma: no cover
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError):  # pragma: no cover
             await anyio.lowlevel.checkpoint()
 
     async with anyio.create_task_group() as tg:

tests/client/test_stdio.py

@@ -620,3 +620,50 @@ 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)."
     )
+
+
+@pytest.mark.anyio
+async def test_stdio_client_quick_exit_race_condition():
+    """Test that stdio_client handles quick context exits without crashing.
+
+    This reproduces the race condition where:
+    1. Subprocess is spawned and starts outputting data
+    2. User code exits the context quickly (e.g., timeout, error, disconnect)
+    3. Cleanup code closes streams while background tasks are still using them
+    4. Background tasks should handle closed streams gracefully (no BrokenResourceError)
+
+    The fix ensures:
+    - Tasks are cancelled before streams are closed
+    - Tasks handle BrokenResourceError gracefully as a fallback
+    """
+
+    # Create a Python script that continuously outputs data
+    # This simulates a subprocess that's slow to shut down
+    continuous_output_script = textwrap.dedent(
+        """
+        import sys
+        import time
+
+        # Continuously output to keep stdout_reader busy
+        for i in range(100):
+            print(f'{{"jsonrpc":"2.0","id":{i},"result":{{}}}}')
+            sys.stdout.flush()
+            time.sleep(0.01)
+        """
+    )
+
+    server_params = StdioServerParameters(
+        command=sys.executable,
+        args=["-c", continuous_output_script],
+    )
+
+    # This should not raise an ExceptionGroup or BrokenResourceError
+    # The background tasks should handle stream closure gracefully
+    async with stdio_client(server_params) as (read_stream, write_stream):
+        # Immediately exit - triggers cleanup while subprocess is still outputting
+        pass
+
+    # If we get here without exception, the race condition is handled correctly
+    # The tasks either:
+    # 1. Were cancelled before stream closure (proper fix)
+    # 2. Handled BrokenResourceError gracefully (defense in depth)