<!-- title: 用Inspector调通带进度上报的MCP -->

## 概述

本教程聚焦 MCP（Model Context Protocol）服务器开发中最容易被忽略、却决定长任务工具体验的两个能力：**进度通知（progress notifications）**与**日志通知（logging notifications）**。你将从零构建一个名为 `dupsweep` 的重复文件查找服务器，用 `ctx.report_progress` 实时回传扫描进度，用 `ctx.info` 推送运行日志，并全程用官方 **MCP Inspector** 抓取 JSON-RPC 报文完成调试，最后接入 Claude Code 验证。适合已跑通过基础 MCP 工具、想做出"会反馈、可观测"的高质量服务器的开发者。

## 前置条件

- 已安装 **Node.js 18+**（用于运行 Inspector，`npx` 即可）
- 已安装 **Python 3.10+**，并会用 `venv` 创建虚拟环境
- 了解 JSON-RPC 基本概念（请求/响应/通知）
- 已在 Claude Code 中成功连接过任意一个 MCP 服务器（若没有，可先跑官方 `npx @modelcontextprotocol/server-everything` 体验）
- 一台含较多文件的目录用于测试（如 `~/Documents`）

---

## Step 1：先搞懂进度上报的消息机制

绝大多数教程只教你"造一个返回 JSON 的工具"，但真实世界里很多任务——递归扫描目录、批量下载、训练评估——会持续几十秒甚至几分钟。如果工具在结束前一句话不说，客户端（和背后的模型）会以为它卡死了。MCP 的解法是一套**带回执的通知机制**，核心是 `progressToken`。

关键点：**进度不是服务器想发就发，而是"请求-回执"配对的。** 流程如下：

1. 客户端发起 `tools/call` 时，在 `params._meta.progressToken` 里塞一个令牌（字符串或整数）。
2. 服务器在执行过程中，用**同一个 token** 反向发送 `notifications/progress` 通知。
3. 客户端凭 token 把进度归类到对应的那次调用上。
4. 服务器还可以随时发 `notifications/message`（日志通知），按 `debug/info/warning/error` 分级。

对应的报文长这样，调试时一定要能在 Inspector 里认出来：

客户端 → 服务器（带令牌的调用）：
```json
{
  "jsonrpc": "2.0", "id": 7, "method": "tools/call",
  "params": { "name": "index_directory", "arguments": { "path": "/data" },
              "_meta": { "progressToken": "t1" } }
}
```

服务器 → 客户端（进度通知，注意没有 `id`，这是通知而非请求）：
```json
{ "jsonrpc": "2.0", "method": "notifications/progress",
  "params": { "progressToken": "t1", "progress": 42, "total": 100, "message": "已处理42个" } }
```

服务器 → 客户端（日志通知）：
```json
{ "jsonrpc": "2.0", "method": "notifications/message",
  "params": { "level": "info", "logger": "dupsweep", "data": "开始索引 100 个文件" } }
```

> **常见误区**：如果你在 Inspector 里调用工具却看不到进度条，多半是**客户端没传 token**。这是协议可选能力，并非所有客户端都实现。Inspector 默认会传，所以用它来验证服务器实现是最佳路径。

## Step 2：搭建项目骨架

新建项目目录并准备虚拟环境：

```bash
mkdir dupsweep && cd dupsweep
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "fastmcp>=2.3"
pip freeze > requirements.txt
```

FastMCP 是官方 SDK 的高层封装，它把"读取 token、回填 token、构造通知报文"这些机械活全包了，你只需要在工具函数里拿到 `Context` 对象，调用 `ctx.report_progress()` 和 `ctx.info()`。这正是本教程选 Python 的原因——同样的逻辑在底层 TypeScript SDK 里要手写十几行报文构造。

项目最终结构：
```
dupsweep/
├── server.py           # MCP 服务器主程序
├── .venv/
├── requirements.txt
└── .mcp.json           # 接入 Claude Code 用（后续 Step 6 创建）
```

## Step 3：写第一个带进度上报的工具

创建 `server.py`，先实现核心的 `index_directory`——它递归扫描目录、对每个文件算 SHA-256，并实时上报进度：

