Proposal: 为 aquaregia 引入 MCP (Model Context Protocol) client 支持

[minorcell]

Proposal: 为 aquaregia 引入 MCP (Model Context Protocol) client 支持

1. Summary

新增一个默认关闭的 mcp feature,在 core crate 内提供一个薄适配器:连接 MCP server → 拉取其 tools → 转成 aquaregia Tool,直接喂给现有 .tools()。协议、传输、握手全部委托官方 SDK rmcp,我们只做 rmcp::model::Tool ↔ aquaregia::Tool 的双向映射。对不使用 MCP 的用户零成本。

2. User Stories / Motivation

• 接入现成生态工具:用户想让 agent 使用 filesystem / GitHub / Postgres 等已有 MCP server 暴露的工具,而不是为每个工具手写 tool().execute_raw()。
• 换 server 不改代码:用户切换或新增一个 MCP server,agent 的工具集应随之变化,无需改动业务代码。
• 动态工具集:某些 MCP server 在运行期通过 notifications/tools/list_changed 变更工具列表,用户希望 agent 能感知。

共同需求:把"一个 MCP server 暴露的一组 tools"作为一等公民接入 agent 的工具循环,且不侵入 core 的轻量定位。

3. Current Workaround

今天用户有两条路,都有明显冗余:

1. 为每个 MCP 工具手写 aquaregia 工具:自己起 rmcp 连接,list_tools 后对每个工具手写 tool(name).raw_schema(schema).execute_raw(...),并在闭包里手动 call_tool + 拍平 content。每个 server、每次工具变更都要重写。
2. 完全绕过 aquaregia 的工具循环,自己在 rmcp 层驱动。失去 agent hooks / error policy / prepare_step 等能力。

两者都要求用户重复实现"MCP content → aquaregia 结果"的映射,这正是本提案要下沉的冗余。

4. Goals

• 提供 aquaregia::mcp(feature-gated)把一个 MCP server 的 tools 一次性转成 Vec<Tool>。
• 支持两种传输:stdio/子进程(本地 server)与 Streamable HTTP(远程 server)。
• MCP 工具在 agent 眼里与手写工具完全一致——复用现有 ToolExecutor / 工具循环 / hooks / error policy,不引入平行机制。
• 通过现有 prepare_step 表达"工具热更新",不新增 actor / 注册中心。
• 默认关闭;开启后依赖被隔离,不污染核心用户。

5. Out of Scope

• MCP server 侧(把 aquaregia 工具反向暴露为 MCP server)。
• prompts / resources / elicitation / sampling:本提案只做 tools。
• 多 MCP server 的聚合调度 / 命名空间冲突自动消解:用户可自行合并多个 Vec<Tool>,重名由现有 ToolRegistry 报错(见 §7)。
• 自定义 rmcp ClientHandler 全量能力:只暴露工具桥接所需的最小面。

6. Proposal

6.1 Design Rule

一个 MCP 工具就是一个 aquaregia::Tool。适配器只负责:①rmcp::model::Tool 的 name/description/inputSchema → ToolDescriptor;②在 ToolExecutor::execute 内把 JSON 参数交给 rmcp 的 call_tool,并把返回的 content 映射成 serde_json::Value。不新增任何面向用户的工具抽象。

这条规则成立的前提(现状已满足):ToolExecutor 已是对象安全的 async fn execute(&self, Value) -> Result<Value, ToolExecError>,与 MCP 的 call_tool(args) -> content 同构。

6.2 API / Interface

// Cargo.toml
// aquaregia = { version = "...", features = ["mcp"] }

#[cfg(feature = "mcp")]
pub mod mcp {
    use crate::tool::Tool;

    /// 启动本地 MCP server 子进程并完成握手(stdio 传输)。
    pub async fn connect_stdio(command: Command) -> Result<McpConnection, McpError>;

    /// 连接远程 MCP server(Streamable HTTP 传输)。
    pub async fn connect_http(url: impl Into<String>) -> Result<McpConnection, McpError>;

