From 5e524617a432d714a980addb32da0f16b607999c Mon Sep 17 00:00:00 2001 From: ray_le Date: Sat, 8 Aug 2026 16:55:56 +0800 Subject: [PATCH] =?UTF-8?q?fix(runner):=20=E5=B0=86=20MCP=20=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E5=A4=B1=E8=B4=A5=E7=9A=84=E7=9C=9F=E5=AE=9E=E5=8E=9F?= =?UTF-8?q?=E5=9B=A0=E4=BC=A0=E9=80=92=E7=BB=99=E4=B8=8B=E6=B8=B8=20error?= =?UTF-8?q?=5Fmessage?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit add_tool_server 此前对连接失败的可观测性不足,失败原因到不了 Error.error_message: 1. connect() 返回 False 时只抛空 reason 的 BaseError,真实异常丢失。 现在让各 McpClient 记录 _last_connect_error,add_tool_server 读取 它填入 CONNECTION_ERROR 的 reason。该自产 BaseError 在 except BaseException 入口即放行,避免被重包成 ADD_ERROR 而语义降级、 异常类型被埋进嵌套 message。 2. connect/list_tools 抛裸 CancelledError(MCP SDK streamable-http anyio task-group teardown)或含 GeneratorExit 的 BaseExceptionGroup 时,会穿透 except Exception 经 add_mcp_server 静默吞掉。现归一为 带 reason 的 BaseError(AD_ERROR),KeyboardInterrupt 原样透传。 新增单测固定 connect=False 穿透、CancelledError 归一、 BaseExceptionGroup 归一三条路径。 --- .../foundation/tool/mcp/client/mcp_client.py | 1 + .../foundation/tool/mcp/client/sse_client.py | 3 +- .../tool/mcp/client/stdio_client.py | 1 + .../tool/mcp/client/streamable_http_client.py | 28 +++++- .../core/runner/resources_manager/base.py | 10 +++ .../runner/resources_manager/tool_manager.py | 58 ++++++++++++- .../runner/test_tool_manager_mcp_dedup.py | 87 +++++++++++++++++++ 7 files changed, 180 insertions(+), 8 deletions(-) diff --git a/openjiuwen/core/foundation/tool/mcp/client/mcp_client.py b/openjiuwen/core/foundation/tool/mcp/client/mcp_client.py index 85d4d5588..207bf0450 100644 --- a/openjiuwen/core/foundation/tool/mcp/client/mcp_client.py +++ b/openjiuwen/core/foundation/tool/mcp/client/mcp_client.py @@ -14,6 +14,7 @@ def __init__(self, config: McpServerConfig): super().__init__() self._server_path = config.server_path self._include_image_content = bool(getattr(config, "include_image_content", False)) + self._last_connect_error: Optional[BaseException] = None @abstractmethod async def connect(self, *, retry_times: int = 1, timeout: float = NO_TIMEOUT) -> bool: diff --git a/openjiuwen/core/foundation/tool/mcp/client/sse_client.py b/openjiuwen/core/foundation/tool/mcp/client/sse_client.py index fd94db652..1f8fb4678 100644 --- a/openjiuwen/core/foundation/tool/mcp/client/sse_client.py +++ b/openjiuwen/core/foundation/tool/mcp/client/sse_client.py @@ -157,8 +157,7 @@ async def _do_connect(self, *, timeout: float) -> bool: except Exception as e: logger.error("[SseClient] SSE connection failed to %s: %s: %r", self._server_path, type(e).__name__, e) - # Clean up whatever partial state we have, but don't let cleanup - # exceptions mask the original connection error. + self._last_connect_error = e try: await self._do_disconnect(timeout=NO_TIMEOUT) except Exception as cleanup_exc: diff --git a/openjiuwen/core/foundation/tool/mcp/client/stdio_client.py b/openjiuwen/core/foundation/tool/mcp/client/stdio_client.py index 6448bfd3c..4cedcbbdb 100644 --- a/openjiuwen/core/foundation/tool/mcp/client/stdio_client.py +++ b/openjiuwen/core/foundation/tool/mcp/client/stdio_client.py @@ -53,6 +53,7 @@ async def connect(self, *, timeout: float = NO_TIMEOUT) -> bool: return True except Exception as e: logger.error(f"Stdio connection failed: {e}") + self._last_connect_error = e await self.disconnect() return False diff --git a/openjiuwen/core/foundation/tool/mcp/client/streamable_http_client.py b/openjiuwen/core/foundation/tool/mcp/client/streamable_http_client.py index 63be5d2d7..163a1fc5e 100644 --- a/openjiuwen/core/foundation/tool/mcp/client/streamable_http_client.py +++ b/openjiuwen/core/foundation/tool/mcp/client/streamable_http_client.py @@ -1,5 +1,6 @@ # coding: utf-8 # Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved. +import asyncio from contextlib import AsyncExitStack from typing import Any, Dict, List, Optional @@ -113,9 +114,32 @@ async def connect(self, *, timeout: float = NO_TIMEOUT) -> bool: self._is_disconnected = False logger.info(f"Streamable-http client connected successfully to {self._server_path}") return True - except Exception as e: + except BaseException as e: + # Extract HTTPStatusError from BaseExceptionGroup (caused by GeneratorExit + # being BaseException) so real HTTP errors aren't misreported as cancelled. + # CancelledError/KeyboardInterrupt propagate. + if isinstance(e, (KeyboardInterrupt, asyncio.CancelledError)) \ + and not isinstance(e, BaseExceptionGroup): + raise logger.error(f"Streamable-http connection failed to {self._server_path}: {e}") - await self.disconnect() + real_err: BaseException = e + if isinstance(e, BaseExceptionGroup): + exc_sub, _ = e.split(Exception) + if exc_sub is not None: + subs = list(exc_sub.exceptions) + real_err = subs[0] if len(subs) == 1 else exc_sub + else: + real_err = RuntimeError(f"MCP server '{self._name}' connect failed: {e}") + if isinstance(real_err, Exception): + self._last_connect_error = real_err + else: + self._last_connect_error = RuntimeError( + f"MCP server '{self._name}' connect failed: {type(e).__name__}: {e}" + ) + try: + await self.disconnect() + except BaseException as disc_exc: # noqa: BLE001 + logger.debug("[StreamableHttpClient] disconnect after connect failure failed: %r", disc_exc) return False async def disconnect(self, *, timeout: float = NO_TIMEOUT) -> bool: diff --git a/openjiuwen/core/runner/resources_manager/base.py b/openjiuwen/core/runner/resources_manager/base.py index de1f110f3..9d6d59d4a 100644 --- a/openjiuwen/core/runner/resources_manager/base.py +++ b/openjiuwen/core/runner/resources_manager/base.py @@ -295,6 +295,16 @@ def error(self) -> E: """ return self._error + @property + def error_message(self) -> E: + """Property alias for the error value (mirrors ``msg()``/``error()``). + + Downstream callers commonly probe a Result via ``getattr(result, + "error_message", None)``; expose the value through that name so the + underlying error is reachable without calling a method. + """ + return self._error + Result: TypeAlias = Ok[T] | Error[E] """ diff --git a/openjiuwen/core/runner/resources_manager/tool_manager.py b/openjiuwen/core/runner/resources_manager/tool_manager.py index 1b5ce017d..6d5f09cee 100644 --- a/openjiuwen/core/runner/resources_manager/tool_manager.py +++ b/openjiuwen/core/runner/resources_manager/tool_manager.py @@ -8,7 +8,7 @@ from openjiuwen.core.common.clients.client_registry import get_client_registry from openjiuwen.core.common.exception.codes import StatusCode -from openjiuwen.core.common.exception.errors import build_error +from openjiuwen.core.common.exception.errors import BaseError, build_error from openjiuwen.core.common.logging import runner_logger as logger from openjiuwen.core.foundation.tool import ( McpClient, @@ -120,18 +120,68 @@ async def add_tool_server( cards.append(deepcopy(tool.card)) return cards client = self._create_client(server_config) + connected = False try: connected = await client.connect() if not connected: + last_err = getattr(client, "_last_connect_error", None) + reason = f"{type(last_err).__name__}: {last_err}" if last_err else "" raise build_error( - StatusCode.RESOURCE_MCP_SERVER_CONNECTION_ERROR, server_config=server_config, reason="" + StatusCode.RESOURCE_MCP_SERVER_CONNECTION_ERROR, + server_config=server_config, reason=reason, ) results = await self._inner_refresh_mcp_tools(client, server_config, expiry_time) self._mcp_server_name_to_ids.setdefault(server_config.server_name, []).append(server_config.server_id) return results - except Exception as e: + except BaseException as e: + # ``connect``/``list_tools`` may raise a bare ``CancelledError`` or a + # ``BaseExceptionGroup`` (anyio task-group teardown, closing the + # generator-based client). Both are ``BaseException`` subclasses that + # escape ``except Exception``; coerce them into a ``BaseError`` so + # ``add_mcp_server`` wraps it in an ``Error`` and the real reason + # reaches the frontend instead of a silent "task cancelled". + if isinstance(e, KeyboardInterrupt): + raise + # ``connect()`` False path already raised CONNECTION_ERROR above; + # let it through to keep its status/reason. + if isinstance(e, BaseError) and e.status is StatusCode.RESOURCE_MCP_SERVER_CONNECTION_ERROR: + raise + if connected: + try: + await client.disconnect() + except BaseException as disconnect_exc: # noqa: BLE001 + logger.warning( + "add_tool_server cleanup failed: %s, server_id=%s", + disconnect_exc, server_config.server_id, + ) + if isinstance(e, Exception) and not isinstance(e, asyncio.CancelledError): + cause = e + elif isinstance(e, asyncio.CancelledError): + cause = RuntimeError( + f"MCP server '{server_config.server_name}' connect/register cancelled " + f"(anyio task-group teardown): {e}" + ) + elif isinstance(e, BaseExceptionGroup): + exc_subgroup, base_subgroup = e.split(Exception) + if exc_subgroup is not None and base_subgroup is None: + subs = list(exc_subgroup.exceptions) + cause = subs[0] if len(subs) == 1 else exc_subgroup + elif base_subgroup is not None: + cause = RuntimeError( + f"MCP server '{server_config.server_name}' register failed: {base_subgroup}" + ) + else: + cause = RuntimeError( + f"MCP server '{server_config.server_name}' register failed: {e}" + ) + else: + cause = RuntimeError( + f"MCP server '{server_config.server_name}' register failed: " + f"{type(e).__name__}: {e}" + ) raise build_error( - StatusCode.RESOURCE_MCP_SERVER_ADD_ERROR, cause=e, server_config=server_config, reason=str(e) + StatusCode.RESOURCE_MCP_SERVER_ADD_ERROR, cause=cause, server_config=server_config, + reason=str(cause), ) from e @staticmethod diff --git a/tests/unit_tests/core/runner/test_tool_manager_mcp_dedup.py b/tests/unit_tests/core/runner/test_tool_manager_mcp_dedup.py index 3c9539e20..d2d3b76b9 100644 --- a/tests/unit_tests/core/runner/test_tool_manager_mcp_dedup.py +++ b/tests/unit_tests/core/runner/test_tool_manager_mcp_dedup.py @@ -16,6 +16,8 @@ import pytest +from openjiuwen.core.common.exception.codes import StatusCode +from openjiuwen.core.common.exception.errors import BaseError from openjiuwen.core.foundation.tool import McpServerConfig, McpToolCard from openjiuwen.core.runner.resources_manager.tool_manager import ToolMgr @@ -132,3 +134,88 @@ def fake_create(config: McpServerConfig) -> MagicMock: assert [c.name for c in cards_b] == ["y"] assert fake_client_a.connect.await_count == 1 assert fake_client_b.connect.await_count == 1 + + +@pytest.mark.asyncio +async def test_add_tool_server_surfaces_connect_false_reason_as_connection_error() -> None: + """connect() returning False must raise CONNECTION_ERROR carrying the real + underlying exception in reason, and must NOT be re-wrapped into ADD_ERROR + by the BaseException handler below.""" + mgr = ToolMgr() + cfg = _make_server_config(server_id="false-srv") + + fake_client = MagicMock() + fake_client.connect = AsyncMock(return_value=False) + fake_client._last_connect_error = TimeoutError("handshake timed out") + fake_client.disconnect = AsyncMock(return_value=True) + + with patch.object(ToolMgr, "_create_client", staticmethod(lambda c: fake_client)): + with pytest.raises(BaseError) as exc_info: + await mgr.add_tool_server(cfg) + + err = exc_info.value + # Status preserved as CONNECTION_ERROR, not re-wrapped to ADD_ERROR. + assert err.status == StatusCode.RESOURCE_MCP_SERVER_CONNECTION_ERROR + # Real exception surfaced in reason, not a bare or nested-wrapped message. + assert "TimeoutError" in err.message + assert "handshake timed out" in err.message + + +@pytest.mark.asyncio +async def test_add_tool_server_coerces_cancelled_error_into_base_error() -> None: + """A bare ``CancelledError`` from connect (anyio task-group teardown) must be + coerced into a ``BaseError`` so ``add_mcp_server`` can wrap it in an ``Error`` + result and the frontend sees a connect failure instead of a silent cancel.""" + mgr = ToolMgr() + cfg = _make_server_config(server_id="cancel-srv") + + fake_client = MagicMock() + fake_client.connect = AsyncMock(side_effect=asyncio.CancelledError("teardown")) + fake_client.disconnect = AsyncMock(return_value=True) + + with patch.object(ToolMgr, "_create_client", staticmethod(lambda c: fake_client)): + with pytest.raises(BaseError) as exc_info: + await mgr.add_tool_server(cfg) + + err = exc_info.value + assert err.status == StatusCode.RESOURCE_MCP_SERVER_ADD_ERROR + # The original CancelledError is preserved on __cause__ for diagnosis. + assert isinstance(err.__cause__, asyncio.CancelledError) + # The surfaced reason must mention the server, not read as a bare "cancelled". + assert "cancel-srv" in err.message + # connect() never returned True, so disconnect must not be attempted. + assert fake_client.disconnect.await_count == 0 + + +@pytest.mark.asyncio +async def test_add_tool_server_coerces_exception_group_into_base_error() -> None: + """A ``BaseExceptionGroup`` mixing a regular ``Exception`` with a + ``GeneratorExit`` (the anyio task-group teardown shape) must be coerced into + a ``BaseError`` carrying a readable reason instead of escaping as a bare + cancel/silent teardown.""" + mgr = ToolMgr() + cfg = _make_server_config(server_id="egroup-srv") + + inner = RuntimeError("boom from list_tools") + # Mix a non-Exception BaseException (GeneratorExit) so the group stays a + # real BaseExceptionGroup rather than collapsing to ExceptionGroup. + fake_client = MagicMock() + fake_client.connect = AsyncMock(return_value=True) + fake_client.list_tools = AsyncMock( + side_effect=BaseExceptionGroup("group", [inner, GeneratorExit()]) + ) + fake_client.disconnect = AsyncMock(return_value=True) + + with patch.object(ToolMgr, "_create_client", staticmethod(lambda c: fake_client)): + with pytest.raises(BaseError) as exc_info: + await mgr.add_tool_server(cfg) + + err = exc_info.value + assert err.status == StatusCode.RESOURCE_MCP_SERVER_ADD_ERROR + # The mixed group is wrapped into a RuntimeError cause (the non-Exception + # sub-group cannot be re-raised as a plain Exception), surfacing a readable + # reason that names the server. + assert isinstance(err.cause, RuntimeError) + assert "demo" in str(err.cause) # server_name + # connect() returned True, so disconnect cleanup must run on the failure path. + assert fake_client.disconnect.await_count == 1