rpi-mcp-adapter 源码解读：一个 Rust MCP 适配器如何管理工具调用

2026年10月14日15 分钟6,477 字#rpi #Rust #MCP #源码分析 #Agent

很多 MCP 文章讲的是“Host、Client、Server”三个概念，但真正接入一个 MCP Server 时，代码要处理的事情远不止发一条 JSON-RPC：配置如何解析？本地进程如何启动？远程 HTTP 会话如何保存？超时后能不能安全退出？工具列表变化后是否重新发现？

这篇文章不再从 MCP 规范泛讲，而是直接阅读 rpi-mcp-adapter 的代码，回答一个具体问题：rpi 是如何把外部 MCP Server 接进自己的工具循环的？

本文对应 rpi-packages 中的 rpi-mcp-adapter 0.2.1，源码范围主要是：

packages/rpi-mcp-adapter/
├── src/lib.rs          插件入口和三个宿主工具
├── src/config.rs       mcpServers 配置解析
├── src/connection.rs   MCP 初始化与 tools 请求构造
├── src/runtime.rs      连接缓存、发现、调用和关闭
├── src/transport.rs    stdio 子进程传输
└── src/http.rs         Streamable HTTP 传输

源码仓库：https://github.com/pi-rust/rpi-package/tree/master/packages/rpi-mcp-adapter

先把 MCP 协议本身讲清楚

如果直接从 mcp_call 读起，很容易把 MCP 理解成“给一个 URL 发 tools/call”。这会漏掉适配器真正需要管理的状态。MCP 是一个基于 JSON-RPC 2.0 的、有生命周期和能力协商的客户端—服务端协议；工具只是其中一种能力。

MCP 的三层角色：Host、Client、Server

在本文里，三者的关系是：

MCP 角色	在这条调用链中的实例	职责
Host	rpi 宿主/Agent	持有模型、会话和用户上下文，并向模型暴露 `mcp_call` 等 rpi 工具
MCP Client	`rpi-mcp-adapter` 内部的 `Session`	为某个 MCP Server 建立一条独立连接，协商能力、发送请求、匹配响应
MCP Server	Playwright 等外部服务	声明工具及 JSON Schema，执行 `tools/call` 并返回结果

这里最容易混淆的是：mcp_call 是 rpi Host 暴露给模型的工具，不是 MCP 的协议方法；真正发给外部 Server 的方法名是 tools/call。同理，mcp_list 不是 MCP 的 tools/list，它是适配器给 Host 的发现接口，内部才会调用 tools/list。

一条完整的数据流是：

用户任务
  → rpi 模型决定调用 mcp_call
  → rpi Plugin ABI 调用适配器
  → MCP Client Session
  → JSON-RPC tools/call
  → 外部 MCP Server 执行工具
  → MCP result
  → 适配器转换为 rpi ToolResult
  → 模型继续推理

MCP Server 看不到 rpi 的完整对话。它只收到这一次 MCP 请求以及请求参数；Host 负责决定哪些工具暴露给模型、什么时候调用，以及是否需要用户确认。

JSON-RPC 消息为什么需要 request id？

MCP 的每条消息都是 JSON-RPC 2.0。最小的请求、响应和通知分别是：

// 请求：需要响应，id 用于匹配
{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"browser_tabs","arguments":{"action":"list"}}}

// 响应：必须带回同一个 id
{"jsonrpc":"2.0","id":2,"result":{"content":[{"type":"text","text":"..."}],"isError":false}}

// 通知：没有 id，不等待响应
{"jsonrpc":"2.0","method":"notifications/initialized"}

id 不是工具名，也不是 MCP session id。适配器可以同时管理多条连接、在 HTTP 中处理 SSE 消息、在 stdio 中遇到 Server 发来的 ping；只有用 request id 匹配，才知道某个响应属于哪个请求。runtime.rs 中的 next_id 就是这一层的请求编号，而 Session 是连接级状态。

JSON-RPC 的错误也分两层：

