nanobot 源码解读 · 06 | MCP 集成：把任何外部服务变成 Agent 的工具

# agent/agent.py (loop.py:738)
async def _connect_mcp(self) -> None:
    """建立到所有已配置 MCP 服务器的连接。
    连接成功后，MCP 工具会被注册进 self.tools，LLM 就能在回合中调用它们。
    由网关在 loop 启动时(首次 run 前)调用一次。"""
    await agent_context.connect_mcp(self, self.tools)

# agent/loop.py:1368 —— 在 run() 的主循环前
self._running = True
try:
    await self._connect_mcp()  # 先建好 MCP 连接，工具才可用
    logger.info("Agent loop started")
    while self._running:
        ...

# mcp.py:984
transport_type = cfg.type
if not transport_type:
    if cfg.command:
        transport_type = "stdio"                      # 有 command → 本地子进程
    elif cfg.url:
        transport_type = (
            "sse" if cfg.url.rstrip("/").endswith("/sse") else "streamableHttp"
        )
    else:
        logger.warning("MCP server '{}': no command or url configured, skipping", name)
        await server_stack.aclose()
        return name, None

# mcp.py:997
if transport_type in {"sse", "streamableHttp"}:
    ok, error = validate_url_target(cfg.url)          # 05 篇的 SSRF 黑名单
    if not ok:
        logger.warning("MCP server '{}': blocked unsafe URL {} ({})", name, _redact_url(cfg.url), error)
        await server_stack.aclose()
        return name, None

# mcp.py:1023 / 1051
if not await _probe_http_url(cfg.url):
    logger.warning("MCP server '{}': {} unreachable, skipping", name, _redact_url(cfg.url))
    await server_stack.aclose()
    return name, None

# mcp.py:254
def _pinned_transport_kwargs() -> dict[str, Any]:
    kwargs: dict[str, Any] = {"transport": PinnedDNSAsyncTransport()}  # 05 篇的防 rebinding 传输
    mounts = httpx_env_proxy_mounts()
    if mounts:
        kwargs["mounts"] = mounts
    return kwargs

# mcp.py:262
async def _validate_mcp_request_url(request: httpx.Request) -> None:
    """Validate each outgoing MCP HTTP request, including redirect targets."""
    ok, error = validate_url_target(str(request.url))
    if not ok:
        raise httpx.RequestError(
            f"Blocked unsafe MCP URL {_redact_url(str(request.url))} ({error})", request=request
        )

# mcp.py:277
def _normalize_windows_stdio_command(command, args, env):
    ...
    # 对 npx/npm/pnpm/yarn/bunx 以及 .cmd/.bat，包一层 cmd.exe /d /c
    return comspec, ["/d", "/c", command, *normalized_args], env

# mcp.py:1073
read = _filter_malformed_mcp_progress_notifications(read, name)   # 过滤畸形 progress 通知
session = await server_stack.enter_async_context(ClientSession(read, write))
await session.initialize()                                        # 握手

tools = await session.list_tools()                               # 拉工具清单

# mcp.py:1083
available_wrapped_names = [_sanitize_mcp_tool_name(f"mcp_{name}_{tool_def.name}") for tool_def in tools.tools]

# mcp.py:158
def _sanitize_name(name: str) -> str:
    return _SANITIZE_RE.sub("_", re.sub(r"[^a-zA-Z0-9_-]", "_", name))

# mcp.py:167
def _limit_tool_name(name: str, max_length: int = _MAX_TOOL_NAME_LENGTH) -> str:
    if len(name) <= max_length:
        return name
    digest = hashlib.sha1(name.encode("utf-8")).hexdigest()[:_HASH_LENGTH]   # 超 64 字符用 sha1 截断
    prefix_length = max_length - _HASH_LENGTH - 1
    return f"{name[:prefix_length]}_{digest}"

# mcp.py:1078
enabled_tools = set(cfg.enabled_tools)
allow_all_tools = "*" in enabled_tools
...
for tool_def in tools.tools:
    wrapped_name = _sanitize_mcp_tool_name(f"mcp_{name}_{tool_def.name}")
    if (not allow_all_tools
        and tool_def.name not in enabled_tools
        and wrapped_name not in enabled_tools):
        continue   # 不在白名单 → 跳过

