首页 › tutorials › Rust构建零依赖MCP服务器

概述

本教程带你用 Rust 官方 MCP SDK rmcp 从零构建一个可被 Claude Code 直接调用的工具服务器。你将掌握 #[tool] 宏定义工具、JsonSchema 自动生成参数契约、ServerHandler 身份声明、stdio 传输层接入，以及单元测试与协议级联调。最终产物是一个单文件静态二进制——没有 node_modules，没有虚拟环境，丢到任意机器即可运行。适合已有任意一门语言基础、想用 Rust 给 AI 助手"装手脚"的开发者。

前置条件

为什么是 Rust：MCP 服务器的隐藏成本

当你在 Python/TypeScript 里写 MCP 服务器时，分发它意味着附带一份运行时：Python 要带 venv 和依赖锁，Node 要带 node_modules（动辄数百 MB）。而 MCP 服务器本质是长期常驻的子进程，它夹在 AI 助手和你的系统之间，频繁做文件 I/O、网络调用甚至敏感操作。

Rust 在这里有三重优势：

  1. 单一静态二进制：cargo build --release 产出一个几 MB 的可执行文件，零运行时依赖，跨机部署只需 scp。
  2. 内存安全：没有 GC 停顿，没有空指针/数据竞争，处理文件和网络时更可靠。
  3. 类型即契约：编译期就把"工具接受什么参数、返回什么"固定下来，错误在到达 AI 之前就被编译器拦截。

rmcp 是 MCP 官方维护的 Rust SDK，截至 2026 年初在 crates.io 累计下载超 470 万次，已发布到 1.x 稳定版（docs.rs 当前为 1.7.0），宏驱动 API 让定义工具像写普通方法一样自然。

MCP 协议速览：客户端、服务器与传输层

MCP（Model Context Protocol）是 Anthropic 提出的开放标准，基于 JSON-RPC 2.0，规定 AI 应用（客户端，如 Claude Code）如何与外部能力提供方（服务器）通信。架构借鉴了编辑器里的语言服务器协议（LSP）。

一个 MCP 服务器可暴露三类能力：

原语 作用 典型场景
Tools AI 可调用的函数 "统计这个目录的代码行数"
Resources AI 可读取的数据 读取配置文件、数据库记录
Prompts 可复用的提示词模板 标准化的代码审查模板

本教程聚焦 Tools——它最直接有用。通信发生在传输层之上，rmcp 内置两套：

致命细节：stdio 传输下，永远不要在服务器里用 println!()。stdout 是协议通道，任何一行调试输出都会污染 JSON-RPC 流，导致客户端静默断开。所有日志必须走 stderr（eprintln! 或 tracing 框架）。

第一步：创建项目与依赖

操作说明

cargo new dupfind
cd dupfind

编辑 Cargo.toml：

[package]
name = "dupfind"
version = "0.1.0"
edition = "2021"

[dependencies]
rmcp = { version = "0.16", features = ["server", "transport-io", "macros"] }
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
schemars = "0.8"
anyhow = "1"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

各依赖的作用：

预期结果

cargo build
#    Compiling dupfind v0.1.0
#    Compiling rmcp v0.16.x ...
#     Finished dev [unoptimized + debuginfo] target(s)

常见问题

第二步：定义服务器结构体与第一个工具

我们的实战项目是 dupfind——重复文件检测器，给 Claude Code 两个工具：按内容找出重复文件、按扩展名统计磁盘占用。这类"扫目录+比内容"的任务正是 Rust 的强项。

操作说明

编辑 src/main.rs，先放导入和服务器结构体：

use anyhow::Result;
use rmcp::handler::server::wrapper::Json;
use rmcp::model::ServerInfo;
use rmcp::{tool, ServerHandler};
use schemars::JsonSchema;
use serde::Deserialize;
use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use tracing::info;

#[derive(Debug, Clone)]
pub struct DupFindServer;

结构体刻意保持无状态——每次工具调用自包含。若你的服务器需要维护连接或缓存，就在这里加字段。

接着定义第一个工具的输入类型。给字段加 JsonSchema 派生，客户端就能看到参数描述：

