Use asyncio.gather for exception-safe transport cleanup

Per CodeRabbit review feedback: the sequential for-loop could leave remaining transports unclosed if an earlier close() raises an exception.

Using asyncio.gather with return_exceptions=True ensures all transports close regardless of individual failures.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

src/fastmcp/client/transports.py

@@ -974,12 +974,23 @@ class MCPConfigTransport(ClientTransport):
     async def connect_session(
         self, **session_kwargs: Unpack[SessionKwargs]
     ) -> AsyncIterator[ClientSession]:
-        async with self.transport.connect_session(**session_kwargs) as session:
-            yield session
+        try:
+            async with self.transport.connect_session(**session_kwargs) as session:
+                yield session
+        finally:
+            # Clean up underlying transports to ensure subprocesses terminate
+            # Use gather with return_exceptions to ensure all transports close even if one fails
+            await asyncio.gather(
+                *(transport.close() for transport in self._underlying_transports),
+                return_exceptions=True,
+            )
 
     async def close(self):
-        for transport in self._underlying_transports:
-            await transport.close()
+        # Use gather with return_exceptions to ensure all transports close even if one fails
+        await asyncio.gather(
+            *(transport.close() for transport in self._underlying_transports),
+            return_exceptions=True,
+        )
 
     def __repr__(self) -> str:
         return f"<MCPConfigTransport(config='{self.config}')>"

tests/test_mcp_config.py

@@ -340,17 +340,12 @@ async def test_multi_client_lifespan(tmp_path: Path):
 
     gc_collect_harder()
 
-    # This test will fail while debugging because the debugger holds a reference to the underlying transport
+    # Verify processes have terminated
+    with pytest.raises(psutil.NoSuchProcess):
+        psutil.Process(pid_1).status()
 
     with pytest.raises(psutil.NoSuchProcess):
-        while True:
-            psutil.Process(pid_1)
-            await asyncio.sleep(0.01)
-
-    with pytest.raises(psutil.NoSuchProcess):
-        while True:
-            psutil.Process(pid_2)
-            await asyncio.sleep(0.01)
+        psutil.Process(pid_2).status()
 
 
 @pytest.mark.skipif(