MCP 服务器可以跑在不同地方，config.py 用三个模型描述（config.py:12 / config.py:53 / config.py:90）：

# config.py:12  ——本地进程，标准输入输出通信
class MCPServerStdio(BaseModel):
    command: str                         # 'python' / 'node' / 'npx' / 'uvx'
    args: list[str] = Field(default_factory=list)  # ['server.py'] 等
    env: dict[str, str] | None = None    # 传给子进程的环境变量
    tool_filter: ToolFilter | None = None
    cache_tools_list: bool = False       # 缓存工具清单，加速后续访问

# config.py:53  ——远程 HTTP（可流式）
class MCPServerHTTP(BaseModel):
    url: str                             # 'https://api.example.com/mcp'
    headers: dict[str, str] | None = None  # 认证头等
    streamable: bool = True
    tool_filter: ToolFilter | None = None
    cache_tools_list: bool = False

# config.py:90  ——远程 SSE（服务器推送事件）
class MCPServerSSE(BaseModel):
    url: str
    headers: dict[str, str] | None = None
    tool_filter: ToolFilter | None = None
    cache_tools_list: bool = False

# config.py:123
MCPServerConfig = MCPServerStdio | MCPServerHTTP | MCPServerSSE

MCPToolResolver.resolve（tool_resolver.py:68）是总入口，按引用类型分流：

# tool_resolver.py:68
def resolve(self, mcps: list[str | MCPServerConfig]) -> list[BaseTool]:
    all_tools: list[BaseTool] = []
    amp_refs: list[tuple[str, str | None]] = []
    for mcp_config in mcps:
        if isinstance(mcp_config, str) and mcp_config.startswith("https://"):
            all_tools.extend(self._resolve_external(mcp_config))    # ① 直接 URL
        elif isinstance(mcp_config, str):
            amp_refs.append(self._parse_amp_ref(mcp_config))        # ② AMP 短引用（"notion"）
        else:
            tools, clients = self._resolve_native(mcp_config)       # ③ 配置对象
            all_tools.extend(tools)
            self._clients.extend(clients)
    if amp_refs:
        tools, clients = self._resolve_amp(amp_refs)                # 批量解析 AMP
        all_tools.extend(tools)
        self._clients.extend(clients)
    return all_tools

_resolve_native（tool_resolver.py:313）先连服务器列出工具，再把每个包装成 MCPNativeTool：

# tool_resolver.py:333（发现）
async def _setup_client_and_list_tools() -> list[dict]:
    ...
    tools_list = await discovery_client.list_tools()   # ★问服务器：你有哪些工具
    ...
# tool_resolver.py:411（为每次调用准备"造客户端"的闭包）
def _client_factory() -> MCPClient:
    return MCPClient(transport=..., cache_tools_list=mcp_config.cache_tools_list, ...)
# tool_resolver.py:419（逐个包装）
for tool_def in tools_list:
    tool_name = tool_def.get("name", "")
    original_tool_name = tool_def.get("original_name", tool_name)
    if not tool_name:
        continue
    args_schema = None
    if tool_def.get("inputSchema"):
        try:
            args_schema = self._json_schema_to_pydantic(   # JSON schema → Pydantic 模型
                tool_name, tool_def["inputSchema"])
        except Exception as e:
            self._logger.log("warning", f"Failed to build args schema ... {e}")
            args_schema = None                              # 建不出 schema 也不放弃
    tool_schema = {"description": tool_def.get("description", ""), "args_schema": args_schema}
    native_tool = MCPNativeTool(                            # ★包装成 BaseTool
        client_factory=_client_factory, tool_name=tool_name,
        tool_schema=tool_schema, server_name=server_name,
        original_tool_name=original_tool_name)
    tools.append(native_tool)

MCPNativeTool._run（mcp_native_tool.py:73）要在"可能已有事件循环"的环境里跑异步代码：

# mcp_native_tool.py:73
def _run(self, **kwargs) -> str:
    try:
        try:
            asyncio.get_running_loop()              # 当前已在事件循环里？
            import concurrent.futures
            ctx = contextvars.copy_context()        # 复制上下文
            with concurrent.futures.ThreadPoolExecutor() as executor:
                coro = self._run_async(**kwargs)
                future = executor.submit(ctx.run, asyncio.run, coro)  # ★另开线程跑
                return future.result()
        except RuntimeError:
            return asyncio.run(self._run_async(**kwargs))  # 没有事件循环 → 直接跑
    except Exception as e:
        raise RuntimeError(f"Error executing MCP tool {self.original_tool_name}: {e!s}") from e