#[derive(Debug, Deserialize, JsonSchema)]
pub struct FindDuplicatesInput {
    /// 要扫描的目录绝对路径
    pub path: String,
    /// 只统计大于此字节数的文件，默认 0（不限制）
    pub min_size: Option<u64>,
}

字段上的文档注释会变成 Claude Code 看到的参数描述。好的描述能帮 AI 判断何时、如何调用你的工具——这是工具设计里最容易被忽视、却最影响触发准确率的一环。

然后实现工具本身，用 rmcp 的宏系统：

#[tool(tool_box)]
impl DupFindServer {
    #[tool(description = "扫描目录，按文件内容找出重复文件，并报告可回收的磁盘空间")]
    pub async fn find_duplicates(
        &self,
        #[tool(aggr)] input: Json<FindDuplicatesInput>,
    ) -> Result<String, anyhow::Error> {
        let root = PathBuf::from(&input.path);
        if !root.exists() {
            return Ok(format!("错误：路径 '{}' 不存在", input.path));
        }
        if !root.is_dir() {
            return Ok(format!("错误：路径 '{}' 不是目录", input.path));
        }
        let min_size = input.min_size.unwrap_or(0);

        // 第一遍：按文件大小分桶（大小不同必然内容不同）
        let mut by_size: HashMap<u64, Vec<PathBuf>> = HashMap::new();
        index_by_size(&root, min_size, &mut by_size)?;

        // 第二遍：对同大小文件比较完整内容
        let mut wasted: u64 = 0;
        let mut report = String::from("重复文件报告：\n\n");
        let mut group_no = 0;
        for (_size, files) in by_size.iter().filter(|(_, v)| v.len() > 1) {
            let groups = group_identical(files);
            for group in groups {
                if group.len() > 1 {
                    group_no += 1;
                    let keep = &group[0];
                    report.push_str(&format!(
                        "组 #{}：保留 {}\n",
                        group_no,
                        keep.display()
                    ));
                    for dup in &group[1..] {
                        let bytes = fs::metadata(dup)?.len();
                        wasted += bytes;
                        report.push_str(&format!("  重复 {} ({})\n", dup.display(), human(bytes)));
                    }
                    report.push('\n');
                }
            }
        }
        if group_no == 0 {
            report.push_str("未发现重复文件。");
        } else {
            report.push_str(&format!("共 {} 组重复，可回收 {}", group_no, human(wasted)));
        }
        Ok(report)
    }
}

几个要点：

补上辅助函数。index_by_size 递归遍历并跳过常见无意义目录：

fn index_by_size(dir: &Path, min_size: u64, out: &mut HashMap<u64, Vec<PathBuf>>) -> Result<()> {
    for entry in fs::read_dir(dir)? {
        let path = entry?.path();
        let name = path.file_name().unwrap_or_default().to_string_lossy().to_string();
        if path.is_dir() {
            // 跳过隐藏目录和构建产物——把领域知识固化进工具
            if name.starts_with('.') || matches!(name.as_str(), "target" | "node_modules" | "vendor" | ".git") {
                continue;
            }
            index_by_size(&path, min_size, out)?;
        } else {
            let bytes = fs::metadata(&path)?.len();
            if bytes >= min_size && bytes > 0 {
                out.entry(bytes).or_default().push(path);
            }
        }
    }
    Ok(())
}

fn group_identical(files: &[PathBuf]) -> Vec<Vec<PathBuf>> {
    // 逐字节比较，把内容相同的文件归到一组
    let mut groups: Vec<(Vec<u8>, PathBuf)> = Vec::new();
    let mut result: Vec<Vec<PathBuf>> = Vec::new();
    for f in files {
        let content = fs::read(f).unwrap_or_default();
        if let Some((_, idx)) = groups.iter().zip(0..).find(|((c, _), _)| *c == content) {
            result[idx].push(f.clone());
        } else {
            groups.push((content.clone(), f.clone()));
            result.push(vec![f.clone()]);
        }
    }
    result
}

fn human(bytes: u64) -> String {
    if bytes >= 1_048_576 {
        format!("{:.1} MB", bytes as f64 / 1_048_576.0)
    } else if bytes >= 1024 {
        format!("{:.1} KB", bytes as f64 / 1024.0)
    } else {
        format!("{} B", bytes)
    }
}