```python
# server.py
import asyncio
import hashlib
from pathlib import Path
from collections import defaultdict

from fastmcp import FastMCP, Context

mcp = FastMCP("dupsweep")

# 进程内状态：哈希值 -> 该哈希对应的文件路径列表
_index: dict[str, list[str]] = defaultdict(list)


def _hash_file(path: Path) -> str:
    """分块读取并计算 SHA-256，避免一次性把大文件读进内存。"""
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):  # 1MB 块
            h.update(chunk)
    return h.hexdigest()


@mcp.tool
async def index_directory(path: str, ctx: Context) -> str:
    """递归扫描目录并计算每个文件的 SHA-256 建立索引。
    大目录会通过进度通知实时回传处理进度。请先调用本工具，再调用 find_duplicates。"""
    root = Path(path).expanduser()
    if not root.is_dir():
        raise ValueError(f"不是有效目录: {path}")

    files = [p for p in root.rglob("*") if p.is_file()]
    total = len(files)
    if total == 0:
        return f"目录 {path} 中没有文件"

    _index.clear()
    # 日志通知：notifications/message，level=info
    await ctx.info(f"开始索引 {total} 个文件: {path}")

    for i, p in enumerate(files, start=1):
        try:
            # 用线程池跑磁盘 IO，避免阻塞事件循环（关键！）
            digest = await asyncio.to_thread(_hash_file, p)
            _index[digest].append(str(p))
        except OSError as e:
            await ctx.warning(f"跳过无权限文件 {p}: {e}")

        # 进度通知：notifications/progress，FastMCP 自动回填客户端传来的 token
        await ctx.report_progress(i, total)
        if i % 25 == 0:
            await ctx.info(f"已处理 {i}/{total}")

    dup_groups = [v for v in _index.values() if len(v) > 1]
    return f"索引完成: {total} 个文件, {len(dup_groups)} 组疑似重复"


if __name__ == "__main__":
    mcp.run()  # 默认启动 stdio 传输
```

两个细节决定了这段代码能不能在生产里用：

1. **`asyncio.to_thread` 必不可少**。`_hash_file` 是同步阻塞的磁盘 IO，若直接 `await` 不存在——你会在循环里阻塞整个事件循环，导致进度通知卡在缓冲区发不出去，Inspector 里看到"进度一次性全涌出来"。丢进线程池后，事件循环才有机会把通知真正推到客户端。
2. **`ctx.report_progress(i, total)` 是魔法所在**。FastMCP 会从 `ctx` 里取出当前请求携带的 `progressToken`，自动构造合规的 `notifications/progress` 报文。如果客户端没传 token，这个调用变成空操作（no-op），不会报错——这就是为什么进度功能是"优雅降级"的。

**预期结果**：在能跑的客户端里调用 `index_directory("/some/dir")`，会持续收到进度，最终返回汇总字符串。

> **常见问题**：报错 `AttributeError: 'Context' object has no attribute 'report_progress'` → FastMCP 版本太旧，升级到 `>=2.3`。报错 `cannot import name 'Context'` → 同样是版本问题，`Context` 在 2.x 才从顶层导出。

## Step 4：补齐业务工具

光有索引不够，再加一个查询工具复用进程内状态：

```python
@mcp.tool
async def find_duplicates(min_size_mb: float = 0.0) -> str:
    """返回所有重复文件分组。仅返回单文件大于 min_size_mb 的组，默认全部返回。
    必须先调用 index_directory 建立索引。"""
    if not _index:
        return "索引为空，请先调用 index_directory"

    threshold = int(min_size_mb * 1024 * 1024)
    groups = []
    for digest, paths in _index.items():
        if len(paths) < 2:
            continue
        # 用第一个文件的大小代表整组体积
        try:
            size = Path(paths[0]).stat().st_size
        except OSError:
            continue
        if size < threshold:
            continue
        groups.append((size, sorted(paths)))

    if not groups:
        return "没有发现重复文件"

    groups.sort(key=lambda x: -x[0])  # 按体积从大到小
    lines = []
    wasted = 0
    for size, paths in groups:
        wasted += size * (len(paths) - 1)
        size_str = f"{size / 1048576:.2f}MB"
        lines.append(f"[{size_str}] {' = '.join(paths)}")
    return (
        f"发现 {len(groups)} 组重复文件，"
        f"若去重最多可释放 {wasted / 1048576:.2f}MB:\n" + "\n".join(lines)
    )
```

注意 `find_duplicates` 不需要 `ctx`——它是个纯查询，不发任何通知。MCP 工具的参数中 `ctx` 是**可选注入**的：函数签名里有 `ctx: Context` 才会被注入，没有就正常调用。这让"纯计算工具"和"会反馈工具"可以混在一个服务器里。

至此服务器完成。运行 `python server.py` 会等待 stdio 输入（这是正常的，MCP 服务器是被客户端驱动的）。

## Step 5：用 MCP Inspector 调试（本教程的核心）

这是大多数教程跳过、却最该掌握的环节。**Inspector 是官方的可视化调试器**，能让你在不接 LLM 的情况下，逐条查看客户端与服务器之间的 JSON-RPC 报文，是定位"进度为什么不显示""工具为什么报错"的唯一可靠手段。