而 _run_async（mcp_native_tool.py:101）每次都新建一个客户端：

# mcp_native_tool.py:101
async def _run_async(self, **kwargs) -> str:
    client = self._client_factory()          # ★每次调用造一个全新客户端
    await client.connect()
    try:
        result = await client.call_tool(self.original_tool_name, kwargs)
    finally:
        await client.disconnect()            # 用完必断
    if isinstance(result, str):
        return result
    if hasattr(result, "content") and result.content:   # 提取文本内容
        if isinstance(result.content, list) and len(result.content) > 0:
            content_item = result.content[0]
            if hasattr(content_item, "text"):
                return str(content_item.text)
            return str(content_item)
        return str(result.content)
    return str(result)

另一种包装 MCPToolWrapper（mcp_tool_wrapper.py:16）带更完整的容错。先看常量和重试主循环（mcp_tool_wrapper.py:91）：

# mcp_tool_wrapper.py:10
MCP_CONNECTION_TIMEOUT = 15
MCP_TOOL_EXECUTION_TIMEOUT = 60
MCP_MAX_RETRIES = 3
# mcp_tool_wrapper.py:91
async def _retry_with_exponential_backoff(self, operation_func, **kwargs) -> str:
    last_error = None
    for attempt in range(MCP_MAX_RETRIES):
        result, error, should_retry = await self._execute_single_attempt(operation_func, **kwargs)
        if result is not None:
            return result               # 成功 → 返回
        if not should_retry:
            return error                # 不可重试的错 → 直接返回错误
        last_error = error
        if attempt < MCP_MAX_RETRIES - 1:
            wait_time = 2**attempt      # ★指数退避：1s, 2s, 4s...
            await asyncio.sleep(wait_time)
    return f"MCP tool execution failed after {MCP_MAX_RETRIES} attempts: {last_error}"

关键在 _execute_single_attempt（mcp_tool_wrapper.py:117）对错误分类——哪些该重试、哪些不该：

# mcp_tool_wrapper.py:117
async def _execute_single_attempt(self, operation_func, **kwargs):
    try:
        result = await operation_func(**kwargs)
        return result, "", False
    except ImportError:
        return None, "MCP library not available. Please install with: pip install mcp", False
    except asyncio.TimeoutError:
        return None, f"Connection timed out after {MCP_TOOL_EXECUTION_TIMEOUT} seconds", True  # 可重试
    except Exception as e:
        error_str = str(e).lower()
        if "authentication" in error_str or "unauthorized" in error_str:
            return None, f"Authentication failed for MCP server: {e!s}", False   # 不可重试
        if "not found" in error_str:
            return None, f"Tool '{self.original_tool_name}' not found on MCP server", False
        if "connection" in error_str or "network" in error_str:
            return None, f"Network connection failed: {e!s}", True               # 可重试
        if "json" in error_str or "parsing" in error_str:
            return None, f"Server response parsing error: {e!s}", True           # 可重试
        return None, f"MCP execution error: {e!s}", False

不是每个服务器的工具都想全暴露给 Agent。tool_filter 支持静态和动态两种（filters.py:17 / filters.py:38）：

# filters.py:17
class ToolFilterContext(BaseModel):
    agent: Any = Field(..., description="The agent requesting tools.")
    server_name: str = Field(..., description="Name of the MCP server.")
    run_context: dict[str, Any] | None = None
# filters.py:32
ToolFilter = (
    Callable[[ToolFilterContext, dict[str, Any]], bool]   # 动态：带上下文判断
    | Callable[[dict[str, Any]], bool])                   # 静态：只看工具本身
# filters.py:38
class StaticToolFilter:                # 简单的 allow/block 名单
    def __init__(self, allowed_tool_names=None, blocked_tool_names=None): ...

过滤应用在发现循环里（tool_resolver.py:383），而连接用完由 cleanup 统一断开（tool_resolver.py:90）：

# tool_resolver.py:90
def cleanup(self) -> None:
    if not self._clients:
        return
    async def _disconnect_all() -> None:
        for client in self._clients:
            if client and hasattr(client, "connected") and client.connected:
                await client.disconnect()      # 逐个断开
    try:
        asyncio.run(_disconnect_all())
    except Exception as e:
        self._logger.log("error", f"Error during MCP client cleanup: {e}")
    finally:
        self._clients.clear()                  # 清空登记表