    /// 一个活动的 MCP 连接。**必须在使用其工具期间保持存活**:
    /// 连接被 drop 后,由它产出的工具调用将返回 `McpError::Disconnected`。
    pub struct McpConnection { /* 持有 rmcp RunningService + peer sink */ }

    impl McpConnection {
        /// 拉取 server 当前暴露的全部工具,转成 aquaregia `Tool`。
        pub async fn tools(&self) -> Result<Vec<Tool>, McpError>;

        /// 每次调用的超时(默认 300s;`None` 关闭)。防止传输层丢包导致 agent 永久挂起。
        pub fn with_call_timeout(self, timeout: impl Into<Option<Duration>>) -> Self;
    }

    #[derive(Debug, thiserror::Error)]
    pub enum McpError {
        Connect(String),      // 握手/传输建立失败
        ListTools(String),    // 拉取工具列表失败
        Disconnected,         // 连接已关闭
        CallTimeout,          // 单次调用超时
    }
}

所有权 / 生命周期:tools() 返回的每个 Tool 内部持有连接 sink 的克隆;真正驱动连接的 RunningService 归 McpConnection 所有。调用方须让 McpConnection 活得和 agent 一样久(见 §7 断连行为)。

6.3 Examples as Specification

// ── 现状(workaround):每个工具手写 + 手动 call_tool + 手动拍平 ──
let service = ().serve(TokioChildProcess::new(Command::new("mcp-server-fs"))?).await?;
let peer = service.peer().clone();
let mut tools = Vec::new();
for t in peer.list_all_tools().await? {
    let peer = peer.clone();
    let name = t.name.to_string();
    tools.push(
        aquaregia::tool(name.clone())
            .description(t.description.unwrap_or_default())
            .raw_schema(serde_json::to_value(&t.input_schema)?)
            .execute_raw(move |args| {
                let peer = peer.clone(); let name = name.clone();
                async move { /* call_tool + 拍平 content + is_error 处理 */ }
            }),
    );
}

// ── 提案(proposed form):三行 ──
let mcp = aquaregia::mcp::connect_stdio(Command::new("mcp-server-fs")).await?;
let agent = client.agent("gpt-5.5")
    .tools(mcp.tools().await?)
    .build()?;
// 注意:`mcp` 需与 `agent` 同生命周期。

// ── 等价形式(desugared):证明无需新增 core 原语 ──
// `mcp.tools()` 内部对每个工具做的,正是上面 workaround 里 `tool(..).execute_raw(..)` 那段;
// 适配器只是把它下沉进库,并统一 content 映射与超时。

热更新(用现有 prepare_step,非新机制):

let agent = client.agent("gpt-5.5")
    .prepare_step(move |_ctx| {
        let mcp = mcp.clone();
        async move { StepOverrides::new().tools(mcp.tools().await?) } // 每步重新拉取
    })
    .build()?;

6.4 Boundary Cases(content 映射为规范)

MCP CallToolResult.content → ToolExecutor 的 Value,规则如下:

MCP content	映射为 aquaregia 工具结果
单个 text 块	折叠成该字符串(JSON string),最符合模型预期
多个块 / 非纯文本	[{ "type": "text"/"image"/"resource", ... }] 的 JSON 数组,保留结构
image	{ "type": "image", "mime_type", "data"(base64) }
resource	{ "type": "resource", "uri", "mime_type", "text"/"blob" }
audio	首版返回 McpError 映射的工具错误(不支持,与 Rig 对齐);后续再定
is_error: true	ToolExecError::Execution(拼接的文本),交回工具循环由 error policy 处置

7. Error Handling

Condition	Behavior
server 进程启动失败 / HTTP 不可达	connect_* 返回 McpError::Connect
list_tools 失败	tools() 返回 McpError::ListTools
拉取到的工具名非法(超 64 字符 / 非法字符)	沿用现有 ToolRegistry 校验,agent 构建期报错
多个 server 工具重名	现有 ToolRegistry::from_tools 返回 duplicate 错误(不自动加前缀,见 §5)
McpConnection 已 drop 后仍调用工具	工具调用返回 ToolExecError::Execution(源自 McpError::Disconnected)
单次 call_tool 超过超时	ToolExecError(源自 McpError::CallTimeout),loop 可恢复;调用在本地放弃,不向 server 发取消

