Python: Fix MCPStreamableHTTPTool leaking asyncio.CancelledError when MCP server is unreachable

python/packages/core/agent_framework/__init__.py

@@ -352,8 +352,8 @@
     "ContinuationToken",
     "ConversationSplit",
     "ConversationSplitter",
-    "Default",
     "DeduplicatingSkillsSource",
+    "Default",
     "DelegatingSkillsSource",
     "Edge",
     "EdgeCondition",

python/packages/core/agent_framework/_mcp.py

@@ -158,6 +158,22 @@ def streamable_http_client(*args: Any, **kwargs: Any) -> _AsyncGeneratorContextM
     return _streamable_http_client(*args, **kwargs)  # type: ignore[return-value]
 
 
+def _should_propagate_cancelled_error(ex: BaseException) -> bool:
+    """Return True if *ex* is a genuine task-cancellation that should propagate unchanged.
+
+    On Python >= 3.11, ``task.cancelling() > 0`` distinguishes a real caller-driven
+    cancellation from a CancelledError raised internally by a library (e.g. via an
+    anyio cancel scope).  On older Python versions the API is unavailable, so we
+    always return False and let callers wrap the error in ToolException instead.
+    """
+    if not isinstance(ex, asyncio.CancelledError):
+        return False
+    if sys.version_info < (3, 11):
+        return False
+    task = asyncio.current_task()
+    return task is not None and task.cancelling() > 0
+
+
 # region: MCP Plugin
 
 
@@ -627,6 +643,17 @@ async def _safe_close_exit_stack(self) -> None:
         except asyncio.CancelledError:
             logger.warning("Could not cleanly close MCP exit stack because the lifecycle owner task was cancelled.")
 
+    async def _close_and_check_cancelled(self, ex: BaseException) -> bool:
+        """Close the exit stack and return True if *ex* is a genuine task cancellation.
+
+        Callers should immediately re-raise when this returns True::
+
+            if await self._close_and_check_cancelled(ex):
+                raise
+        """
+        await self._safe_close_exit_stack()
+        return _should_propagate_cancelled_error(ex)
+
     async def connect(self, *, reset: bool = False) -> None:
         if self._is_lifecycle_owner_task():
             await self._connect_on_owner(reset=reset)
@@ -655,14 +682,23 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None:
         if not self.session:
             try:
                 transport = await self._exit_stack.enter_async_context(self.get_mcp_client())
-            except Exception as ex:
-                await self._safe_close_exit_stack()
+            except (Exception, asyncio.CancelledError) as ex:
+                # On Python >= 3.11, re-raise genuine task cancellation (task.cancelling() > 0)
+                # instead of wrapping it in ToolException. On Python < 3.11, task.cancelling()
+                # is unavailable so MCP-internal CancelledErrors cannot be distinguished from
+                # caller-driven cancellation; they are wrapped as ToolException in that case.
+                if await self._close_and_check_cancelled(ex):
+                    raise
                 command = getattr(self, "command", None)
                 if command:
                     error_msg = f"Failed to start MCP server '{command}': {ex}"
                 else:
                     error_msg = f"Failed to connect to MCP server: {ex}"
-                raise ToolException(error_msg, inner_exception=ex) from ex
+                # CancelledError is a BaseException (not Exception) on Python >= 3.8, so
+                # inner_exception=None and ToolException.__init__ won't log exc_info.
+                if isinstance(ex, asyncio.CancelledError):
+                    logger.debug(error_msg, exc_info=True)
+                raise ToolException(error_msg, inner_exception=ex if isinstance(ex, Exception) else None) from ex
             try:
                 try:
                     from mcp import types
@@ -692,16 +728,21 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None:
                         sampling_capabilities=sampling_capabilities,
                     )
                 )
-            except Exception as ex:
-                await self._safe_close_exit_stack()
+            except (Exception, asyncio.CancelledError) as ex:
+                if await self._close_and_check_cancelled(ex):
+                    raise
+                session_error_msg = f"Failed to create MCP session: {ex}"
+                if isinstance(ex, asyncio.CancelledError):
+                    logger.debug(session_error_msg, exc_info=True)
                 raise ToolException(
-                    message="Failed to create MCP session. Please check your configuration.",
-                    inner_exception=ex,
+                    message=session_error_msg,
+                    inner_exception=ex if isinstance(ex, Exception) else None,
                 ) from ex
             try:
                 await session.initialize()
-            except Exception as ex:
-                await self._safe_close_exit_stack()
+            except (Exception, asyncio.CancelledError) as ex:
+                if await self._close_and_check_cancelled(ex):
+                    raise
                 # Provide context about initialization failure
                 command = getattr(self, "command", None)
                 if command:
@@ -710,7 +751,9 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None:
                     error_msg = f"MCP server '{full_command}' failed to initialize: {ex}"
                 else:
                     error_msg = f"MCP server failed to initialize: {ex}"
-                raise ToolException(error_msg, inner_exception=ex) from ex
+                if isinstance(ex, asyncio.CancelledError):
+                    logger.debug(error_msg, exc_info=True)
+                raise ToolException(error_msg, inner_exception=ex if isinstance(ex, Exception) else None) from ex
             self.session = session
         elif self.session._request_id == 0:  # type: ignore[attr-defined]
             # If the session is not initialized, we need to reinitialize it

python/packages/core/agent_framework/_skills.py

@@ -446,14 +446,10 @@ async def run(self, skill: Skill, args: dict[str, Any] | None = None, **kwargs:
         """
         if not isinstance(skill, FileSkill):
             raise TypeError(
-                f"File-based script '{self.name}' requires a FileSkill "
-                f"but received '{type(skill).__name__}'."
+                f"File-based script '{self.name}' requires a FileSkill but received '{type(skill).__name__}'."
             )
         if self._runner is None:
-            raise ValueError(
-                f"Script '{self.name}' requires a runner. "
-                "Provide a script_runner for file-based scripts."
-            )
+            raise ValueError(f"Script '{self.name}' requires a runner. Provide a script_runner for file-based scripts.")
         result = self._runner(skill, self, args)
         if inspect.isawaitable(result):
             return await result
@@ -570,8 +566,7 @@ def _validate_skill_description(name: str, description: str) -> None:
         raise ValueError("Skill description cannot be empty.")
     if len(description) > MAX_DESCRIPTION_LENGTH:
         raise ValueError(
-            f"Skill '{name}' has an invalid description: "
-            f"Must be {MAX_DESCRIPTION_LENGTH} characters or fewer."
+            f"Skill '{name}' has an invalid description: Must be {MAX_DESCRIPTION_LENGTH} characters or fewer."
         )
 
 
@@ -1993,10 +1988,7 @@ def _get_validated_resource_path(skill_dir: str, resource_name: str) -> str:
             raise ValueError(f"Resource file '{resource_name}' not found in skill directory '{skill_dir}'.")
 
         if FileSkillsSource._has_symlink_in_path(resource_full_path, root_directory_path):
-            raise ValueError(
-                f"Resource file '{resource_name}' "
-                "has a symlink in its path; symlinks are not allowed."
-            )
+            raise ValueError(f"Resource file '{resource_name}' has a symlink in its path; symlinks are not allowed.")
 
         return resource_full_path