Agent CLI 源码剖析 05:MCP、传输层与扩展生态

MCP 在 Agent CLI 中的位置

本地工具是写在 Agent CLI 里的。MCP 允许外部程序把工具暴露给 Agent。

flowchart TD
    A[QueryEngine] --> B[Tool Registry]
    B --> C[Local Tools]
    B --> D[MCP Tool Wrappers]
    D --> E[MCP Client]
    E --> F[Transport]
    F --> G[MCP Server]

关键原则:

MCP 工具进入 Tool Registry 后,就应该像本地工具一样走 schema、权限、审计、超时和输出截断。

MCP 解决的问题

没有 MCP,每个 Agent 客户端都要自己适配外部系统:

代码块TEXT · 3 行收起展开

Agent A -> notes API
Agent B -> notes API
Agent C -> notes API

有 MCP:

代码块TEXT · 1 行收起展开

notes MCP server -> any MCP client

MCP 的价值不是“让模型更聪明”,而是:

统一工具发现。
统一资源读取。
统一工具调用。
让工具生态可复用。

MCP 配置

代码块JSON · 14 行收起展开

{
  "mcpServers": {
    "notes": {
      "command": "node",
      "args": ["notes-mcp-server.js"],
      "env": {
        "NOTES_ROOT": "F:\\2. ObsidianNotes"
      }
    },
    "browser": {
      "url": "http://localhost:3001/mcp"
    }
  }
}

两种常见连接:

类型	说明
stdio	启动本地子进程,通过 stdin/stdout 通信
HTTP/SSE/WebSocket	连接远程或本地服务

Transport 抽象

MCP 底层通常是 JSON-RPC 风格消息。先抽象 transport:

代码块TS · 22 行收起展开

// JsonRpcMessage = MCP 底层 JSON-RPC 消息。
type JsonRpcMessage = {
  jsonrpc: "2.0";       // 协议版本。
  id?: string | number; // request/response 关联 ID。
  method?: string;      // 请求方法，例如 tools/list。
  params?: unknown;     // 请求参数。
  result?: unknown;     // 成功响应结果。
  error?: {
    code: number;       // 错误码。
    message: string;    // 错误信息。
    data?: unknown;     // 可选错误详情。
  };
};

// Transport = MCP client 依赖的传输接口。
// stdio/http/websocket 都可以实现它。
interface Transport {
  start(): Promise<void>; // 启动连接。
  send(message: JsonRpcMessage): Promise<void>; // 发送消息。
  onMessage(handler: (message: JsonRpcMessage) => void): void; // 注册接收回调。
  close(): Promise<void>; // 关闭连接。
}

MCP Client 只依赖 Transport,不关心底层是 stdio 还是 WebSocket。

StdioTransport

代码块TS · 71 行收起展开

// StdioTransport = 本地子进程 MCP server 的传输实现。
class StdioTransport implements Transport {
  // MCP server 子进程。
  private child?: ChildProcess;

  // 收到完整 JSON-RPC 消息后要调用的 handler。
  private handlers: Array<(message: JsonRpcMessage) => void> = [];

  // stdout 可能分包，所以需要 buffer 拼行。
  private buffer = "";

  constructor(
    private command: string, // 启动命令。
    private args: string[],  // 参数。
    private env: Record<string, string | undefined> // 环境变量。
  ) {}

  async start() {
    // 启动子进程。
    this.child = spawn(this.command, this.args, {
      stdio: ["pipe", "pipe", "pipe"],
      env: { ...process.env, ...this.env }
    });

    // stdout 收 MCP JSON-RPC 消息。
    this.child.stdout?.on("data", chunk => {
      // 先追加到 buffer，不直接 JSON.parse。
      this.buffer += chunk.toString("utf8");
      this.drainLines();
    });

    // stderr 当日志，不当协议消息。
    this.child.stderr?.on("data", chunk => {
      appendMcpLog("stderr", chunk.toString("utf8"));
    });
  }

  // drainLines = 从 buffer 中取出完整行。
  private drainLines() {
    while (this.buffer.includes("\n")) {
      const index = this.buffer.indexOf("\n");
      // 取出一行并 trim。
      const line = this.buffer.slice(0, index).trim();
      // 剩余内容留在 buffer，等待下一次 data。
      this.buffer = this.buffer.slice(index + 1);

      if (!line) continue;

      // 一行是一个 JSON-RPC 消息。
      const message = JSON.parse(line);
      for (const handler of this.handlers) {
        handler(message);
      }
    }
  }

  async send(message: JsonRpcMessage) {
    // 发送时也是一行一个 JSON。
    this.child?.stdin?.write(JSON.stringify(message) + "\n");
  }