设计哲学：注意我们跳过了 target、node_modules、.git。这是"把领域知识固化进工具"——AI 不必每次判断该跳过哪些目录，工具替它处理好。MCP 工具不是越通用越好，而是有主见、让正确的事变简单。AI 是你的用户，它最喜欢"清晰、结构化、上下文丰富"的回复。

第三步：实现 ServerHandler 特征

rmcp 要求在服务器结构体上实现 ServerHandler 特征，声明身份与能力：

#[tool(tool_box)]
impl ServerHandler for DupFindServer {
    fn get_info(&self) -> ServerInfo {
        ServerInfo {
            instructions: "重复文件检测服务器。用 find_duplicates 扫描目录找出内容重复的文件；"
                .into(),
            ..Default::default()
        }
    }
}

instructions() 返回的描述帮助 AI 理解服务器用途与使用时机。这个 impl 块上的 #[tool(tool_box)] 告诉 rmcp 自动把前面定义的工具装配进来。

第四步：main 函数与传输层启动

#[tokio::main]
async fn main() -> Result<()> {
    // 关键：日志写 stderr，绝不能碰 stdout
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::from_default_env()
                .add_directive("dupfind=info".parse()?),
        )
        .with_writer(std::io::stderr)
        .init();

    info!("启动 dupfind MCP 服务器");
    let server = DupFindServer;
    let transport = rmcp::transport::io::stdio();
    let handle = server.serve(transport).await?;
    handle.waiting().await?;
    Ok(())
}

执行流程：先配置 tracing_subscriber 写向 stderr（对 stdio 服务器非可选，否则协议必坏）；from_default_env() 让你用 RUST_LOG 环境变量控制日志级别；然后创建 stdio 传输，serve() 启动 JSON-RPC 消息循环，waiting() 阻塞直到客户端断开。

第五步：本地构建与冒烟测试

操作说明

cargo build --release
# 二进制位于 target/release/dupfind

手工发一条 JSON-RPC 初始化消息验证握手：

echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"0.1.0"}}}' \
  | ./target/release/dupfind

预期结果

应返回包含服务器能力与工具列表的 JSON 响应。若什么都没有或报错，检查是否编译干净、是否在别处误写了 stdout。

常见问题：服务器在 Claude Code 里不出现

