fix(tools): Prevent AnyIO CancelScope task boundary violations during MCP session creation failure

Co-authored-by: Sasha Sobran <asobran@google.com>
PiperOrigin-RevId: 916220631

Author: sasha-gitg

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

@@ -542,10 +542,13 @@ async def create_session(
             sampling_capabilities=self._sampling_capabilities,
         )
 
-        session = await asyncio.wait_for(
-            exit_stack.enter_async_context(session_context),
-            timeout=timeout_in_seconds,
-        )
+        if is_feature_enabled(FeatureName._MCP_GRACEFUL_ERROR_HANDLING):  # pylint: disable=protected-access
+          session = await exit_stack.enter_async_context(session_context)
+        else:
+          session = await asyncio.wait_for(
+              exit_stack.enter_async_context(session_context),
+              timeout=timeout_in_seconds,
+          )
 
         # Store session, exit stack, and loop in the pool. The pool storage
         # remains a tuple for backward-compatibility with downstream tests

tests/unittests/tools/mcp_tool/test_mcp_session_manager.py

@@ -1043,3 +1043,35 @@ def test_env_var_disable_acts_as_kill_switch(self):
         os.environ[disable] = saved_disable
       if saved_enable is not None:
         os.environ[enable] = saved_enable
+
+  @pytest.mark.asyncio
+  @patch("google.adk.tools.mcp_tool.mcp_session_manager.asyncio.wait_for")
+  async def test_create_session_does_not_use_wait_for_when_ge_is_enabled(
+      self, mock_wait_for
+  ):
+    """create_session must not wrap enter_async_context in asyncio.wait_for when GE is enabled."""
+    from google.adk.features import FeatureName
+    from google.adk.features._feature_registry import temporary_feature_override
+
+    manager = MCPSessionManager(
+        StdioConnectionParams(
+            server_params=StdioServerParameters(command="dummy", args=[]),
+            timeout=5.0,
+        )
+    )
+    with temporary_feature_override(
+        FeatureName._MCP_GRACEFUL_ERROR_HANDLING, True
+    ):
+      with patch(
+          "google.adk.tools.mcp_tool.mcp_session_manager.AsyncExitStack"
+      ) as mock_stack:
+        mock_stack.return_value.enter_async_context = AsyncMock()
+        with patch(
+            "google.adk.tools.mcp_tool.mcp_session_manager.SessionContext"
+        ):
+          with patch(
+              "google.adk.tools.mcp_tool.mcp_session_manager.stdio_client"
+          ):
+            await manager.create_session()
+
+    mock_wait_for.assert_not_called()