From a3b7527ef0e78abdc7a878a5fb30bac2ad2dae0c Mon Sep 17 00:00:00 2001 From: Copilot Date: Thu, 7 May 2026 04:41:46 +0000 Subject: [PATCH 1/7] fix: wrap asyncio.CancelledError in ToolException in _connect_on_owner (#5667) asyncio.CancelledError is a BaseException (not Exception) in Python 3.8+. When an MCP server is unreachable, the MCP library's internal anyio task group raises CancelledError, which escaped all three 'except Exception' handlers in _connect_on_owner(). This propagated through _run_lifecycle_owner -> _run_on_lifecycle_owner -> connect -> __aenter__, bypassing user except Exception blocks entirely. Fix: change the three except-Exception clauses in _connect_on_owner to 'except (Exception, asyncio.CancelledError)' so spurious CancelledErrors from the MCP transport layer are caught and wrapped in ToolException, consistent with the method's documented contract. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../packages/core/agent_framework/__init__.py | 2 +- python/packages/core/agent_framework/_mcp.py | 12 +-- .../packages/core/agent_framework/_skills.py | 16 +--- python/packages/core/tests/core/test_mcp.py | 87 +++++++++++++++++++ .../packages/core/tests/core/test_skills.py | 24 +++-- 5 files changed, 113 insertions(+), 28 deletions(-) diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py index eb439c35435..4592f8c7169 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/python/packages/core/agent_framework/__init__.py @@ -352,8 +352,8 @@ "ContinuationToken", "ConversationSplit", "ConversationSplitter", - "Default", "DeduplicatingSkillsSource", + "Default", "DelegatingSkillsSource", "Edge", "EdgeCondition", diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py index 9dfb29932f9..5da1ac23d72 100644 --- a/python/packages/core/agent_framework/_mcp.py +++ b/python/packages/core/agent_framework/_mcp.py @@ -655,14 +655,14 @@ 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: + except (Exception, asyncio.CancelledError) as ex: await self._safe_close_exit_stack() 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 + raise ToolException(error_msg, inner_exception=ex if isinstance(ex, Exception) else None) from ex try: try: from mcp import types @@ -692,15 +692,15 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: sampling_capabilities=sampling_capabilities, ) ) - except Exception as ex: + except (Exception, asyncio.CancelledError) as ex: await self._safe_close_exit_stack() raise ToolException( message="Failed to create MCP session. Please check your configuration.", - inner_exception=ex, + inner_exception=ex if isinstance(ex, Exception) else None, ) from ex try: await session.initialize() - except Exception as ex: + except (Exception, asyncio.CancelledError) as ex: await self._safe_close_exit_stack() # Provide context about initialization failure command = getattr(self, "command", None) @@ -710,7 +710,7 @@ 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 + 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 diff --git a/python/packages/core/agent_framework/_skills.py b/python/packages/core/agent_framework/_skills.py index 082c6f1b699..06612b4df0a 100644 --- a/python/packages/core/agent_framework/_skills.py +++ b/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 diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 01cf1717bdd..2b56281d1ea 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -1,5 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. # type: ignore[reportPrivateUsage] +import asyncio import json import logging import os @@ -2264,6 +2265,92 @@ async def test_connect_cleanup_on_initialization_failure(): tool._exit_stack.aclose.assert_called_once() + +async def test_connect_cancelled_error_during_transport_creation_raises_tool_exception(): + """Test that CancelledError from transport creation is wrapped in ToolException.""" + tool = MCPStreamableHTTPTool(name="test", url="http://example.com") + tool.get_mcp_client = Mock(side_effect=asyncio.CancelledError("cancel scope")) + + with pytest.raises(ToolException, match="Failed to connect to MCP server"): + await tool.connect() + + +async def test_connect_cancelled_error_during_session_creation_raises_tool_exception(): + """Test that CancelledError from session creation is wrapped in ToolException.""" + tool = MCPStreamableHTTPTool(name="test", url="http://example.com") + + mock_transport = (Mock(), Mock()) + mock_context_manager = Mock() + mock_context_manager.__aenter__ = AsyncMock(return_value=mock_transport) + mock_context_manager.__aexit__ = AsyncMock(return_value=None) + tool.get_mcp_client = Mock(return_value=mock_context_manager) + + with patch("mcp.client.session.ClientSession") as mock_session_class: + mock_session_class.return_value.__aenter__ = AsyncMock(side_effect=asyncio.CancelledError("cancel scope")) + mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) + + with pytest.raises(ToolException, match="Failed to create MCP session"): + await tool.connect() + + +async def test_connect_cancelled_error_during_initialize_raises_tool_exception(): + """Test that CancelledError from session.initialize() is wrapped in ToolException. + + This is the primary regression test for the bug: when an MCP server is unreachable, + the MCP library raises asyncio.CancelledError internally, which previously escaped + all except Exception handlers and could not be caught by user code. + """ + tool = MCPStreamableHTTPTool(name="test", url="http://example.com") + + mock_transport = (Mock(), Mock()) + mock_context_manager = Mock() + mock_context_manager.__aenter__ = AsyncMock(return_value=mock_transport) + mock_context_manager.__aexit__ = AsyncMock(return_value=None) + tool.get_mcp_client = Mock(return_value=mock_context_manager) + + mock_session = Mock() + mock_session.initialize = AsyncMock(side_effect=asyncio.CancelledError("Cancelled via cancel scope")) + + with patch("mcp.client.session.ClientSession") as mock_session_class: + mock_session_class.return_value.__aenter__ = AsyncMock(return_value=mock_session) + mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) + + with pytest.raises(ToolException, match="MCP server failed to initialize"): + await tool.connect() + + +async def test_aenter_cancelled_error_during_connect_is_catchable_as_exception(): + """Test that CancelledError during __aenter__ is catchable as Exception. + + Verifies the end-to-end fix: async with MCPStreamableHTTPTool(...) raises an + exception that can be caught by a normal `except Exception` block. + """ + tool = MCPStreamableHTTPTool(name="test", url="http://example.com") + + mock_session = Mock() + mock_session.initialize = AsyncMock(side_effect=asyncio.CancelledError("Cancelled via cancel scope")) + + mock_transport = (Mock(), Mock()) + mock_context_manager = Mock() + mock_context_manager.__aenter__ = AsyncMock(return_value=mock_transport) + mock_context_manager.__aexit__ = AsyncMock(return_value=None) + tool.get_mcp_client = Mock(return_value=mock_context_manager) + + with patch("mcp.client.session.ClientSession") as mock_session_class: + mock_session_class.return_value.__aenter__ = AsyncMock(return_value=mock_session) + mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) + + caught = None + try: + async with tool: + pass + except Exception as e: + caught = e + + assert caught is not None, "Expected an exception to be caught by except Exception" + assert isinstance(caught, ToolException) + + def test_mcp_stdio_tool_get_mcp_client_with_env_and_kwargs(): """Test MCPStdioTool.get_mcp_client() with environment variables and client kwargs.""" env_vars = {"PATH": "/usr/bin", "DEBUG": "1"} diff --git a/python/packages/core/tests/core/test_skills.py b/python/packages/core/tests/core/test_skills.py index 36906d55b86..8e8c6a8aedf 100644 --- a/python/packages/core/tests/core/test_skills.py +++ b/python/packages/core/tests/core/test_skills.py @@ -1190,7 +1190,9 @@ async def get_user_data(**kwargs: Any) -> Any: provider = SkillsProvider([skill]) await _init_provider(provider) - result = await provider._read_skill_resource(_raw_skills(provider), "prog-skill", "get_user_data", auth_token="abc") + result = await provider._read_skill_resource( + _raw_skills(provider), "prog-skill", "get_user_data", auth_token="abc" + ) assert result == "data with token=abc" async def test_read_callable_resource_without_kwargs_ignores_extra_args(self) -> None: @@ -2059,6 +2061,7 @@ async def test_read_sync_function(self) -> None: async def test_read_async_function(self) -> None: """read() awaits an async function and returns its result.""" + async def get_data() -> str: return "async result" @@ -2068,6 +2071,7 @@ async def get_data() -> str: async def test_read_function_with_kwargs(self) -> None: """read() forwards kwargs to functions that accept them.""" + def get_config(**kwargs: Any) -> str: return f"user={kwargs.get('user_id')}" @@ -2077,6 +2081,7 @@ def get_config(**kwargs: Any) -> str: async def test_read_async_function_with_kwargs(self) -> None: """read() forwards kwargs to async functions that accept them.""" + async def get_config(**kwargs: Any) -> str: return f"user={kwargs.get('user_id')}" @@ -2086,6 +2091,7 @@ async def get_config(**kwargs: Any) -> str: async def test_read_function_without_kwargs_ignores_extra(self) -> None: """read() does not pass kwargs to functions that don't accept them.""" + def simple() -> str: return "fixed" @@ -2095,6 +2101,7 @@ def simple() -> str: async def test_read_function_raises_propagates(self) -> None: """read() propagates exceptions from the function.""" + def failing() -> str: raise RuntimeError("boom") @@ -2747,6 +2754,7 @@ async def async_func(x: int = 0) -> str: async def test_code_script_returns_object(self) -> None: """Code-defined scripts can return non-string objects.""" + def returns_dict() -> dict: return {"status": "ok", "value": 42} @@ -2855,8 +2863,8 @@ def process(**kwargs: Any) -> str: provider = SkillsProvider([skill]) await _init_provider(provider) - result = await provider._run_skill_script(_raw_skills(provider), - "my-skill", "process", args={"mode": "llm-value"}, mode="runtime-value" + result = await provider._run_skill_script( + _raw_skills(provider), "my-skill", "process", args={"mode": "llm-value"}, mode="runtime-value" ) assert "Error" in result @@ -2946,6 +2954,7 @@ async def test_require_script_approval_does_not_affect_other_tools(self) -> None async def test_code_script_exception_returns_error(self) -> None: """A code script function that raises should return an error string.""" + def failing_script() -> str: raise RuntimeError("Something went wrong") @@ -3170,6 +3179,7 @@ async def test_code_skill_no_scripts_element(self) -> None: async def test_code_skill_scripts_element_contains_parameters(self) -> None: """Scripts XML includes parameters schema when the function has typed parameters.""" + def analyze(query: str, limit: int = 10) -> str: return "result" @@ -3755,9 +3765,7 @@ async def test_file_source_with_script_runner(self, tmp_path: Path) -> None: ) (skill_dir / "run.py").write_text("print('hi')", encoding="utf-8") - source = DeduplicatingSkillsSource( - FileSkillsSource(str(tmp_path), script_runner=_noop_script_runner) - ) + source = DeduplicatingSkillsSource(FileSkillsSource(str(tmp_path), script_runner=_noop_script_runner)) provider = SkillsProvider(source) await _init_provider(provider) assert "my-skill" in _ctx(provider)[0] @@ -3798,9 +3806,7 @@ async def source_runner(skill: Any, script: Any, args: Any = None) -> str: call_log.append("source") return "source" - source = DeduplicatingSkillsSource( - FileSkillsSource(str(tmp_path), script_runner=source_runner) - ) + source = DeduplicatingSkillsSource(FileSkillsSource(str(tmp_path), script_runner=source_runner)) provider = SkillsProvider(source) await _init_provider(provider) From 4a85384dac865ab5562d179eea59b2587e02006a Mon Sep 17 00:00:00 2001 From: Copilot Date: Thu, 7 May 2026 05:14:12 +0000 Subject: [PATCH 2/7] fix(mcp): propagate genuine task CancelledError in connect() (#5667) On Python >= 3.11, check task.cancelling() > 0 before wrapping CancelledError as ToolException in the three except blocks inside _connect_on_owner(). When the current task is being cancelled by its caller, the CancelledError now propagates after cleanup, consistent with the existing pattern at _mcp.py:560-564 and _runner.py:115-120. On Python < 3.11 task.cancelling() is unavailable, so MCP-internal CancelledErrors still cannot be reliably distinguished from caller-driven cancellation; they continue to be wrapped as ToolException with a comment documenting the trade-off. Tests: - Add cleanup assertion to transport-creation CancelledError test - Add MCPStdioTool variants exercising the 'command' message branches for both transport-creation and initialize CancelledError paths - Add Python 3.11+-gated tests verifying genuine task cancellation propagates (and still cleans up) for transport and initialize stages Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- python/packages/core/agent_framework/_mcp.py | 19 +++++ python/packages/core/tests/core/test_mcp.py | 83 ++++++++++++++++++++ 2 files changed, 102 insertions(+) diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py index 5da1ac23d72..dbed024fe15 100644 --- a/python/packages/core/agent_framework/_mcp.py +++ b/python/packages/core/agent_framework/_mcp.py @@ -656,6 +656,15 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: try: transport = await self._exit_stack.enter_async_context(self.get_mcp_client()) 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 isinstance(ex, asyncio.CancelledError) and sys.version_info >= (3, 11): + _task = asyncio.current_task() + if _task is not None and _task.cancelling() > 0: + await self._safe_close_exit_stack() + raise await self._safe_close_exit_stack() command = getattr(self, "command", None) if command: @@ -693,6 +702,11 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: ) ) except (Exception, asyncio.CancelledError) as ex: + if isinstance(ex, asyncio.CancelledError) and sys.version_info >= (3, 11): + _task = asyncio.current_task() + if _task is not None and _task.cancelling() > 0: + await self._safe_close_exit_stack() + raise await self._safe_close_exit_stack() raise ToolException( message="Failed to create MCP session. Please check your configuration.", @@ -701,6 +715,11 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: try: await session.initialize() except (Exception, asyncio.CancelledError) as ex: + if isinstance(ex, asyncio.CancelledError) and sys.version_info >= (3, 11): + _task = asyncio.current_task() + if _task is not None and _task.cancelling() > 0: + await self._safe_close_exit_stack() + raise await self._safe_close_exit_stack() # Provide context about initialization failure command = getattr(self, "command", None) diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 2b56281d1ea..6387b8c96d9 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -2,6 +2,7 @@ # type: ignore[reportPrivateUsage] import asyncio import json +import sys import logging import os from contextlib import _AsyncGeneratorContextManager # type: ignore @@ -2269,11 +2270,26 @@ async def test_connect_cleanup_on_initialization_failure(): async def test_connect_cancelled_error_during_transport_creation_raises_tool_exception(): """Test that CancelledError from transport creation is wrapped in ToolException.""" tool = MCPStreamableHTTPTool(name="test", url="http://example.com") + tool._exit_stack.aclose = AsyncMock() tool.get_mcp_client = Mock(side_effect=asyncio.CancelledError("cancel scope")) with pytest.raises(ToolException, match="Failed to connect to MCP server"): await tool.connect() + tool._exit_stack.aclose.assert_called_once() + + +async def test_connect_cancelled_error_during_transport_creation_stdio_raises_tool_exception(): + """Test that CancelledError from transport creation uses the command-specific message for MCPStdioTool.""" + tool = MCPStdioTool(name="test", command="my-server") + tool._exit_stack.aclose = AsyncMock() + tool.get_mcp_client = Mock(side_effect=asyncio.CancelledError("cancel scope")) + + with pytest.raises(ToolException, match="Failed to start MCP server 'my-server'"): + await tool.connect() + + tool._exit_stack.aclose.assert_called_once() + async def test_connect_cancelled_error_during_session_creation_raises_tool_exception(): """Test that CancelledError from session creation is wrapped in ToolException.""" @@ -2319,6 +2335,73 @@ async def test_connect_cancelled_error_during_initialize_raises_tool_exception() await tool.connect() +async def test_connect_cancelled_error_during_initialize_stdio_raises_tool_exception(): + """Test that CancelledError from session.initialize() uses the command-specific message for MCPStdioTool.""" + tool = MCPStdioTool(name="test", command="my-server", args=["--port", "8080"]) + + mock_transport = (Mock(), Mock()) + mock_context_manager = Mock() + mock_context_manager.__aenter__ = AsyncMock(return_value=mock_transport) + mock_context_manager.__aexit__ = AsyncMock(return_value=None) + tool.get_mcp_client = Mock(return_value=mock_context_manager) + + mock_session = Mock() + mock_session.initialize = AsyncMock(side_effect=asyncio.CancelledError("Cancelled via cancel scope")) + + with patch("mcp.client.session.ClientSession") as mock_session_class: + mock_session_class.return_value.__aenter__ = AsyncMock(return_value=mock_session) + mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) + + with pytest.raises(ToolException, match="MCP server 'my-server --port 8080' failed to initialize"): + await tool.connect() + + +@pytest.mark.skipif(sys.version_info < (3, 11), reason="task.cancelling() requires Python >= 3.11") +async def test_connect_genuine_cancellation_during_transport_creation_propagates(): + """Test that genuine task cancellation (task.cancelling() > 0) propagates as CancelledError.""" + tool = MCPStreamableHTTPTool(name="test", url="http://example.com") + tool._exit_stack.aclose = AsyncMock() + + mock_cancelled_task = Mock() + mock_cancelled_task.cancelling.return_value = 1 + + with patch("asyncio.current_task", return_value=mock_cancelled_task): + tool.get_mcp_client = Mock(side_effect=asyncio.CancelledError("task cancelled")) + with pytest.raises(asyncio.CancelledError): + await tool.connect() + + tool._exit_stack.aclose.assert_called_once() + + +@pytest.mark.skipif(sys.version_info < (3, 11), reason="task.cancelling() requires Python >= 3.11") +async def test_connect_genuine_cancellation_during_initialize_propagates(): + """Test that genuine task cancellation during initialize() propagates as CancelledError.""" + tool = MCPStreamableHTTPTool(name="test", url="http://example.com") + tool._exit_stack.aclose = AsyncMock() + + mock_transport = (Mock(), Mock()) + mock_context_manager = Mock() + mock_context_manager.__aenter__ = AsyncMock(return_value=mock_transport) + mock_context_manager.__aexit__ = AsyncMock(return_value=None) + tool.get_mcp_client = Mock(return_value=mock_context_manager) + + mock_session = Mock() + mock_session.initialize = AsyncMock(side_effect=asyncio.CancelledError("task cancelled")) + + mock_cancelled_task = Mock() + mock_cancelled_task.cancelling.return_value = 1 + + with patch("asyncio.current_task", return_value=mock_cancelled_task): + with patch("mcp.client.session.ClientSession") as mock_session_class: + mock_session_class.return_value.__aenter__ = AsyncMock(return_value=mock_session) + mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) + + with pytest.raises(asyncio.CancelledError): + await tool.connect() + + tool._exit_stack.aclose.assert_called_once() + + async def test_aenter_cancelled_error_during_connect_is_catchable_as_exception(): """Test that CancelledError during __aenter__ is catchable as Exception. From a0db1c9e6a5178c51b8c9e35d1961ac7eb693d4a Mon Sep 17 00:00:00 2001 From: Copilot Date: Thu, 7 May 2026 05:17:01 +0000 Subject: [PATCH 3/7] fix(mcp): log CancelledError with exc_info before wrapping in ToolException (#5667) CancelledError inherits from BaseException (not Exception) on Python >= 3.8, so the 'inner_exception=ex if isinstance(ex, Exception) else None' guard always yields None for CancelledError. This means ToolException.__init__ calls logger.log(level, message, exc_info=None), dropping the traceback. Add an explicit logger.debug(error_msg, exc_info=ex) before each raise ToolException(...) in the three CancelledError handlers so the full traceback is preserved in debug logs when MCP-internal cancellation is wrapped rather than propagated. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- python/packages/core/agent_framework/_mcp.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py index dbed024fe15..7d8fc2d6b9c 100644 --- a/python/packages/core/agent_framework/_mcp.py +++ b/python/packages/core/agent_framework/_mcp.py @@ -671,6 +671,10 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: error_msg = f"Failed to start MCP server '{command}': {ex}" else: error_msg = f"Failed to connect to MCP server: {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=ex) raise ToolException(error_msg, inner_exception=ex if isinstance(ex, Exception) else None) from ex try: try: @@ -708,8 +712,11 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: await self._safe_close_exit_stack() raise await self._safe_close_exit_stack() + session_error_msg = "Failed to create MCP session. Please check your configuration." + if isinstance(ex, asyncio.CancelledError): + logger.debug(session_error_msg, exc_info=ex) raise ToolException( - message="Failed to create MCP session. Please check your configuration.", + message=session_error_msg, inner_exception=ex if isinstance(ex, Exception) else None, ) from ex try: @@ -729,6 +736,8 @@ 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}" + if isinstance(ex, asyncio.CancelledError): + logger.debug(error_msg, exc_info=ex) 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] From 99a24c3b787027ad11fa583deb519f518f786444 Mon Sep 17 00:00:00 2001 From: Copilot Date: Thu, 7 May 2026 05:22:29 +0000 Subject: [PATCH 4/7] Address review feedback for #5667: Python: [Bug]: Error Handling Issue regarding Python MCPStreamableHTTPTool Class --- python/packages/core/agent_framework/_mcp.py | 12 ++++++------ python/packages/core/tests/core/test_mcp.py | 17 +++++++++-------- 2 files changed, 15 insertions(+), 14 deletions(-) diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py index 7d8fc2d6b9c..ee5437725c6 100644 --- a/python/packages/core/agent_framework/_mcp.py +++ b/python/packages/core/agent_framework/_mcp.py @@ -661,8 +661,8 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: # is unavailable so MCP-internal CancelledErrors cannot be distinguished from # caller-driven cancellation; they are wrapped as ToolException in that case. if isinstance(ex, asyncio.CancelledError) and sys.version_info >= (3, 11): - _task = asyncio.current_task() - if _task is not None and _task.cancelling() > 0: + task = asyncio.current_task() + if task is not None and task.cancelling() > 0: await self._safe_close_exit_stack() raise await self._safe_close_exit_stack() @@ -707,8 +707,8 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: ) except (Exception, asyncio.CancelledError) as ex: if isinstance(ex, asyncio.CancelledError) and sys.version_info >= (3, 11): - _task = asyncio.current_task() - if _task is not None and _task.cancelling() > 0: + task = asyncio.current_task() + if task is not None and task.cancelling() > 0: await self._safe_close_exit_stack() raise await self._safe_close_exit_stack() @@ -723,8 +723,8 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: await session.initialize() except (Exception, asyncio.CancelledError) as ex: if isinstance(ex, asyncio.CancelledError) and sys.version_info >= (3, 11): - _task = asyncio.current_task() - if _task is not None and _task.cancelling() > 0: + task = asyncio.current_task() + if task is not None and task.cancelling() > 0: await self._safe_close_exit_stack() raise await self._safe_close_exit_stack() diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 6387b8c96d9..2127d25a1df 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -2,9 +2,9 @@ # type: ignore[reportPrivateUsage] import asyncio import json -import sys import logging import os +import sys from contextlib import _AsyncGeneratorContextManager # type: ignore from typing import Any from unittest.mock import AsyncMock, Mock, patch @@ -2266,7 +2266,6 @@ async def test_connect_cleanup_on_initialization_failure(): tool._exit_stack.aclose.assert_called_once() - async def test_connect_cancelled_error_during_transport_creation_raises_tool_exception(): """Test that CancelledError from transport creation is wrapped in ToolException.""" tool = MCPStreamableHTTPTool(name="test", url="http://example.com") @@ -2391,13 +2390,15 @@ async def test_connect_genuine_cancellation_during_initialize_propagates(): mock_cancelled_task = Mock() mock_cancelled_task.cancelling.return_value = 1 - with patch("asyncio.current_task", return_value=mock_cancelled_task): - with patch("mcp.client.session.ClientSession") as mock_session_class: - mock_session_class.return_value.__aenter__ = AsyncMock(return_value=mock_session) - mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) + with ( + patch("asyncio.current_task", return_value=mock_cancelled_task), + patch("mcp.client.session.ClientSession") as mock_session_class, + ): + mock_session_class.return_value.__aenter__ = AsyncMock(return_value=mock_session) + mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) - with pytest.raises(asyncio.CancelledError): - await tool.connect() + with pytest.raises(asyncio.CancelledError): + await tool.connect() tool._exit_stack.aclose.assert_called_once() From d68ff983e3830a5627d26cbdb39a67c9607da0c0 Mon Sep 17 00:00:00 2001 From: Copilot Date: Thu, 7 May 2026 05:34:25 +0000 Subject: [PATCH 5/7] refactor(_mcp): extract cancellation helper, fix session error msg and exc_info - Extract _should_propagate_cancelled_error() helper to eliminate duplicated genuine-cancellation detection logic across the three connect() except blocks - Fix session-creation ToolException message to include exception details (e.g. 'Failed to create MCP session: ') matching the transport and initialize failure paths - Change exc_info=ex to exc_info=True in all three logger.debug() calls for idiomatic logging - Add tests for _should_propagate_cancelled_error helper - Add regression test asserting session error message includes exception text - Add test verifying logger.debug is called with exc_info=True Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- python/packages/core/agent_framework/_mcp.py | 48 +++++++----- python/packages/core/tests/core/test_mcp.py | 82 ++++++++++++++++++++ 2 files changed, 111 insertions(+), 19 deletions(-) diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py index ee5437725c6..0baab211d72 100644 --- a/python/packages/core/agent_framework/_mcp.py +++ b/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 @@ -660,11 +676,9 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: # 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 isinstance(ex, asyncio.CancelledError) and sys.version_info >= (3, 11): - task = asyncio.current_task() - if task is not None and task.cancelling() > 0: - await self._safe_close_exit_stack() - raise + if _should_propagate_cancelled_error(ex): + await self._safe_close_exit_stack() + raise await self._safe_close_exit_stack() command = getattr(self, "command", None) if command: @@ -674,7 +688,7 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: # 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=ex) + logger.debug(error_msg, exc_info=True) raise ToolException(error_msg, inner_exception=ex if isinstance(ex, Exception) else None) from ex try: try: @@ -706,15 +720,13 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: ) ) except (Exception, asyncio.CancelledError) as ex: - if isinstance(ex, asyncio.CancelledError) and sys.version_info >= (3, 11): - task = asyncio.current_task() - if task is not None and task.cancelling() > 0: - await self._safe_close_exit_stack() - raise + if _should_propagate_cancelled_error(ex): + await self._safe_close_exit_stack() + raise await self._safe_close_exit_stack() - session_error_msg = "Failed to create MCP session. Please check your configuration." + session_error_msg = f"Failed to create MCP session: {ex}" if isinstance(ex, asyncio.CancelledError): - logger.debug(session_error_msg, exc_info=ex) + logger.debug(session_error_msg, exc_info=True) raise ToolException( message=session_error_msg, inner_exception=ex if isinstance(ex, Exception) else None, @@ -722,11 +734,9 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: try: await session.initialize() except (Exception, asyncio.CancelledError) as ex: - if isinstance(ex, asyncio.CancelledError) and sys.version_info >= (3, 11): - task = asyncio.current_task() - if task is not None and task.cancelling() > 0: - await self._safe_close_exit_stack() - raise + if _should_propagate_cancelled_error(ex): + await self._safe_close_exit_stack() + raise await self._safe_close_exit_stack() # Provide context about initialization failure command = getattr(self, "command", None) @@ -737,7 +747,7 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: else: error_msg = f"MCP server failed to initialize: {ex}" if isinstance(ex, asyncio.CancelledError): - logger.debug(error_msg, exc_info=ex) + 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] diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 2127d25a1df..0b14ff4e3b0 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -29,6 +29,7 @@ _build_prefixed_mcp_name, _get_input_model_from_mcp_prompt, _normalize_mcp_name, + _should_propagate_cancelled_error, logger, ) from agent_framework._middleware import FunctionMiddlewarePipeline @@ -2178,6 +2179,7 @@ async def test_connect_session_creation_failure(): await tool.connect() assert "Failed to create MCP session" in str(exc_info.value) + assert "Session creation failed" in str(exc_info.value) # exception text is now part of the message assert "Session creation failed" in str(exc_info.value.__cause__) @@ -2435,6 +2437,86 @@ async def test_aenter_cancelled_error_during_connect_is_catchable_as_exception() assert isinstance(caught, ToolException) + +# Tests for _should_propagate_cancelled_error helper + + +def test_should_propagate_cancelled_error_returns_false_for_non_cancelled_error(): + assert _should_propagate_cancelled_error(RuntimeError("boom")) is False + + +def test_should_propagate_cancelled_error_returns_false_when_no_current_task(): + with patch("asyncio.current_task", return_value=None): + assert _should_propagate_cancelled_error(asyncio.CancelledError()) is False + + +@pytest.mark.skipif(sys.version_info < (3, 11), reason="task.cancelling() requires Python >= 3.11") +def test_should_propagate_cancelled_error_returns_true_when_task_is_cancelling(): + mock_task = Mock() + mock_task.cancelling.return_value = 1 + with patch("asyncio.current_task", return_value=mock_task): + assert _should_propagate_cancelled_error(asyncio.CancelledError()) is True + + +@pytest.mark.skipif(sys.version_info < (3, 11), reason="task.cancelling() requires Python >= 3.11") +def test_should_propagate_cancelled_error_returns_false_when_task_not_cancelling(): + mock_task = Mock() + mock_task.cancelling.return_value = 0 + with patch("asyncio.current_task", return_value=mock_task): + assert _should_propagate_cancelled_error(asyncio.CancelledError()) is False + + +async def test_connect_cancelled_error_during_session_creation_includes_exception_in_message(): + """Test that CancelledError from session creation includes exception details in ToolException message.""" + tool = MCPStreamableHTTPTool(name="test", url="http://example.com") + + mock_transport = (Mock(), Mock()) + mock_context_manager = Mock() + mock_context_manager.__aenter__ = AsyncMock(return_value=mock_transport) + mock_context_manager.__aexit__ = AsyncMock(return_value=None) + tool.get_mcp_client = Mock(return_value=mock_context_manager) + + with patch("mcp.client.session.ClientSession") as mock_session_class: + mock_session_class.return_value.__aenter__ = AsyncMock( + side_effect=asyncio.CancelledError("cancel scope detail") + ) + mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) + + with pytest.raises(ToolException) as exc_info: + await tool.connect() + + assert "Failed to create MCP session" in str(exc_info.value) + assert "cancel scope detail" in str(exc_info.value) + + +async def test_connect_cancelled_error_during_session_creation_logs_with_exc_info(): + """Test that CancelledError from session creation is logged with exc_info=True.""" + tool = MCPStreamableHTTPTool(name="test", url="http://example.com") + + mock_transport = (Mock(), Mock()) + mock_context_manager = Mock() + mock_context_manager.__aenter__ = AsyncMock(return_value=mock_transport) + mock_context_manager.__aexit__ = AsyncMock(return_value=None) + tool.get_mcp_client = Mock(return_value=mock_context_manager) + + with patch("mcp.client.session.ClientSession") as mock_session_class: + mock_session_class.return_value.__aenter__ = AsyncMock(side_effect=asyncio.CancelledError("cancel scope")) + mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) + + from agent_framework._mcp import logger as mcp_logger + + with patch.object(mcp_logger, "debug") as mock_debug: + with pytest.raises(ToolException): + await tool.connect() + + # Verify logger.debug was called with exc_info=True (not an exception instance) + debug_calls = mock_debug.call_args_list + cancel_calls = [c for c in debug_calls if "Failed to create MCP session" in str(c)] + assert cancel_calls, "Expected a debug log for the cancelled session creation" + _, kwargs = cancel_calls[0] + assert kwargs.get("exc_info") is True + + def test_mcp_stdio_tool_get_mcp_client_with_env_and_kwargs(): """Test MCPStdioTool.get_mcp_client() with environment variables and client kwargs.""" env_vars = {"PATH": "/usr/bin", "DEBUG": "1"} From 8bda2a87c0be2604eed4aefd779e2d1da2701cc6 Mon Sep 17 00:00:00 2001 From: Copilot Date: Thu, 7 May 2026 05:39:40 +0000 Subject: [PATCH 6/7] refactor: factor out _close_and_check_cancelled helper in _connect_on_owner Addresses review comment on PR #5687: 1. Add _close_and_check_cancelled() helper method that combines _safe_close_exit_stack() + _should_propagate_cancelled_error() into a single await-able call. This eliminates the duplicated close-then-check pattern that appeared identically in all three connect phases (transport, session, initialize), reducing future drift risk. 2. Comments 2 and 3 (missing {ex} in session error message and non-idiomatic exc_info=ex) were already addressed in the current code: all error messages include {ex} and all logger.debug calls use exc_info=True. 3. Add test_connect_genuine_cancellation_during_session_creation_propagates to cover the previously untested genuine-cancellation path in the session-creation phase (transport and initialize phases already had tests). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- python/packages/core/agent_framework/_mcp.py | 24 ++++++++++------ python/packages/core/tests/core/test_mcp.py | 30 ++++++++++++++++++++ 2 files changed, 45 insertions(+), 9 deletions(-) diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py index 0baab211d72..a8e6b9ac524 100644 --- a/python/packages/core/agent_framework/_mcp.py +++ b/python/packages/core/agent_framework/_mcp.py @@ -643,6 +643,18 @@ 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) @@ -676,10 +688,8 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: # 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 _should_propagate_cancelled_error(ex): - await self._safe_close_exit_stack() + if await self._close_and_check_cancelled(ex): raise - await self._safe_close_exit_stack() command = getattr(self, "command", None) if command: error_msg = f"Failed to start MCP server '{command}': {ex}" @@ -720,10 +730,8 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: ) ) except (Exception, asyncio.CancelledError) as ex: - if _should_propagate_cancelled_error(ex): - await self._safe_close_exit_stack() + if await self._close_and_check_cancelled(ex): raise - await self._safe_close_exit_stack() session_error_msg = f"Failed to create MCP session: {ex}" if isinstance(ex, asyncio.CancelledError): logger.debug(session_error_msg, exc_info=True) @@ -734,10 +742,8 @@ async def _connect_on_owner(self, *, reset: bool = False) -> None: try: await session.initialize() except (Exception, asyncio.CancelledError) as ex: - if _should_propagate_cancelled_error(ex): - await self._safe_close_exit_stack() + if await self._close_and_check_cancelled(ex): raise - await self._safe_close_exit_stack() # Provide context about initialization failure command = getattr(self, "command", None) if command: diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 0b14ff4e3b0..1d336241778 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -2405,6 +2405,36 @@ async def test_connect_genuine_cancellation_during_initialize_propagates(): tool._exit_stack.aclose.assert_called_once() +@pytest.mark.skipif(sys.version_info < (3, 11), reason="task.cancelling() requires Python >= 3.11") +async def test_connect_genuine_cancellation_during_session_creation_propagates(): + """Test that genuine task cancellation during session creation propagates as CancelledError.""" + tool = MCPStreamableHTTPTool(name="test", url="http://example.com") + tool._exit_stack.aclose = AsyncMock() + + mock_transport = (Mock(), Mock()) + mock_context_manager = Mock() + mock_context_manager.__aenter__ = AsyncMock(return_value=mock_transport) + mock_context_manager.__aexit__ = AsyncMock(return_value=None) + tool.get_mcp_client = Mock(return_value=mock_context_manager) + + mock_cancelled_task = Mock() + mock_cancelled_task.cancelling.return_value = 1 + + with ( + patch("asyncio.current_task", return_value=mock_cancelled_task), + patch("mcp.client.session.ClientSession") as mock_session_class, + ): + mock_session_class.return_value.__aenter__ = AsyncMock( + side_effect=asyncio.CancelledError("task cancelled") + ) + mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) + + with pytest.raises(asyncio.CancelledError): + await tool.connect() + + tool._exit_stack.aclose.assert_called_once() + + async def test_aenter_cancelled_error_during_connect_is_catchable_as_exception(): """Test that CancelledError during __aenter__ is catchable as Exception. From ab3f98ac3fd611f2f5d225d6631e78465ea73305 Mon Sep 17 00:00:00 2001 From: Copilot Date: Thu, 7 May 2026 05:44:46 +0000 Subject: [PATCH 7/7] Address review feedback for #5667: review comment fixes --- python/packages/core/agent_framework/_mcp.py | 1 - python/packages/core/tests/core/test_mcp.py | 5 +---- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py index a8e6b9ac524..0d85b1699ae 100644 --- a/python/packages/core/agent_framework/_mcp.py +++ b/python/packages/core/agent_framework/_mcp.py @@ -643,7 +643,6 @@ 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. diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 1d336241778..487331e3f0d 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -2424,9 +2424,7 @@ async def test_connect_genuine_cancellation_during_session_creation_propagates() patch("asyncio.current_task", return_value=mock_cancelled_task), patch("mcp.client.session.ClientSession") as mock_session_class, ): - mock_session_class.return_value.__aenter__ = AsyncMock( - side_effect=asyncio.CancelledError("task cancelled") - ) + mock_session_class.return_value.__aenter__ = AsyncMock(side_effect=asyncio.CancelledError("task cancelled")) mock_session_class.return_value.__aexit__ = AsyncMock(return_value=None) with pytest.raises(asyncio.CancelledError): @@ -2467,7 +2465,6 @@ async def test_aenter_cancelled_error_during_connect_is_catchable_as_exception() assert isinstance(caught, ToolException) - # Tests for _should_propagate_cancelled_error helper