协议错误：响应带 error，例如方法不存在、参数结构不合法或 JSON-RPC 请求失败；
工具执行错误：响应仍然是 result，但结果里 isError: true，例如 API 限流、业务校验失败或工具执行失败。

connection.rs 先把 JSON-RPC error 转成 Rust 错误，再检查 isError；两者不能混成“只看 HTTP 状态码”。HTTP 200 也可能包含 MCP 工具执行失败。

一条 MCP 连接不是一次请求

MCP 的初始化必须先于正常业务请求：

Client → initialize
Server → InitializeResult
Client → notifications/initialized
Client ↔ tools/list
Client ↔ tools/call
Client → shutdown/断开传输

initialize 做三件事：协商协议版本、交换能力、交换实现信息。当前适配器发送 2025-06-18，并用 parse_initialize 接受 2024-11-05、2025-03-26 和 2025-06-18 的响应；同时要求 Server 声明 capabilities.tools 和 serverInfo。随后必须发送 notifications/initialized，才能进入正常操作阶段。

能力声明不是装饰信息：Server 没有声明 tools，适配器就不会把它当作可调用的工具 Server；Server 声明了 resources 或 prompts，也不代表当前这个适配器已经实现了相应 dispatcher。协议能力、传输能力和本 package 已实现的能力是三件事。

`tools/list` 返回的不是“工具名数组”

MCP 工具描述至少包含三个关键部分：

{
  "name": "browser_tabs",
  "description": "操作浏览器标签页",
  "inputSchema": {
    "type": "object",
    "properties": {"action": {"type": "string"}},
    "required": ["action"]
  }
}

inputSchema 是给 Client 和模型看的 JSON Schema。它决定模型应该生成怎样的 arguments，但不等于适配器已经替 Server 验证了所有业务规则。rpi-mcp-adapter 在进入外部连接前只要求 arguments 是 object，并限制工具名和超时；具体字段校验仍由 MCP Server 负责。

tools/list 支持 nextCursor 分页。适配器没有把第一页当成完整列表，而是持续请求下一页，并拒绝重复 cursor。这样处理的原因是工具列表本身也是协议状态：可能很长，也可能随着 Server 状态变化。

`tools/call` 的结果为什么有两种内容？

工具结果通常包含：

{
  "content": [
    {"type":"text","text":"结果说明"},
    {"type":"image","data":"...","mimeType":"image/png"}
  ],
  "structuredContent": {"count": 3},
  "isError": false
}

content 是面向交互的内容块，可以是文本、图片、音频、资源链接或嵌入资源；structuredContent 是结构化 JSON 数据，适合程序继续处理。它们不是二选一：一个工具可以同时返回可展示文本和结构化结果。

当前适配器会把文本和图片转换成 rpi 宿主能理解的原生内容，把其他内容转成文本表示，并把原始结果（包括 structuredContent）保留在 details 中。这是一个有损但可追踪的适配边界，不是把 MCP result 原封不动地交给模型。

下面再看 rpi-mcp-adapter 如何把这套协议实现成可复用的连接运行时。

先看结论：它不是 MCP Server，而是 MCP Client 侧适配器

rpi-mcp-adapter 的 Cargo 描述是：

Session-scoped stdio and Streamable HTTP MCP adapter for the rpi Rust agent

它注册的不是一个供其他客户端调用的 MCP endpoint，而是三个 rpi 工具：

mcp_list       列出配置，或发现某个 MCP Server 的工具
mcp_call       调用已发现的 MCP 工具
mcp_request    发送底层 HTTP JSON-RPC 请求

因此调用方向是：

rpi Agent
   │ 调用 rpi 工具
   ▼
rpi-mcp-adapter
   │ 管理连接并发送 MCP JSON-RPC
   ├── stdio → 外部 MCP 子进程
   └── HTTP  → 远程 MCP endpoint

这也解释了一个容易误读的地方：虽然代码里有 mcp_list 和 mcp_call，它们不是 MCP 协议中的 tools/list 和 tools/call 本身，而是 rpi 暴露给模型的上一层工具。真正的 MCP 方法由适配器内部构造。