  onMessage(handler: (message: JsonRpcMessage) => void) {
    // 注册回调。
    this.handlers.push(handler);
  }

  async close() {
    // 关闭子进程。
    this.child?.kill();
  }
}

注意 buffer。stdout 可能分包,不一定一次 data 就是一条完整 JSON。

MCP Client

代码块TS · 98 行收起展开

// McpClient = MCP client，负责请求/响应配对和工具调用。
class McpClient {
  // 下一个 request id。
  private nextId = 1;

  // pending 保存等待响应的请求。
  private pending = new Map<number, {
    resolve(value: JsonRpcMessage): void;
    reject(error: Error): void;
    // 每个请求都有 timeout，避免永远挂着。
    timeout: NodeJS.Timeout;
  }>();

  constructor(
    public readonly name: string, // server 名。
    private transport: Transport  // 底层传输。
  ) {}

  async connect() {
    // 先注册消息处理，避免 start 后马上有消息但没人接。
    this.transport.onMessage(message => this.handleMessage(message));
    await this.transport.start();

    // 初始化握手。
    await this.request("initialize", {
      protocolVersion: "2026-03-26",
      clientInfo: {
        name: "learning-agent-cli",
        version: "0.1.0"
      }
    });
  }

  async listTools(): Promise<McpTool[]> {
    // 请求远端工具列表。
    const response = await this.request("tools/list", {});
    return (response.result as any).tools;
  }

  async callTool(name: string, input: unknown): Promise<ToolResult> {
    // 请求远端执行工具。
    const response = await this.request("tools/call", {
      name,
      arguments: input
    });

    // 转成本地统一 ToolResult。
    return normalizeMcpToolResult(response.result);
  }

  // request = 发送 JSON-RPC 请求并等待响应。
  private async request(method: string, params: unknown): Promise<JsonRpcMessage> {
    // 分配 ID。
    const id = this.nextId++;

    const promise = new Promise<JsonRpcMessage>((resolve, reject) => {
      // 超时后清理 pending 并 reject。
      const timeout = setTimeout(() => {
        this.pending.delete(id);
        reject(new Error(`MCP request timeout: ${method}`));
      }, 30_000);

      this.pending.set(id, { resolve, reject, timeout });
    });

    // 发出请求。
    await this.transport.send({
      jsonrpc: "2.0",
      id,
      method,
      params
    });

    // 等待 handleMessage resolve/reject。
    return promise;
  }

  // handleMessage = 收到 JSON-RPC response 后按 id 找回 pending。
  private handleMessage(message: JsonRpcMessage) {
    // 没有数字 id 的消息先忽略，例如 notification。
    if (typeof message.id !== "number") return;

    // 找对应请求。
    const pending = this.pending.get(message.id);
    if (!pending) return;

    // 清理 timeout 和 pending。
    clearTimeout(pending.timeout);
    this.pending.delete(message.id);

    // JSON-RPC error 转 reject，否则 resolve。
    if (message.error) {
      pending.reject(new Error(message.error.message));
    } else {
      pending.resolve(message);
    }
  }
}

重点:

每个 request 有 id。
pending map 等待响应。
超时必须清理 pending。
JSON-RPC error 转异常。

MCP Tool Wrapper

MCP tool 要包装成本地 Tool:

代码块TS · 22 行收起展开

// wrapMcpTool = 把远程 MCP tool 包装成本地 Tool。
// 包装后进入 Tool Registry，走和本地工具一样的权限/审计/超时。
function wrapMcpTool(client: McpClient, remoteTool: McpTool): Tool<any, any> {
  return {
    // namespace 防止不同 server 的 search/read 等工具重名。
    name: `mcp__${client.name}__${remoteTool.name}`,
    // 描述里标明来源。
    description: `[MCP:${client.name}] ${remoteTool.description}`,
    // 远程 schema 作为本地 inputSchema。
    inputSchema: remoteTool.inputSchema,
    // 根据名字和描述推断默认风险。
    risk: inferMcpRisk(remoteTool),
    // MCP 工具也必须有超时。
    timeoutMs: 30_000,
    // 返回内容也要限制大小。
    maxOutputBytes: 64_000,
    async run(input) {
      // 本地工具执行时，转发到远端 MCP server。
      return client.callTool(remoteTool.name, input);
    }
  };
}

为什么要 namespace?

因为多个 server 可能都有 search:

代码块TEXT · 3 行收起展开

mcp__notes__search
mcp__browser__search
mcp__database__search

MCP 风险推断

MCP server 不是天然可信。