wrapper = MCPToolWrapper(session, name, tool_def, tool_timeout=cfg.tool_timeout)
registry.register(wrapper)

# mcp.py:1119
# Only register resources and prompts when no tool restriction is active.
# enabledTools is a per-tool allowlist; resources and prompts have no
# equivalent name filter, so they must be skipped whenever the operator
# specified a tool subset.
register_extras = allow_all_tools
if register_extras:
    # 注册 MCPResourceWrapper / MCPPromptWrapper
else:
    logger.info("MCP server '{}': skipping resource/prompt registration ...", name)

# mcp.py:467
class _MCPWrapperBase(Tool):
    """Common reconnect handling for wrappers bound to one MCP server session."""
    _plugin_discoverable = False

# mcp.py:475
def _set_mcp_connection(self, session, server_name):
    self._session = session
    self._server_name = server_name
    self._reconnect = None

def set_reconnect_handler(self, reconnect):
    self._reconnect = reconnect

# mcp.py:563
class MCPToolWrapper(_MCPWrapperBase):
    def __init__(self, session, server_name, tool_def, tool_timeout=30):
        self._set_mcp_connection(session, server_name)
        self._original_name = tool_def.name                          # 远端真名，调用时用
        self._name = _sanitize_mcp_tool_name(f"mcp_{server_name}_{tool_def.name}")
        self._description = tool_def.description or tool_def.name
        raw_schema = tool_def.inputSchema or {"type": "object", "properties": {}}
        self._parameters = _normalize_schema_for_openai(raw_schema)  # ← 翻译！
        self._tool_timeout = tool_timeout

# mcp.py:459
def _normalize_schema_for_openai(schema):
    if not isinstance(schema, dict):
        return {"type": "object", "properties": {}}
    schema_mapping = cast(dict[str, Any], schema)
    return _normalize_nullable_schema(_rewrite_local_schema_refs(schema_mapping))

# mcp.py:411
raw_type = normalized.get("type")
if isinstance(raw_type, list):
    type_values = cast(list[Any], raw_type)
    non_null = [item for item in type_values if item != "null"]
    if "null" in type_values and len(non_null) == 1:
        normalized["type"] = non_null[0]
        normalized["nullable"] = True

# mcp.py:385
name = f"ref_{hashlib.sha256(ref.encode()).hexdigest()[:12]}"   # 用哈希保证名字稳定唯一
generated_defs[name] = {}
generated_defs[name] = rewrite(target)                          # 先占坑再递归，处理自引用

# mcp.py:590
async def execute(self, **kwargs: Any) -> str:
    retried_transient = False
    refreshed_session = False
    while True:
        try:
            result = await asyncio.wait_for(
                self._session.call_tool(self._original_name, arguments=kwargs),
                timeout=self._tool_timeout,
            )
        except asyncio.TimeoutError:
            return ToolResult.error(f"(MCP tool call timed out after {self._tool_timeout}s)")
        except asyncio.CancelledError:
            if task_is_cancelling():           # 只有"外部真取消"（如用户 /stop）才往上抛
                raise
            return ToolResult.error("(MCP tool call was cancelled)")   # SDK 泄漏的取消则吞掉
        except Exception as exc:
            if await self._refresh_session_after_termination(exc, refreshed_session, "tool"):
                refreshed_session = True       # 会话断了 → 自动重连后重试
                continue
            if _is_transient(exc):
                if not retried_transient:
                    retried_transient = True
                    await asyncio.sleep(1)      # 瞬时错误（连接抖动）：退避 1s 重试一次
                    continue
                return ToolResult.error(f"(MCP tool call failed after retry: {type(exc).__name__})")
            return ToolResult.error(f"(MCP tool call failed: {type(exc).__name__})")
        else:
            try:
                rendered = self._render_call_result(result.content, kwargs)
                if getattr(result, "isError", False):
                    return ToolResult.error(rendered)
                return rendered
            except Exception as exc:
                return ToolResult.error(f"(MCP tool returned malformed content: {type(exc).__name__})")

# mcp.py:667
def _render_call_result(self, content, arguments):
    text_parts = []
    artifacts = []
    for block in content:
        if isinstance(block, types.TextContent):
            text_parts.append(block.text)
            continue
        data_url = _image_block_data_url(block, types)     # 解出 image 的 base64
        if data_url is not None:
            stored = self._store_image_block(data_url, arguments)  # 落盘成 artifact 文件
            if stored is not None:
                artifacts.append(stored)
            continue
        text_parts.append(str(block))
    if artifacts:
        return _mcp_image_tool_result(text_parts, artifacts)  # 只回传 artifact 路径 + 指引
    return "\n".join(text_parts) or "(no output)"