入口：插件注册了三个稳定工具

入口在 src/lib.rs。export_plugin! 回调拿到宿主的 register_tool 后，注册三个工具及其 JSON Schema：

let tools = [
    ("mcp_request", /* 原始 HTTP JSON-RPC */),
    ("mcp_call",    /* 持久 stdio/HTTP 连接 */),
    ("mcp_list",    /* 配置列表或工具发现 */),
];

每个工具都使用同一个 ABI 执行模式：

宿主调用 execute
       │
       ▼
kit::execute(params, free, builder)
       │
       ▼
后台任务执行 runtime / transport
       │
       ├── poll：非阻塞取得进度
       ├── cancel：取消控制信号
       └── destroy：释放 StepHandle

这意味着 MCP 的网络或子进程 I/O 不会直接堵住宿主线程。tool_poll、tool_cancel 和 tool_destroy 最终都委托给 kit。

插件还注册了 SessionShutdown 事件。宿主卸载扩展或结束会话时，会调用 shutdown_connections()，停止并清理连接，而不是把 MCP 子进程留在后台。

配置解析：`mcpServers` 兼容两种形状

src/config.rs 定义了顶层配置：

pub struct McpConfig {
    #[serde(default, rename = "mcpServers")]
    pub servers: BTreeMap<String, McpServerConfig>,
}

每个服务只有两种配置：

pub enum McpServerConfig {
    Stdio(StdioServer),
    Http(HttpServer),
}

一个实际配置可以是：

{
  "mcpServers": {
    "playwright": {
      "command": "node",
      "args": ["server.js", "--extension"],
      "timeout": 60
    },
    "remote": {
      "url": "http://localhost:8931/mcp",
      "headers": {}
    }
  }
}

解析阶段做了几项边界检查：

Server 名称只能包含 ASCII 字母、数字、- 和 _；
stdio Server 的 command 不能是空字符串；
HTTP Server 的 url 不能是空字符串；
默认超时是 60 秒；
timeoutSeconds 的单次覆盖范围是 1 到 300 秒。

exposure 会被解析保存，但 README 明确说明，当前稳定 dispatcher 没有实现 Pi 旧版的动态 exposure 模式。也就是说，配置能读进去，不代表每种 exposure 都会改变实际暴露行为。

`mcp_call` 的完整调用链

从宿主工具到远端工具，实际经过了配置、连接缓存、MCP 初始化、工具发现和工具调用多个阶段：

flowchart TD
    T[rpi 工具 mcp_call] --> V[校验 tool、arguments、timeout]
    V --> K[根据 host session/cwd<br/>配置路径和 Server 名生成 key]
    K --> C{连接已初始化？}
    C -->|否| I[创建 stdio 或 HTTP transport<br/>发送 initialize]
    I --> N[校验 protocolVersion、tools capability<br/>发送 initialized]
    C -->|是| L[复用连接]
    N --> D[tools/list 分页发现]
    L --> D
    D --> H{目标工具存在？}
    H -->|否| E[返回错误]
    H -->|是| X[tools/call]
    X --> R[转换 MCP result<br/>回到 rpi 工具循环]

这条链说明 mcp_call 不是对远端 URL 直接转发：适配器会先确认连接、协商协议、刷新工具列表，然后才允许调用。

第一步：参数验证

runtime::call 先要求 tool 非空且不超过 200 个字符，arguments 必须是 JSON object。数组、字符串或其他 JSON 类型会被拒绝：

let arguments = params
    .get("arguments")
    .cloned()
    .unwrap_or_else(|| json!({}));
if !arguments.is_object() {
    return Err("MCP arguments must be an object".into());
}

这层检查发生在真正连接 MCP Server 之前，避免把明显非法的参数送入外部进程或网络。

第二步：按调用上下文确定连接 key

runtime.rs 的 host_scope 会优先读取宿主注入的 sessionId，没有时再使用规范化后的工作目录，最后退回 library：

session:<host session id>
project:<canonical cwd>
library

