## 概述 本教程聚焦 MCP（Model Context Protocol）里两条最常被忽视的**通知原语**——`notifications/progress`（进度通知）与 `notifications/message`（日志）。你将学会用 Python 官方 SDK（FastMCP）构建一个能让客户端"看得见进度、读得到日志"的长任务工具，并把它接入 Claude Code 实地观察。适合已经写过"第一个 MCP 工具"、想让工具更可观测、更易调试的中级开发者，预计上手 60–90 分钟。 ## 前置条件 - Python 3.10+，会装包（`uv` 或 `pip` 均可） - 装好 Claude Code CLI（`claude --version` 可用） - 读过任意一篇 FastMCP"造首个工具"教程，知道 `@mcp.tool()` 与 `mcp.run()` 怎么跑 - 一台能联网的机器（用于跑 `@modelcontextprotocol/inspector`） --- ## 一、为什么你的 MCP 工具需要"通知" 大多数入门教程只演示**请求-响应**：客户端调用一个工具，等待，拿到结果。这种模型在工具秒级返回时没问题，但一旦工具要做的事变长——递归扫一个万行代码库、拉一个大表、跑一段构建——客户端就只能干等。用户盯着一个没有反馈的转圈，根本不知道是卡死了还是在干活。 MCP 在 JSON-RPC 之上额外定义了两种**带外通知（out-of-band notification）**，专门解决"长任务的黑盒问题"： | 原语 | 通知方法 | 方向 | 解决的问题 | 客户端可控项 | |---|---|---|---|---| | 进度 Progress | `notifications/progress` | 双向 | "任务跑到哪了" | 在请求里塞 `progressToken` 决定要不要收 | | 日志 Logging | `notifications/message` | 服务器→客户端 | "服务器内部发生了什么" | 用 `logging/setLevel` 调最低级别 | 注意一个关键事实：**这两条通道都是"通知"，不是"请求"**——通知没有 `id`，不需要对方回 `result`，因此服务器可以一边干活一边往外推消息，不阻塞主请求的执行。理解这一点，是后面所有代码的地基。 --- ## 二、进度原语：progressToken 是怎么接上的 进度的机制比看上去精巧。它**不是**服务器单方面狂推，而是一个"约定"流程： 1. 客户端在发起某个请求时，在 `params._meta.progressToken` 里塞一个自己挑的 token（字符串或整数，必须全局唯一）。 2. 服务器在处理这个请求的过程中，针对**同一个 token** 反复发 `notifications/progress`，每次带 `progress`（当前值）、可选 `total`（总量）、可选 `message`（人类可读说明）。 3. 请求结束、响应返回后，进度通知**必须停止**。 ```json // 1. 客户端：请求里带 token { "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "_meta": { "progressToken": "abc123" } } } // 2. 服务器：反复推进度 { "jsonrpc": "2.0", "method": "notifications/progress", "params": { "progressToken": "abc123", "progress": 50, "total": 100, "message": "扫描中..." } } ``` 规范里有几条硬约束值得记牢： - `progress` 值**必须单调递增**（哪怕总量未知）——这是客户端判断"有没有卡住"的依据； - `progress` 和 `total` 可以是浮点，所以你完全可以用 `0.55` 表示 55%； - 服务器**可以**选择完全不发进度（opt-in），也可以用任意自认合理的频率； - 收发双方**都应**做限频，避免进度通知把通道刷爆。 在 FastMCP 里你**不用手写** token 协商——框架在 `tools/call` 时自动从请求里取出 token 并在 `Context` 里准备好。你只要在工具里 `await ctx.report_progress(progress=, total=, message=)` 即可。 --- ## 三、日志原语：带级别的能力声明 日志是另一条更"轻量"的通道，用来吐服务器内部的非关键信息。规范遵循 RFC 5424 的 syslog 严重级别，从轻到重一共 8 档： | 级别 | 用途 | 在 MCP 工具里的典型场景 | |---|---|---| | `debug` | 详细调试 | 进入/退出某个函数、单步细节 | | `info` | 常规信息 | 阶段完成、统计概览 | | `notice` | 显著的正常事件 | 配置变更、缓存命中 | | `warning` | 警告 | 跳过坏文件、用了已弃用参数 | | `error` | 错误 | 单次操作失败（但工具整体未崩） | | `critical` / `alert` / `emergency` | 严重/紧急/系统不可用 | 组件级故障、数据损坏 | 三个要点决定了日志"能不能被看见"： 1. **能力声明**：服务器若要发日志，必须在初始化握手时声明 `capabilities.logging = {}`（FastMCP 默认已声明）。 2. **客户端控级别**：客户端用 `logging/setLevel` 请求告诉服务器"我只关心 `info` 及以上"，服务器据此过滤。默认情况下很多客户端把级别卡得较高，`debug` 往往收不到。 3. **安全红线**：日志**绝对不能**包含密钥、凭据、个人隐私信息——它是会推给客户端、甚至可能落盘的内容。 FastMCP 的 `Context` 提供了一组语义化方法，背后都是发 `notifications/message`： ```python await ctx.debug("进入扫描循环") await ctx.info(f"发现 {n} 个文件") await ctx.warning(f"跳过 {name}: 权限不足") await ctx.error(f"读取失败: {e}") await ctx.log(level="notice", data={"hit": True, "key": "x"}) # 带结构化 data ``` --- ## 四、动手：搭一个会"汇报"的 FastMCP 服务器 新建目录 `codebase-stats`，装好官方 SDK： ```bash mkdir codebase-stats && cd codebase-stats python -m venv .venv && source .venv/bin/activate pip install "mcp[cli]" ``` 写 `server.py`。这个工具 `analyze_codebase` 会递归扫描一个目录，统计文件数、总行数、按扩展名的分布，并在扫描过程中**同时**上报进度和日志： ```python from __future__ import annotations from pathlib import Path from collections import Counter from mcp.server.fastmcp import Context, FastMCP from mcp.server.session import ServerSession mcp = FastMCP("codebase-stats") EXCLUDED = {".git", "node_modules", "__pycache__", ".venv", "dist", "build"} CODE_EXT = {".py", ".js", ".ts", ".tsx", ".go", ".rs", ".java", ".md", ".yaml", ".yml"} @mcp.tool() async def analyze_codebase( path: str, ctx: Context[ServerSession, None] ) -> dict: """递归统计代码库：文件数、总行数、按扩展名分布；扫描中持续上报进度与日志。""" root = Path(path).expanduser().resolve() if not root.exists(): await ctx.error(f"路径不存在: {root}") return {"error": f"path not found: {root}"} # 先一次性收集待扫描清单，便于给出准确的 total files = [ p for p in root.rglob("*") if p.is_file() and not any(part in EXCLUDED for part in p.parts) and p.suffix.lower() in CODE_EXT ] total = len(files) await ctx.info(f"发现 {total} 个待扫描文件，开始统计") by_ext: Counter[str] = Counter() total_lines = 0 skipped = 0 # 每 5% 上报一次，避免刷屏 step = max(1, total // 20) for i, p in enumerate(files, start=1): try: with p.open("r", encoding="utf-8", errors="replace") as f: lines = sum(1 for _ in f) except OSError as e: skipped += 1 await ctx.warning(f"跳过 {p.name}: {e}") continue by_ext[p.suffix.lower() or "(none)"] += 1 total_lines += lines if i == total or i % step == 0: await ctx.report_progress( progress=i, total=total, message=f"已扫描 {i}/{total} 个文件", ) await ctx.debug(f"{p.name}: +{lines} 行") await ctx.info(f"完成：{total} 文件，{total_lines} 行，{skipped} 个跳过") return { "root": str(root), "files_scanned": total - skipped, "skipped": skipped, "total_lines": total_lines, "by_extension": dict(by_ext.most_common(10)), } if __name__ == "__main__": mcp.run() # 默认 stdio 传输 ``` 逐行看几个关键设计： - **先建清单再扫**：`rglob` 一遍拿到 `total`，进度才有可信的分母。代价是多一次遍历，但对本地磁盘通常可接受。 - **限频上报**：`i % step == 0` 让通知频率稳定在约 20 次，既给用户连续反馈，又不违反"应做限频"的规范。 - **错误降级为日志**：单个文件读失败不抛异常中断整次扫描，而是 `ctx.warning` 后跳过——错误信息走日志通道，结果里只汇总 `skipped` 计数。 - **结构化返回**：返回 `dict`，FastMCP 会自动按结构化输出（structured output）生成 schema，客户端拿到的就是规整的 JSON。 --- ## 五、用 Inspector 把通知"看"出来 进度和日志是带外通知，跑 `print` 是看不到它们的——它们走的是 JSON-RPC 通知帧。最稳的观察方式是官方的 MCP Inspector： ```bash npx -y @modelcontextprotocol/inspector python server.py ``` 浏览器会打开一个图形界面。流程是： 1. 在 Inspector 里点 **Connect**，确认握手成功； 2. 找到工具 `analyze_codebase`，点 **List Tools** 后填入一个真实目录路径； 3. 在调用前的选项里**勾上 "Progress token"**（Inspector 会自动生成 token 并塞进 `_meta`）； 4. 点 **Run**，右侧 **Notifications** 面板会**实时滚动**出 `notifications/progress` 与 `notifications/message`。 如果你看到 `progress` 从 5%、10% 一路涨到 100%，并且每档日志按级别出现，就说明两条通道都通了。 > **常见问题：Inspector 里看不到日志？** > 多半是级别被客户端（Inspector）卡在 `info` 以上，而你发的多是 `debug`。在 Inspector 面板里把日志级别手动设到 `debug`，或把代码里的 `ctx.debug` 临时换成 `ctx.info` 验证。 --- ## 六、接入 Claude Code 并观察 把服务器注册给 Claude Code（路径用绝对路径）： ```bash claude mcp add codebase-stats -- python /home/you/codebase-stats/server.py ``` 在会话里触发： ``` 帮我统计一下 /home/you/some-repo 的代码分布 ``` Claude Code 会调用 `analyze_codebase`。此时两条通道分别这样落地： - **进度**：Claude Code 在工具调用进行中用它驱动过程指示，`message` 文本会被用来更新"正在做什么"的提示。 - **日志**：Claude Code 消费 `notifications/message` 并在会话上下文里呈现。想看原始通知帧，用调试模式启动： ```bash claude --debug ``` `--debug` 会把 JSON-RPC 往来（含所有 `notifications/progress` 和 `notifications/message`）打印到终端，是排查"通知到底发没发"的第一现场。 ```mermaid sequenceDiagram participant CC as Claude Code (客户端) participant S as codebase-stats (服务器) Note over CC,S: tools/call 带 progressToken CC->>S: analyze_codebase(path) + _meta.progressToken S-->>CC: notifications/message (info: 发现 N 文件) loop 扫描中 S-->>CC: notifications/progress (i/total) S-->>CC: notifications/message (debug: +x 行) end S-->>CC: notifications/message (info: 完成) S->>CC: tools/call 结果 (结构化 JSON) ``` --- ## 七、整体工作流 ```mermaid flowchart LR A[写 server.py
注入 ctx.report_progress / ctx.info] --> B[Inspector
勾 progressToken 调试] B --> C{通知都通?} C -- 否 --> A C -- 是 --> D[claude mcp add 注册] D --> E[claude --debug 观察现场] E --> F[限频 + 脱敏打磨] ``` --- ## 八、调优与陷阱清单 - **进度数字必须递增**。如果你按"剩余"来算导致后一条比前一条小，客户端会以为出错了。永远用"已完成量"当 `progress`。 - **频率是义务**。规范要求双方限频。若你每个文件都 `report_progress`，扫描 10 万个文件就是 10 万条通知，会拖垮通道。用固定步长或时间窗口（比如每 200ms 至多一条）。 - **完成即停**。响应返回后绝不再发该 token 的进度。如果你在后台线程里还在推，要么换新 token，要么显式结束。 - **同步函数拿不到 ctx**。`report_progress` 是 `await`，工具必须是 `async def` 且参数里带 `ctx: Context[...]`；写成普通 `def` 不会注入上下文。 - **默认收不到 debug**。客户端默认级别偏高，正式诊断时要么手动 `setLevel` 到 `debug`，要么用 Inspector 强制低级别。 - **日志是出口，不是出口的垃圾桶**。别把异常堆栈、token、API key、用户邮箱写进 `data`。规范明确禁止凭据/PII 出现在日志里——它会被推送并可能持久化。 - **结构化 data 优于纯字符串**。`ctx.log(level="warning", data={"file": x, "reason": y})` 比拼字符串更便于客户端过滤和展示。 - **进度 vs 日志别混用**。进度回答"到第几步了"，日志回答"这一步里发生了什么"。把进度也塞进日志会让两条通道语义重叠、更难消费。 --- ## 延伸阅读 - MCP 官方规范 · Progress： - MCP 官方规范 · Logging： - Python SDK README（Context、report_progress、logging 方法示例）： - FastMCP 通知与客户端处理文档： - "Add Progress Bars and Logging Notifications to Your MCP Server"： ## 训练营适用性评估 - **适合人群**：已经完成"首个 MCP 工具"、想进阶到"可观测、可调试工具"的中级后端/平台开发者。 - **不适合**：纯零基础（建议先做 FastMCP 首工具）；只做秒级返回工具、无需过程反馈的场景。 - **建议时长**：讲授 + 跟做约 75 分钟，其中 Inspector 观察 20 分钟、Claude Code 接入与 `--debug` 排查 20 分钟最值钱。 - **可扩展作业**：把进度通知用于"安装依赖""跑测试套件""批量迁移文件"三个真实长任务，要求学员各自实现限频与脱敏。 ## 参考来源 1. [Progress — Model Context Protocol Specification (2025-03-26)](https://modelcontextprotocol.io/specification/2025-03-26/basic/utilities/progress) — progressToken 协商与 `notifications/progress` 单调递增、完成后停止等硬约束的权威出处。 2. [Logging — Model Context Protocol Specification (2025-03-26)](https://modelcontextprotocol.io/specification/2025-03-26/server/utilities/logging) — `capabilities.logging`、RFC 5424 八级严重度、`logging/setLevel` 与"日志不得含密钥/PII"安全红线。 3. [modelcontextprotocol/python-sdk — README](https://github.com/modelcontextprotocol/python-sdk) — FastMCP `Context` 的 `report_progress`、`info/debug/warning/error/log` 方法签名与官方 `tool_progress.py` 代码片段。 4. [Notifications — FastMCP 文档](https://gofastmcp.com/clients/notifications) — 客户端侧如何订阅进度、处理 `notifications/message`，与本文 Inspector/Claude Code 观察方法互为印证。 5. [Add Progress Bars and Logging Notifications to Your MCP Server — zazencodes](https://zazencodes.substack.com/p/add-progress-bars-and-logging-notifications) — 把进度条与日志落地到一个示例服务器的开发者实战教程。 > [!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 等场景全覆盖。