首页 › tutorials › 用Inspector调通带进度上报的MCP

概述

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

前置条件


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 里认出来:

客户端 → 服务器(带令牌的调用):

{
  "jsonrpc": "2.0", "id": 7, "method": "tools/call",
  "params": { "name": "index_directory", "arguments": { "path": "/data" },
              "_meta": { "progressToken": "t1" } }
}

服务器 → 客户端(进度通知,注意没有 id,这是通知而非请求):

{ "jsonrpc": "2.0", "method": "notifications/progress",
  "params": { "progressToken": "t1", "progress": 42, "total": 100, "message": "已处理42个" } }

服务器 → 客户端(日志通知):

{ "jsonrpc": "2.0", "method": "notifications/message",
  "params": { "level": "info", "logger": "dupsweep", "data": "开始索引 100 个文件" } }

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

Step 2:搭建项目骨架

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

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,并实时上报进度:

# 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:补齐业务工具

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

@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 指向你的服务器:

fastmcp dev server.py

方式 B——通用 npx 命令(任何 stdio 服务器都能调):

npx @modelcontextprotocol/inspector python server.py

方式 C——指定解释器(venv 场景推荐):避免因系统 Python 缺包而启动失败:

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。两种等价方式:

命令行注册(推荐,自动写绝对路径):

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

或手写项目级配置 .mcp.json(放在项目根目录,团队可共享):

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

注册后验证:

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 的进程架构。

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/>匹配到这次调用并渲染
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 字段放对象而非字符串)写入文件,可离线分析慢查询与失败调用。

训练营适用性评估

参考来源

  1. Progress — Model Context Protocol 官方规范 — 进度通知与 progressToken 的协议级定义,含报文格式与可选性说明
  2. MCP Inspector: Debug Your MCP Servers Like a Pro — Inspector 安装、连接、报文查看与 stdio 限制的完整操作指南
  3. Add Progress Bars and Logging Notifications to Your MCP Server — 进度条与日志通知的实战示例,含 Inspector 演示
  4. Progress Reporting — FastMCP 官方文档 — FastMCP ctx.report_progress 与日志 API 的用法与最佳实践
  5. MCP Inspector: Debug and Test Your MCP Servers (2026 Guide) — 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学通信知识星球海报-扫码加入