再加上配置文件路径和 Server 名称，形成连接注册表的 key。结果是：同一个宿主会话和同一个 MCP Server 复用连接，不同会话不会意外共享同一条连接。

第三步：第一次使用时初始化

连接状态保存在全局 SESSIONS 注册表中，每个 Entry 有一个可选的 Session 和一个停止控制器。第一次访问时，Session::start 根据配置选择：

match config {
    McpServerConfig::Stdio(s) => StdioTransport::start(...),
    McpServerConfig::Http(s) => HttpTransport::new(...),
}

随后统一走 MCP 初始化：

initialize {
  "protocolVersion": "2025-06-18",
  "capabilities": {},
  "clientInfo": {
    "name": "rpi-mcp-adapter",
    "version": "0.2.1"
  }
}

parse_initialize 不只是读取版本。它还要求服务端：

返回受支持的协议版本：2024-11-05、2025-03-26 或 2025-06-18；
在 capabilities 中声明 tools；
提供 serverInfo。

如果服务端只提供 Resources 或 Prompts、没有 Tools capability，这个适配器会拒绝初始化。原因不是 MCP 规范不允许这些能力，而是 rpi-mcp-adapter 的稳定 dispatcher 当前只实现工具调用。

初始化成功后再发送：

{"jsonrpc":"2.0","method":"notifications/initialized"}

HTTP 传输还会把协商得到的版本保存下来，后续请求通过 MCP-Protocol-Version 请求头发送。

为什么每次调用前都重新 `tools/list`？

Session::tools 会分页请求 tools/list，最多 100 页、最多 16384 个工具，并拒绝重复的 nextCursor：

for _ in 0..100 {
    let result = self.request("tools/list", params, control)?;
    tools.extend(tool_list(&result)?);
    // 处理 nextCursor，检测重复 cursor
}

runtime::call 每次 mcp_call 都会执行这次发现，然后检查目标工具是否存在，再调用：

if !tools.iter().any(|t| t.name == tool) {
    return Err(format!("MCP server does not expose tool {tool:?}"));
}

这带来两个实际效果：

Server 的工具列表变化可以被后续调用发现；
模型传来的工具名不会被盲目转发。

代价是每次工具调用都会多一次 tools/list 往返。代码选择了“发现结果新鲜”而不是本地永久缓存；这也是适配器行为，不能泛化成所有 MCP Client 都必须这样做。

stdio：三个后台 worker 保证管道不阻塞

src/transport.rs 的 StdioTransport::start 会启动外部进程，并接管 stdin、stdout、stderr：

外部 MCP Server
  stdin  ← writer worker
  stdout → reader worker → bounded channel
  stderr → logger worker → 最近 8 KiB

三个 worker 分别负责：

writer：从有界队列读取 JSON，写入 stdin 并 flush；
reader：按行读取 stdout，解析 JSON；
logger：读取 stderr，只保留最近 8 KiB，错误时附加到诊断信息。

bounded_line 不使用无界 read_line，而是分块读取并限制单条消息最大 16 MiB。Server 不换行或发送超大响应时，适配器会报错而不是无限增长内存。

reader 还有一个容易忽略的处理：如果读到的是 Server 发来的请求，例如 ping，它不会把这条消息误当成当前调用的响应。client_reply 会自动回复 ping，其他客户端方法返回 -32601 method not supported。

响应匹配通过 JSON-RPC request id 完成；不属于当前请求的响应会继续等待，而不是直接交给错误的调用。

Streamable HTTP：不仅是 POST JSON

src/http.rs 的 HttpTransport 保存了：

reqwest Client
当前 MCP URL
Mcp-Session-Id
MCP-Protocol-Version
timeout

每次 HTTP 请求默认带：

Accept: application/json, text/event-stream

如果已经通过初始化拿到 session，就继续带上：

Mcp-Session-Id: ...
MCP-Protocol-Version: 2025-06-18

返回值支持两种形式：

application/json：读取完整 JSON-RPC 响应；
text/event-stream：增量读取 SSE，直到找到匹配 request id 的 JSON-RPC 响应。