# mcp.py:1198
async def connect_single_server(name, cfg):
    loop = asyncio.get_running_loop()
    ready: asyncio.Future[bool] = loop.create_future()
    close_requested = asyncio.Event()

    async def own_connection() -> None:
        stack = None
        try:
            _, stack = await open_single_server(name, cfg)
            if not ready.done():
                ready.set_result(stack is not None)
            if stack is not None:
                await close_requested.wait()          # 阻塞，直到有人要关
        finally:
            if stack is not None:
                await stack.aclose()                 # 由"拥有者任务"负责关闭

    owner = asyncio.create_task(own_connection(), name=f"mcp:{name}")
    connection = _OwnedMCPConnection(owner, close_requested)
    ...

# mcp.py:70
class _OwnedMCPConnection:
    """Close an MCP transport from the task that originally opened it."""
    async def aclose(self):
        self._close_requested.set()                  # 通知拥有者任务"该关了"
        try:
            await asyncio.shield(self._owner)        # 等它自己跑完清理，但不被外部取消打断
        except asyncio.CancelledError:
            if not self._owner.cancelled():
                raise

# mcp.py:1289
async def reload_servers(state, registry):
    """Reconcile live MCP connections with the current config file."""
    # 重新读配置 → 对比 current vs next
    #   removed  = 当前有、新配置没有 → 注销工具 + 关连接
    #   added    = 新配置有、当前没有 → 新建连
    #   changed  = 签名变了 → 先关后建
    ...
    connected = await connect_mcp_servers(to_connect, registry)
    state._mcp_stacks.update(connected)
    _attach_reconnect_handlers(state, registry, connected)

# mcp.py:1387
async def request_mcp_reload(bus, *, timeout=15.0):
    await bus.publish_inbound(InboundMessage(
        channel="system", sender_id="webui-settings", chat_id="runtime",
        content=RUNTIME_CONTROL_MCP_RELOAD,
        metadata={INBOUND_META_RUNTIME_CONTROL: RUNTIME_CONTROL_MCP_RELOAD, RUNTIME_CONTROL_ACK: ack},
    ))
    result = await asyncio.wait_for(ack, timeout=timeout)

# mcp.py:1453
def _attach_reconnect_handlers(state, registry, server_names):
    async def reconnect(server_name, tool_name, stale_tool):
        return await _refresh_terminated_server(state, registry, server_name, tool_name, stale_tool)
    for server_name in server_names:
        for tool_name in list(registry.tool_names):
            tool = registry.get(tool_name)
            if not _tool_belongs_to_server(tool, tool_name, server_name):
                continue
            if isinstance(tool, _MCPWrapperBase):
                tool.set_reconnect_handler(reconnect)   # 把回调种进工具

[启动期]
配置 mcpServers → AgentLoop._connect_mcp (loop.py:738)
  → connect_mcp_servers → connect_single_server → open_single_server
    → 建传输(stdio/sse/streamableHttp) + ClientSession.initialize()
    → list_tools → 逐个 MCPToolWrapper → registry.register
        工具名 = mcp_{server}_{tool}

[一次用户消息]
用户问"帮我读 /data/report.pdf 的第 3 页"
  → AgentLoop 分发 → ContextBuilder 装配 → AgentRunner 模型循环
    → 模型决定调用 mcp_filesystem_read_pdf
    → runner._run_tool（05 篇：并发分组/安全分类）
      → registry 取出 MCPToolWrapper → Tool.prepare_call 校验参数
      → wrapper.execute(**kwargs)
        → asyncio.wait_for(session.call_tool(原始名, 参数), timeout)
          → 走网络到远端 MCP Server 真正执行
        → 结果(文本/图片) → _render_call_result（图片落盘成 artifact）
        → ToolResult 回写消息流
    → 模型拿到结果，组织回答 → 回复用户

tools:
  mcpServers:
    fs:
      type: stdio
      command: npx
      args: ["-y", "@modelcontextprotocol/server-filesystem", "/data"]
      enabledTools: ["*"]