fix(mcp): cancel the in-flight tool call when the caller is cancelled

src/google/adk/cli/browser/assets/config/runtime-config.json

@@ -1,4 +1,4 @@
 {
   "backendUrl": "",
-  "telemetry": false
+  "telemetry": null
 }

src/google/adk/tools/mcp_tool/session_context.py

@@ -38,6 +38,21 @@
 _T = TypeVar('_T')
 
 
+async def _cancel_and_drain(task: asyncio.Future[Any]) -> None:
+  """Cancels an in-flight call and waits for it to actually stop.
+
+  Returning before the call has unwound would let it keep reading the
+  transport after the caller believes it is finished.
+  """
+  task.cancel()
+  try:
+    await task
+  except BaseException:
+    # Includes the CancelledError just requested, and a cancellation
+    # delivered to this frame while the drain was in progress.
+    pass
+
+
 def _read_timeout(seconds: Optional[float]) -> Optional[float | timedelta]:
   """Converts a timeout in seconds to the type ``ClientSession`` expects.
 
@@ -273,10 +288,18 @@ async def _run_guarded(self, coro: Coroutine[Any, Any, _T]) -> _T:
 
     coro_task = asyncio.ensure_future(coro)
 
-    done, _ = await asyncio.wait(
-        [coro_task, self._task],
-        return_when=asyncio.FIRST_COMPLETED,
-    )
+    try:
+      done, _ = await asyncio.wait(
+          [coro_task, self._task],
+          return_when=asyncio.FIRST_COMPLETED,
+      )
+    except BaseException:
+      # asyncio.wait does not own what it waits on, so it leaves the call
+      # running when this frame is cancelled. The caller's `finally` then
+      # releases the session back to the pool while that call is still
+      # reading the transport, and the pool is free to evict it underneath.
+      await _cancel_and_drain(coro_task)
+      raise
 
     if coro_task in done:
       # If the coroutine itself raised, the exception propagates as-is
@@ -287,11 +310,7 @@ async def _run_guarded(self, coro: Coroutine[Any, Any, _T]) -> _T:
 
     # The background task finished first, indicating a transport crash.
     # Cancel the in-flight tool call and surface the original error.
-    coro_task.cancel()
-    try:
-      await coro_task
-    except BaseException:
-      pass
+    await _cancel_and_drain(coro_task)
 
     exc = self._task.exception() if not self._task.cancelled() else None
     raise ConnectionError(

tests/unittests/tools/mcp_tool/test_session_context.py

@@ -900,6 +900,47 @@ async def kill_background_task():
       finally:
         await killer
 
+  @pytest.mark.asyncio
+  async def test_run_guarded_cancels_coro_when_caller_is_cancelled(self):
+    """Cancelling the caller also cancels the in-flight tool call.
+
+    asyncio.wait does not own the futures it waits on, so a cancelled
+    caller would leave the call running against a session that
+    McpTool._run_async_impl has already released back to the pool.
+    """
+    mock_client = MockClient()
+    session_context = SessionContext(
+        mock_client, timeout=5.0, sse_read_timeout=None
+    )
+
+    with patch(
+        'google.adk.tools.mcp_tool.session_context.ClientSession'
+    ) as mock_session_class:
+      mock_session_class.return_value = MockClientSession()
+      await session_context.start()
+
+      coro_started = asyncio.Event()
+      coro_was_cancelled = False
+
+      async def slow_coro():
+        nonlocal coro_was_cancelled
+        coro_started.set()
+        try:
+          await asyncio.sleep(300)
+          return 'should never reach here'
+        except asyncio.CancelledError:
+          coro_was_cancelled = True
+          raise
+
+      caller = asyncio.create_task(session_context._run_guarded(slow_coro()))
+      await coro_started.wait()
+      caller.cancel()
+
+      with pytest.raises(asyncio.CancelledError):
+        await caller
+
+      assert coro_was_cancelled is True
+
 
 class TestSessionContextFlagOffPreservesPreFixBehavior:
   """Pin down that flag=OFF reproduces pre-fix behavior exactly.