编译通过却连不上，按这个顺序排查（每条都是 rmcp 新手真实踩坑）：

  1. stdout 污染：最常见。任何 println!/dbg! 或写 stdout 的第三方库都会破坏 stdio 传输。审计所有依赖，把日志统一走 tracing_subscriber::fmt().with_writer(std::io::stderr)。
  2. .mcp.json 路径非绝对：Claude Code 从自己的工作目录启动服务器，相对路径会静默失败。永远用 target/release/dupfind 的绝对路径。
  3. 启动即崩溃：直接运行二进制发初始化消息，看 stderr： bash RUST_LOG=dupfind=debug ./target/release/dupfind 2>/tmp/mcp.log <<'JSON' {"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"0.1.0"}}} JSON cat /tmp/mcp.log
  4. 注册作用域错：claude mcp add 默认注册到 local 作用域（绑定当前目录）。要在项目内全局可用，用仓库根目录的 .mcp.json。
  5. Claude Code 缓存旧定义：改完 .mcp.json 或重编译二进制后，重启 Claude Code——MCP 服务器列表每会话只初始化一次。
  6. 协议版本不匹配：用 rmcp = "0.16" 或更新版本，cargo update -p rmcp 升级。

第六步：接入 Claude Code

两种方式。CLI 方式（快速测试）：

claude mcp add --transport stdio dupfind -- /absolute/path/to/dupfind/target/release/dupfind

.mcp.json 方式（团队共享，推荐提交到版本库）。在项目根创建：

{
  "mcpServers": {
    "dupfind": {
      "command": "/absolute/path/to/dupfind/target/release/dupfind"
    }
  }
}

验证注册：

claude mcp list   # 应看到 dupfind

预期结果

在项目目录打开 Claude Code，问："帮我找一下 ~/Downloads 里有没有重复文件。" Claude Code 会自动发现 find_duplicates 工具并调用。你会切身体会到——无需安装任何运行时，这点对用惯 Node/Python MCP 的人格外舒适。

第七步：扩展第二个工具

一个工具有用，多个能协作的工具才强大。再加一个按扩展名统计磁盘占用的工具，补到同一个 #[tool(tool_box)] impl DupFindServer 块里：

#[derive(Debug, Deserialize, JsonSchema)]
pub struct ExtensionSizesInput {
    /// 要分析的目录路径
    pub path: String,
    /// 最多返回多少种扩展名，默认 15
    pub limit: Option<u32>,
}

#[tool(description = "按文件扩展名统计目录的磁盘占用，从大到小排序")]
pub async fn extension_sizes(
    &self,
    #[tool(aggr)] input: Json<ExtensionSizesInput>,
) -> Result<String, anyhow::Error> {
    let root = PathBuf::from(&input.path);
    if !root.is_dir() {
        return Ok(format!("错误：'{}' 不是有效目录", input.path));
    }
    let limit = input.limit.unwrap_or(15) as usize;
    let mut stats: HashMap<String, (u64, u64)> = HashMap::new(); // ext -> (bytes, count)
    collect_ext_sizes(&root, &mut stats)?;
    let mut rows: Vec<_> = stats.into_iter().collect();
    rows.sort_by(|a, b| b.1 .0.cmp(&a.1 .0));
    rows.truncate(limit);
    let total: u64 = rows.iter().map(|(_, v)| v.0).sum();
    let mut out = format!("按扩展名的磁盘占用（前 {}）：\n\n", rows.len());
    for (ext, (bytes, count)) in &rows {
        let label = if ext.is_empty() { "(无扩展名)" } else { ext };
        out.push_str(&format!("{:<10} {:>10}  {} 个文件\n", label, human(*bytes), count));
    }
    out.push_str(&format!("\n合计 {}", human(total)));
    Ok(out)
}

补上遍历函数：

fn collect_ext_sizes(dir: &Path, stats: &mut HashMap<String, (u64, u64)>) -> Result<()> {
    for entry in fs::read_dir(dir)? {
        let path = entry?.path();
        let name = path.file_name().unwrap_or_default().to_string_lossy().to_string();
        if path.is_dir() {
            if name.starts_with('.') || matches!(name.as_str(), "target" | "node_modules" | ".git") {
                continue;
            }
            collect_ext_sizes(&path, stats)?;
        } else {
            let bytes = fs::metadata(&path)?.len();
            let ext = path.extension().map(|e| e.to_string_lossy().to_string()).unwrap_or_default();
            let e = stats.entry(ext).or_insert((0, 0));
            e.0 += bytes;
            e.1 += 1;
        }
    }
    Ok(())
}

重编译 cargo build --release，Claude Code 下次初始化时会自动发现新工具。现在三个能力自然组合：先用 extension_sizes 看哪种扩展名最占空间，再用 find_duplicates 在其中找重复。

实战案例：完整可运行的 dupfind

把上述片段拼起来，整个服务器约 220 行 Rust。完整目录结构：

dupfind/
├── Cargo.toml
└── src/
    └── main.rs

端到端验证流程：

# 1. 编译
cargo build --release

# 2. 准备测试目录（造两个内容相同的文件）
mkdir -p /tmp/dupdemo/sub
echo "important data" > /tmp/dupdemo/a.txt
echo "important data" > /tmp/dupdemo/b.txt
echo "different"      > /tmp/dupdemo/sub/c.txt

# 3. 注册到 Claude Code
claude mcp add --transport stdio dupfind -- $(pwd)/target/release/dupfind

# 4. 在 Claude Code 里提问
# "分析 /tmp/dupdemo，找出重复文件并按扩展名统计占用"

预期结果：find_duplicates 报告 a.txt 与 b.txt 重复、可回收约 14 B；extension_sizes 显示 .txt 合计占用。

工作流程图

flowchart TD
    A[Claude Code 客户端] -->|spawn 子进程 + stdin/stdout| B[dupfind 二进制]
    B --> C{rmcp 消息循环}
    C -->|tools/list| D[返回 JsonSchema 工具契约]
    A -->|看到参数描述| A2[决定调用 find_duplicates]
    A2 -->|JSON-RPC tools/call| C
    C --> E[index_by_size 按大小分桶]
    E --> F[group_identical 逐字节比对]
    F --> G[组装可读报告]
    G -->|stdout JSON-RPC 响应| A
    H[tracing 日志] -.->|stderr 不污染协议| I[/tmp/mcp.log]
    style A fill:#e1f5ff
    style B fill:#fff4e1
    style G fill:#e8f5e9

单元测试与协议级测试

MCP 工具本质是结构体上的 async 方法，无需启动服务器或连客户端就能测。

在 src/main.rs 末尾加：

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs;
    use tempfile::TempDir;

    #[tokio::test]
    async fn test_find_duplicates_detects_identical() {
        let dir = TempDir::new().unwrap();
        fs::write(dir.path().join("a.txt"), "hello").unwrap();
        fs::write(dir.path().join("b.txt"), "hello").unwrap(); // 重复
        fs::write(dir.path().join("c.txt"), "world").unwrap();
        let server = DupFindServer;
        let input = Json(FindDuplicatesInput {
            path: dir.path().to_string_lossy().to_string(),
            min_size: None,
        });
        let result = server.find_duplicates(input).await.unwrap();
        assert!(result.contains("可回收"));
        assert!(result.contains("1 组"));
    }

    #[tokio::test]
    async fn test_invalid_path_returns_friendly_error() {
        let server = DupFindServer;
        let input = Json(FindDuplicatesInput {
            path: "/no/such/path".into(),
            min_size: None,
        });
        let result = server.find_duplicates(input).await.unwrap();
        assert!(result.contains("不存在"));
    }
}

