本教程聚焦 MCP(Model Context Protocol)服务器开发中最容易被忽略、却决定长任务工具体验的两个能力:进度通知(progress notifications)与日志通知(logging notifications)。你将从零构建一个名为 dupsweep 的重复文件查找服务器,用 ctx.report_progress 实时回传扫描进度,用 ctx.info 推送运行日志,并全程用官方 MCP Inspector 抓取 JSON-RPC 报文完成调试,最后接入 Claude Code 验证。适合已跑通过基础 MCP 工具、想做出"会反馈、可观测"的高质量服务器的开发者。
npx 即可)venv 创建虚拟环境npx @modelcontextprotocol/server-everything 体验)~/Documents)绝大多数教程只教你"造一个返回 JSON 的工具",但真实世界里很多任务——递归扫描目录、批量下载、训练评估——会持续几十秒甚至几分钟。如果工具在结束前一句话不说,客户端(和背后的模型)会以为它卡死了。MCP 的解法是一套带回执的通知机制,核心是 progressToken。
关键点:进度不是服务器想发就发,而是"请求-回执"配对的。 流程如下:
tools/call 时,在 params._meta.progressToken 里塞一个令牌(字符串或整数)。notifications/progress 通知。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 默认会传,所以用它来验证服务器实现是最佳路径。
新建项目目录并准备虚拟环境:
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 创建)
创建 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 传输
两个细节决定了这段代码能不能在生产里用:
asyncio.to_thread 必不可少。_hash_file 是同步阻塞的磁盘 IO,若直接 await 不存在——你会在循环里阻塞整个事件循环,导致进度通知卡在缓冲区发不出去,Inspector 里看到"进度一次性全涌出来"。丢进线程池后,事件循环才有机会把通知真正推到客户端。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 才从顶层导出。
光有索引不够,再加一个查询工具复用进程内状态:
@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 服务器是被客户端驱动的)。
这是大多数教程跳过、却最该掌握的环节。Inspector 是官方的可视化调试器,能让你在不接 LLM 的情况下,逐条查看客户端与服务器之间的 JSON-RPC 报文,是定位"进度为什么不显示""工具为什么报错"的唯一可靠手段。
方式 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),用浏览器打开即可。
打开 Web UI 后:
Command 与 Args 正确(如 python + server.py)。这里有个宝藏功能——Environment Variables,可以临时注入 DUPSWEEP_DEBUG=1 之类的变量而无需改代码。tools 列表:index_directory、find_duplicates。index_directory,在 Arguments 面板填入 path(例如 /home/you/Documents),点 Run Tool。info: 已处理 25/320)。这就是 Step 1 里那两类通知的可视化。progressToken 的值(如 "t1")和每次 notifications/progress 的 progress/total 都一览无余。这是确认协议实现正确与否的金标准。| 现象 | 原因 | 解决 |
|---|---|---|
| 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。
调试通过后,把服务器注册给 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而静默失败,表现就是工具"死活连不上"。
进入交互会话,让模型直接调用:
> 帮我扫描 ~/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
notifications/message 之外,结合结构化日志(data 字段放对象而非字符串)写入文件,可离线分析慢查询与失败调用。progressToken 往返报文,并在 Claude Code 里看到进度条实时滚动。find_duplicates 与 .mcp.json 团队配置,只保留"写工具→Inspector 调试→命令行注册"主线,压缩到 2 小时。progressToken 的协议级定义,含报文格式与可选性说明ctx.report_progress 与日志 API 的用法与最佳实践[!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 等场景全覆盖。