代码还明确禁止 HTTP 重定向，避免带有认证 header 的请求被转发到另一个 endpoint；URL 不能嵌入用户名和密码，凭据应该通过配置 headers 提供。

如果已有 session 的请求返回 404，适配器会认为 MCP HTTP session 已过期，丢弃旧连接，下一次调用重新初始化。它不会自动重放失败的工具调用，因为工具可能已经产生副作用。

超时、取消和失败后的处理

Control 和 Deadline 会覆盖多个等待位置：

等待连接锁；
向 stdio 写入；
从 stdout 读取；
HTTP 请求；
等待 SSE 响应。

stdio 超时或取消时，StdioTransport::stop 会停止 writer/reader，杀掉进程树并 join worker。HTTP 取消时会尽力发送 notifications/cancelled。

如果一次请求失败，Session 会标记 failed，with_session 在调用结束后丢弃它。下一次调用会重新初始化，而不是复用坏连接。

这里有一条重要的设计取舍：连接会自动重建，但工具调用不会自动重放。 自动重放可能导致创建两次工单、发送两封邮件或重复写入数据，所以代码选择让上层决定是否重试。

连接注册表最多保存 64 个条目。mcp_list 的 action: "close" 可以主动释放某个连接；扩展收到 SessionShutdown 时则释放全部连接。

结果如何回到 rpi？

MCP 工具结果先经过 JSON-RPC 解析：

JSON-RPC error 变成适配器错误；
MCP result.isError = true 也变成工具错误；
成功的文本和图片块转换成 rpi 宿主原生内容；
其他内容块获得文本表示；
原始 MCP 结果（包括 structured content）保留在 result details 中。

因此，rpi-mcp-adapter 的稳定接口不是“把所有 MCP 能力原样暴露给 Agent”。它选择了一个明确的适配边界：模型通过三个固定的 rpi 工具使用外部 MCP 工具，结果再转换回宿主 ABI。

当前没有实现：

Resources；
Prompts；
Sampling；
Elicitation；
OAuth；
动态注册 rpi 工具。

这部分限制必须以代码和 README 为准，不能因为 MCP 规范支持某能力，就说这个 package 已经支持。

测试也暴露了它真正关心的边界

packages/rpi-mcp-adapter 的测试覆盖的不是“能否发出一条请求”这么简单，还包括：

持久 stdio 进程；
tools/list 分页；
阻塞写入；
取消和超时；
16 MiB 消息限制；
stderr 诊断尾部；
HTTP JSON 和 SSE 生命周期；
session header 与过期；
ping 回复；
图片内容；
子进程树清理。

推荐在源码仓库执行：

cargo test -p rpi-mcp-adapter
cargo clippy -p rpi-mcp-adapter --all-targets -- -D warnings

最后：代码层面的定位

从源码看，rpi-mcp-adapter 的定位可以概括成四句话：

它是 rpi 的 MCP Client 适配层，不是 MCP Server；
它把 stdio 和 Streamable HTTP 收敛成同一套初始化、发现和调用流程；
它以 host session 和配置身份缓存连接，并对 I/O 设置边界；
它只把 MCP Tools 稳定地接入 rpi，其他 MCP 原语目前明确不在范围内。

如果你要继续读源码，推荐顺序是：先看 src/lib.rs 的工具注册，再看 src/runtime.rs 的 with_session，然后看 src/connection.rs 的协议消息，最后分别进入 transport.rs 和 http.rs。

MCP 协议核对入口

本文关于 MCP 角色、生命周期、工具发现、工具结果和传输的协议语义，按以下版本文档核对：

这些规范描述的是 MCP 的通用能力；本文关于 rpi-mcp-adapter 的行为，仍以 rpi-mcp-adapter 0.2.1 源码为准。

代码入口

云服务器推荐

如果你需要部署博客、Docker 服务或个人项目，可以参考：

腾讯云特惠产品合集页阿里云云小站官方上云特惠

本文包含推广链接。通过链接购买不会增加你的费用，但我可能获得少量佣金。

云服务器推荐

评论