代码块TS · 24 行收起展开

// inferMcpRisk = MCP 工具默认风险推断。
// 真实项目应允许用户在配置里覆盖。
function inferMcpRisk(tool: McpTool): ToolRisk {
  // 拼接名称和描述，用关键词粗略判断。
  const text = `${tool.name} ${tool.description}`.toLowerCase();

  // 写/删/更新类工具。
  if (text.includes("delete") || text.includes("write") || text.includes("update")) {
    return "write";
  }

  // 执行命令类工具。
  if (text.includes("run") || text.includes("exec") || text.includes("shell")) {
    return "execute";
  }

  // 网络/浏览器类工具。
  if (text.includes("fetch") || text.includes("browser") || text.includes("http")) {
    return "network";
  }

  // 默认只读。
  return "read";
}

真实项目要允许用户配置覆盖风险等级。

MCP 连接失败处理

不要一个 server 挂了,全局不可用。

代码块TS · 28 行收起展开

// loadMcpTools = 加载所有 MCP server 的工具。
// 单个 server 失败不应该导致整个 Agent CLI 启动失败。
export async function loadMcpTools(config: McpConfig) {
  const tools: Tool<any, any>[] = [];
  const errors: McpLoadError[] = [];

  for (const server of config.servers) {
    try {
      // 连接 server。
      const client = await connectServer(server);

      // 获取远程工具。
      const remoteTools = await client.listTools();

      // 包装成本地工具。
      tools.push(...remoteTools.map(tool => wrapMcpTool(client, tool)));
    } catch (error) {
      // 记录失败，继续下一个 server。
      errors.push({
        server: server.name,
        message: error instanceof Error ? error.message : String(error)
      });
    }
  }

  // 返回成功工具和失败列表，UI 可提示但不中断。
  return { tools, errors };
}

UI 提示即可:

代码块TEXT · 2 行收起展开

MCP server notes failed to connect: timeout.
Continuing without notes tools.

Agent Event Transport 和 MCP Transport 不同

容易混:

名称	通信对象	内容
MCP Transport	Agent CLI <-> MCP server	JSON-RPC
Agent Event Transport	Agent CLI -> 用户界面	AgentEvent

MCP message:

代码块JSON · 1 行收起展开

{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}

Agent event:

代码块JSON · 1 行收起展开

{"type":"tool_start","name":"read_file"}

不要把这两层写到一个模块里。

Command / Skill / Plugin 的位置

扩展生态不只有 MCP。

扩展	作用
Command	程序命令,例如 `/help`、`/mcp`
Skill	可复用工作流说明,注入上下文
Plugin	注册工具、命令、技能
MCP	外部进程/服务提供工具

关系图:

MCP 常见失败模式

失败	表现	修复
stdout 分包	JSON parse 失败	buffer 按行解析
request 无超时	pending 永久挂住	request timeout
工具重名	模型选错工具	namespace
server 挂了	整体不可用	单 server 隔离
MCP 工具绕过权限	越权	wrapper 走 ToolExecutor
无审计	无法定位外部工具问题	audit 记录 server

源码验收门槛

MCP 层的核心验收不是“能调通一个 server”,而是外部工具进入 Agent 后仍然受统一边界控制。

验收项	必须看到的证据	不合格表现
初始化可追踪	`initialize`、`tools/list`、`callTool` 都有 request id 和 timeout	server 卡住导致主循环挂死
工具命名隔离	MCP tool 进入 registry 后带 server namespace	不同 server 同名工具互相覆盖
schema 不丢失	wrapper 保留 input schema、description、风险等级	模型只能看到模糊工具名
权限统一	MCP 工具调用仍经过 `ToolExecutor` / permission gate	外部工具绕过本地审批
审计完整	audit log 记录 server、tool、args 摘要、耗时、错误	出错后不知道是哪台 server
故障隔离	单个 MCP server 失败不影响本地工具和其他 server	一个扩展挂掉拖垮整个 Agent

最小测试集:一个正常 stdio server、一个超时 server、一个输出非法 JSON 的 server、两个同名工具 server。四个 case 都能被隔离和记录,MCP 层才算可用于真实工程。

读源码抓手

看 MCP 和扩展层时,问:

MCP config 从哪里读?
stdio/http transport 怎么抽象?
initialize/listTools/callTool 在哪里?
MCP tool 如何包装成本地 Tool?
是否 namespace?
是否走统一权限?
连接失败是否隔离?
MCP transport 和 UI transport 是否分离?
plugin 能否绕过权限?

MCP 不只是“会调用外部工具”,它是 Agent 工具生态的边界层。