8. Compatibility

纯增量、opt-in。新增 mcp feature(默认关闭)、新增 aquaregia::mcp 模块。不改动任何现有类型或签名。未开启 feature 时依赖树与行为完全不变。

依赖(rmcp 1.7,default-features = false + ["client", "transport-child-process"],HTTP 再加 transport-streamable-http-client):与现有 async-trait/reqwest/serde/tokio/tokio-util/thiserror/base64/futures 重叠;新增 chrono/uuid/rand/tower-service/sse-stream/tokio-stream/http*/tracing/pastey/pin-project-lite 及 proc-macro 链。两处版本重复需知悉(均被 feature 隔离):rmcp 拉 schemars 1.2(本库用 0.8)与 reqwest 0.13(本库用 0.12)。

Phasing(交付顺序,非各自独立提案):

• P1:mcp feature + connect_stdio + tools() + content 映射 + 调用超时。
• P2:connect_http(Streamable HTTP)。
• P3(按需):_meta 透传(auth/session 走 MCP _meta,不暴露给模型)。

9. Alternatives Considered

9.1 独立 crate aquaregia-mcp

绝对隔离 core 依赖,但适配器仅 ~150 行、且完全可用现有 public builder 表达,单独成 crate 只增加打包与双 crate 使用成本。feature-gate 已能达到"默认零成本"的同等效果,故选 feature-gate。

9.2 自研 MCP 协议实现

MCP spec 仍在演进,自研需持续追平握手/传输/通知,收益为负。官方 rmcp 是事实标准,Rig 等库均薄封装之。

9.3 引入 ToolServer actor 做热更新(Rig 的做法)

Rig 为 tools/list_changed 引入了 actor + ClientHandler。但 aquaregia 的 prepare_step 已能每步重建工具集(§6.3),用现有机制即可表达,无需新增并发原语。

10. Testing Strategy

Test case	Method
tool 转换正确	起一个内存/本地 mock MCP server,断言 tools() 的 name/description/schema 与源一致
正常调用	agent 调用 MCP 工具,断言 content 按 §6.4 映射
is_error	mock server 返回 error result,断言进入 error policy 路径
超时	mock 一个不响应的工具,断言 CallTimeout 且 loop 可继续
断连	drop McpConnection 后调用,断言返回 disconnected 错误而非 panic/hang
重名	两个 server 含同名工具,断言 ToolRegistry 构建期报错
feature off	不开 mcp 时编译通过、依赖树不含 rmcp(CI 校验)

11. Summary of Changes

Area	Change
Cargo.toml	新增可选依赖 rmcp;新增 mcp feature
src/mcp/ (新增)	connect_stdio / connect_http / McpConnection / McpError / content 映射 / McpToolExecutor: ToolExecutor
src/lib.rs	#[cfg(feature = "mcp")] pub mod mcp;
docs / examples	一个 examples/mcp 演示接入本地 MCP server
tests	§10 用例

12. Open Questions

1. 传输交付顺序:P1 先做 stdio(本地 server),还是远程 HTTP 更急?(影响首个可用版本)
2. 多块 content 的折叠策略:单 text 折叠成字符串是否符合各 provider 的工具结果预期,还是一律返回结构化数组?
3. 是否借此把本库 reqwest 0.12→0.13、schemars 0.8→1.x 对齐以消除双版本?(独立议题,可另开提案)
4. audio content 首版报错是否可接受,还是需要一个降级为占位文本的选项?

---

本提案基于对 rmcp 1.7.0 feature/API、Rig rig-core 的 MCP 适配实现,以及 aquaregia 当前 tool.rs / client.rs 工具体系的实际代码调研。