Cargo.toml 加开发依赖：

[dev-dependencies]
tempfile = "3"

运行 cargo test。注意一个关键设计选择：工具返回 Ok(String) 里夹带错误信息，而不是用 ? 上抛 Err。这是因为——返回 Err 客户端看到的是协议级错误；返回 Ok 带错误文案，AI 能读懂并智能反应（换个路径、向你澄清、解释问题）。把 Err 留给真正不可恢复的情况。

协议级集成测试则直接喂二进制原始 JSON-RPC：

#!/bin/bash
# test_integration.sh
BIN="./target/release/dupfind"
INIT='{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"t","version":"0.1.0"}}}'
LIST='{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}'
RESP=$(echo -e "${INIT}\n${LIST}" | $BIN 2>/dev/null)
echo "$RESP" | grep -q "find_duplicates"   && echo "PASS: find_duplicates" || echo "FAIL"
echo "$RESP" | grep -q "extension_sizes"   && echo "PASS: extension_sizes" || echo "FAIL"

单元测试能过、连 Claude Code 却失败，问题几乎总在传输层。开详细日志盯 stderr：

RUST_LOG=rmcp=debug,dupfind=trace ./target/release/dupfind 2>/tmp/debug.log

rmcp=debug 会打印收发的每一条 JSON-RPC，dupfind=trace 是应用日志，两者叠加能精确定位通信断点。

生产化与进阶

延伸阅读

训练营适用性评估

参考来源

  1. Build an MCP Server in Rust with rmcp and Claude Code — systemprompt.io — 完整的 rmcp 实战教程，含 #[tool] 宏、ServerHandler、stdio 传输、测试与排错，本文宏语法与 code-stats 示例架构的主要参考。
  2. rmcp — Docs.rs — rmcp 官方 API 文档（当前 1.7.0），提供 feature flags、传输层（stdio / Streamable HTTP）、ServerHandler/ServiceExt 等模块的权威说明。
  3. How to Build a stdio MCP Server in Rust — Shuttle — Rust 构建 stdio MCP 服务器的实操指南，佐证传输层与生命周期细节。
  4. GitHub: modelcontextprotocol/rust-sdk — rmcp 官方仓库（tokio 异步运行时，迁移至 1.x），含示例代码与 README 进阶特性指引。
  5. Building a High-Performance MCP Server with Rust — Medium — Rust 高性能 MCP 服务器的完整实现指南，补充性能与工程化视角。

[!tip] 51学通信知识星球 星主：爱卫生（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。 星球定位：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。 当前规模：5200 篇主题 / 990 个回答，198 元/年。 加入福利：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。 AI 加持：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

51学通信知识星球海报-扫码加入

📎 本分类最新文章