### 5.1 三种启动方式

**方式 A——FastMCP 内置（最省事）**：FastMCP 提供了 `dev` 命令，自动拉起 Inspector 指向你的服务器：

```bash
fastmcp dev server.py
```

**方式 B——通用 npx 命令（任何 stdio 服务器都能调）**：

```bash
npx @modelcontextprotocol/inspector python server.py
```

**方式 C——指定解释器（venv 场景推荐）**：避免因系统 Python 缺包而启动失败：

```bash
npx @modelcontextprotocol/inspector .venv/bin/python server.py
```

执行后，Inspector 会在终端打印一个本地网址（通常是 `http://localhost:6274`），用浏览器打开即可。

### 5.2 在 UI 里完成一次完整调试

打开 Web UI 后：

1. **左侧 Configuration**：确认 `Command` 与 `Args` 正确（如 `python` + `server.py`）。这里有个宝藏功能——**Environment Variables**，可以临时注入 `DUPSWEEP_DEBUG=1` 之类的变量而无需改代码。
2. 点击 **Connect**。若握手成功，右侧会显示服务器声明的 `tools` 列表：`index_directory`、`find_duplicates`。
3. 选中 `index_directory`，在 **Arguments** 面板填入 `path`（例如 `/home/you/Documents`），点 **Run Tool**。
4. **重点观察底部 "Notifications" 区**：你会看到进度条随扫描推进，以及逐条刷出的日志（`info: 已处理 25/320`）。这就是 Step 1 里那两类通知的可视化。
5. 切到 **Server → JSON-RPC** 标签页：能看到原始请求/响应/通知的完整报文，连 `progressToken` 的值（如 `"t1"`）和每次 `notifications/progress` 的 `progress/total` 都一览无余。**这是确认协议实现正确与否的金标准。**

### 5.3 Inspector 调试常见坑

| 现象 | 原因 | 解决 |
|------|------|------|
| Connect 一直转圈 | 服务器启动时报错（缺依赖） | 回到终端看 Inspector 打印的 stderr |
| 工具列表为空 | 服务器 `@mcp.tool` 装饰器没生效 | 确认用 `@mcp.tool` 而非 `@app.tool` |
| 进度通知完全不出现 | 客户端没传 token（Inspector 默认会传，此时多半是 SDK 版本旧） | 升级 FastMCP；核对报文里请求是否含 `_meta.progressToken` |
| 日志被吞 | 用了 `print()` 而非 `ctx.info()` | stdio 模式下 `print` 会污染 JSON-RPC 通道，必须改用 `ctx` 日志或 `logging` 输出到 stderr |
| 大目录扫描中断 | 客户端超时杀进程 | 在 Inspector 配置或后续 Claude Code 里调大超时 |

> **铁律**：stdio 传输下，**永远不要用 `print()` 向 stdout 打印调试信息**。stdout 是 JSON-RPC 专用通道，混入任意文本会让客户端解析失败。要输出日志就用 `ctx.info()`（会变成 `notifications/message`），要本地调试就配 `logging` 写到 **stderr**。

## Step 6：接入 Claude Code

调试通过后，把服务器注册给 Claude Code。两种等价方式：

**命令行注册（推荐，自动写绝对路径）**：

```bash
claude mcp add dupsweep -- /home/you/dupsweep/.venv/bin/python /home/you/dupsweep/server.py
```

**或手写项目级配置** `.mcp.json`（放在项目根目录，团队可共享）：

```json
{
  "mcpServers": {
    "dupsweep": {
      "command": "/home/you/dupsweep/.venv/bin/python",
      "args": ["/home/you/dupsweep/server.py"]
    }
  }
}
```

注册后验证：

```bash
claude mcp list              # 应看到 dupsweep
claude mcp get dupsweep      # 查看配置详情
```

> **关键陷阱**：`command` 务必用**虚拟环境的绝对路径**（如 `.venv/bin/python`），而不是裸 `python`。Claude Code 在非交互 shell 里启动子进程，未必继承你的 venv，裸 `python` 会因找不到 `fastmcp` 而静默失败，表现就是工具"死活连不上"。

## Step 7：在 Claude Code 里验证端到端

进入交互会话，让模型直接调用：

```
> 帮我扫描 ~/Downloads 目录，找出重复文件并按占用空间排序
```

模型会先调用 `index_directory`，再调用 `find_duplicates`。扫描过程中，终端会**实时打印进度条与日志**（Claude Code 会渲染收到的 `notifications/progress` 与 `notifications/message`）。最终你会看到分组列表和"可释放空间"汇总。

