fix: detect MCP session disconnect during tool call to avoid 5min hang

When the MCP server returns a non-2xx HTTP response (e.g. 403), the
transport crashes in a background task without propagating the error to
the pending send_request() call. This causes the agent to hang for ~5
minutes until sse_read_timeout expires.

Add a concurrent health check that polls session stream state alongside
tool calls. If the session disconnects mid-call, raise ConnectionError
immediately instead of waiting for the timeout.

Fixes #4901

Author: Ratnaditya-J

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

@@ -373,8 +373,8 @@ async def _run_async_impl(
     # Resolve progress callback (may be a factory that needs runtime context)
     resolved_callback = self._resolve_progress_callback(tool_context)
 
-    response = await session.call_tool(
-        self._mcp_tool.name,
+    response = await self._call_tool_with_health_check(
+        session=session,
         arguments=args,
         progress_callback=resolved_callback,
         meta=meta_trace_context,
@@ -396,6 +396,93 @@ async def _run_async_impl(
       )
     return result
 
+  async def _call_tool_with_health_check(
+      self,
+      *,
+      session,
+      arguments: dict[str, Any],
+      progress_callback,
+      meta,
+  ) -> Any:
+    """Calls an MCP tool while monitoring session health.
+
+    When the MCP server returns a non-2xx HTTP response, the underlying
+    transport can crash in a background task without propagating the error
+    to the pending send_request() call. This causes send_request() to hang
+    until the read timeout (~5 minutes) expires.
+
+    This method races the tool call against a health check that polls the
+    session's stream state. If the session disconnects mid-call, we fail
+    fast with a ConnectionError instead of hanging.
+
+    Args:
+        session: The MCP ClientSession to use.
+        arguments: The arguments to pass to the tool.
+        progress_callback: Optional progress callback.
+        meta: Optional trace context metadata.
+
+    Returns:
+        The tool call response.
+
+    Raises:
+        ConnectionError: If the session disconnects during the tool call.
+    """
+
+    async def _health_check():
+      """Polls session stream health and raises if disconnected."""
+      # Small initial delay to let the call_tool start
+      await asyncio.sleep(0.1)
+      while True:
+        try:
+          if self._mcp_session_manager._is_session_disconnected(session):
+            raise ConnectionError(
+                'MCP session disconnected during tool call. This typically'
+                ' happens when the MCP server returns a non-2xx HTTP'
+                ' response (e.g. 403 Forbidden). Check the server URL'
+                ' and authentication configuration.'
+            )
+        except (AttributeError, TypeError):
+          # If stream attributes are not accessible, skip the check.
+          # This can happen with certain transport implementations.
+          pass
+        await asyncio.sleep(0.5)
+
+    health_task = asyncio.create_task(_health_check())
+    tool_task = asyncio.create_task(
+        session.call_tool(
+            self._mcp_tool.name,
+            arguments=arguments,
+            progress_callback=progress_callback,
+            meta=meta,
+        )
+    )
+
+    try:
+      done, pending = await asyncio.wait(
+          [tool_task, health_task],
+          return_when=asyncio.FIRST_COMPLETED,
+      )
+
+      # Cancel whichever task didn't finish
+      for task in pending:
+        task.cancel()
+        try:
+          await task
+        except (asyncio.CancelledError, Exception):
+          pass
+
+      # If the tool call finished first, return its result
+      if tool_task in done:
+        return tool_task.result()
+
+      # Health check finished first (i.e., session disconnected)
+      # Re-raise the ConnectionError from the health check
+      health_task.result()
+    except asyncio.CancelledError:
+      tool_task.cancel()
+      health_task.cancel()
+      raise
+
   def _resolve_progress_callback(
       self, tool_context: ToolContext
   ) -> Optional[ProgressFnT]:

tests/unittests/tools/mcp_tool/test_mcp_tool.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import inspect
 from unittest.mock import AsyncMock
 from unittest.mock import Mock
@@ -1127,3 +1128,87 @@ def test_mcp_app_resource_uri_property_none(self):
         mcp_session_manager=self.mock_session_manager,
     )
     assert tool2.mcp_app_resource_uri is None
+
+  @pytest.mark.asyncio
+  async def test_call_tool_with_health_check_disconnected_session(self):
+    """Test that a disconnected session is detected quickly during tool call.
+
+    When the MCP server returns a non-2xx HTTP response, the transport can
+    crash in a background task without propagating the error to the pending
+    send_request(). This test verifies that the health check detects the
+    closed streams and raises a ConnectionError instead of hanging.
+
+    Regression test for https://github.com/google/adk-python/issues/4901
+    """
+    tool = MCPTool(
+        mcp_tool=self.mock_mcp_tool,
+        mcp_session_manager=self.mock_session_manager,
+    )
+
+    # Create a mock session with streams that start open then close
+    mock_session = AsyncMock()
+    mock_read_stream = Mock()
+    mock_write_stream = Mock()
+    mock_read_stream._closed = False
+    mock_write_stream._closed = False
+    mock_session._read_stream = mock_read_stream
+    mock_session._write_stream = mock_write_stream
+
+    # Make call_tool hang indefinitely (simulating the bug)
+    hang_event = asyncio.Event()
+
+    async def hanging_call_tool(*args, **kwargs):
+      await hang_event.wait()
+
+    mock_session.call_tool = hanging_call_tool
+
+    # After a short delay, simulate the transport crashing by closing streams
+    async def simulate_transport_crash():
+      await asyncio.sleep(0.2)
+      mock_read_stream._closed = True
+
+    crash_task = asyncio.create_task(simulate_transport_crash())
+
+    with pytest.raises(ConnectionError, match='MCP session disconnected'):
+      await tool._call_tool_with_health_check(
+          session=mock_session,
+          arguments={"param1": "test"},
+          progress_callback=None,
+          meta=None,
+      )
+
+    crash_task.cancel()
+    try:
+      await crash_task
+    except asyncio.CancelledError:
+      pass
+
+  @pytest.mark.asyncio
+  async def test_call_tool_with_health_check_success(self):
+    """Test that healthy sessions return tool results normally."""
+    tool = MCPTool(
+        mcp_tool=self.mock_mcp_tool,
+        mcp_session_manager=self.mock_session_manager,
+    )
+
+    mock_session = AsyncMock()
+    mock_read_stream = Mock()
+    mock_write_stream = Mock()
+    mock_read_stream._closed = False
+    mock_write_stream._closed = False
+    mock_session._read_stream = mock_read_stream
+    mock_session._write_stream = mock_write_stream
+
+    mcp_response = CallToolResult(
+        content=[TextContent(type="text", text="success")]
+    )
+    mock_session.call_tool = AsyncMock(return_value=mcp_response)
+
+    result = await tool._call_tool_with_health_check(
+        session=mock_session,
+        arguments={"param1": "test"},
+        progress_callback=None,
+        meta=None,
+    )
+
+    assert result == mcp_response