如果进度没显示但结果正常——回到 Step 5 用 Inspector 确认客户端是否传了 token。Claude Code 的 MCP 客户端支持进度渲染，正常情况下一定能看到。

---

## 完整工作流程

下面两张图分别说明**进度令牌如何在协议层流转**，以及**Inspector 的进程架构**。

```mermaid
sequenceDiagram
    autonumber
    participant C as 客户端<br/>(Inspector / Claude Code)
    participant S as MCP服务器<br/>(FastMCP dupsweep)
    C->>S: initialize + tools/list
    S-->>C: 服务器能力声明
    C->>S: tools/call(index_directory)<br/>_meta.progressToken = "t1"
    Note over S: 扫描目录、逐文件哈希
    loop 每处理一个文件
        S-->>C: notifications/progress<br/>{token:"t1", progress:i, total:N}
    end
    Note right of S: 每 25 个文件
    S-->>C: notifications/message<br/>{level:"info", data:"已处理 25/320"}
    S-->>C: tools/call 结果(成功响应, id=原请求id)
    Note over C: 用 token "t1" 把进度<br/>匹配到这次调用并渲染
```

```mermaid
flowchart LR
    subgraph UI["浏览器 :6274"]
        Web["Inspector Web UI<br/>工具调用 / 通知面板 / 报文查看"]
    end
    subgraph Node["Node 进程"]
        Proxy["Inspector Proxy :6275<br/>转发 JSON-RPC"]
    end
    subgraph Py["被调试的子进程"]
        Srv["FastMCP server.py<br/>(stdio 传输)"]
    end
    Web <-->|"HTTP/WebSocket"| Proxy
    Proxy <-->|"stdin/stdout<br/>(JSON-RPC over stdio)"| Srv
    Web -.->|"渲染 progress / message<br/>+ 原始报文"| Web
```

## 延伸阅读

1. **进阶到 HTTP/SSE 远程部署**：本教程用 stdio 本地调试；要把服务器挂到远程，需切换到 Streamable HTTP 传输，参见 MCP 规范的 *Transports* 一节（已覆盖的"MCP服务器远程部署实战"可衔接）。
2. **让工具"会问人"**：若扫描后想交互式确认删除哪些重复文件，可结合 Elicitation（采样由客户端向用户发问），实现"扫描-确认-清理"的人机闭环。
3. **多服务器组合**：真实项目常把"索引""查重""清理"拆成多个 MCP 服务器，用 Claude Code 的子代理按需调用，实现关注点分离。
4. **可观测性深化**：在 `notifications/message` 之外，结合结构化日志（`data` 字段放对象而非字符串）写入文件，可离线分析慢查询与失败调用。

## 训练营适用性评估

- **适合的训练营类型**：MCP 开发进阶营、AI 工程师工具构建营、可观测性专题工作坊。不适合纯入门营（需先有基础 MCP 工具开发经验）。
- **建议教学时长**：3.5–4 小时（含上机）。其中概念 30 分钟、编码 60 分钟、Inspector 调试 75 分钟、接入与验证 45 分钟、答疑与拓展 30 分钟。
- **学员产出验收标准**：能独立用 Inspector 抓到一次完整的 `progressToken` 往返报文，并在 Claude Code 里看到进度条实时滚动。
- **可裁剪方案**：若时间紧张，可省略 `find_duplicates` 与 `.mcp.json` 团队配置，只保留"写工具→Inspector 调试→命令行注册"主线，压缩到 2 小时。

## 参考来源

1. [Progress — Model Context Protocol 官方规范](https://modelcontextprotocol.io/specification/2025-03-26/basic/utilities/progress) — 进度通知与 `progressToken` 的协议级定义，含报文格式与可选性说明
2. [MCP Inspector: Debug Your MCP Servers Like a Pro](https://orchestrator.dev/blog/2026-05-12-mcp-inspector/) — Inspector 安装、连接、报文查看与 stdio 限制的完整操作指南
3. [Add Progress Bars and Logging Notifications to Your MCP Server](https://zazencodes.substack.com/p/add-progress-bars-and-logging-notifications) — 进度条与日志通知的实战示例，含 Inspector 演示
4. [Progress Reporting — FastMCP 官方文档](https://gofastmcp.com/servers/progress) — FastMCP `ctx.report_progress` 与日志 API 的用法与最佳实践
5. [MCP Inspector: Debug and Test Your MCP Servers (2026 Guide)](https://apigene.ai/blog/mcp-inspector) — Inspector 2026 版工作流、环境变量注入与超时处理

> [!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学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)