# 让 MCP 工具实时回报进度与日志：FastMCP Context 实战
## 概述
本教程教你用 Python 的 FastMCP 框架构建一个"会反馈"的 MCP 工具服务器。重点是大多数人踩坑的 `Context` 上下文对象：让长时间运行的工具向 Claude Code **实时回报进度条**、**回流结构化日志**，并**在工具内部读取兄弟资源**。学完你能交付一个带进度反馈的大日志分析服务器，并用 pytest 单测、Inspector 可视化调试，最后接入 Claude Code 观察终端里的实时进度。适合已跑通 Claude Code、想从"调用别人的 MCP"进阶到"造自己的 MCP 工具"的开发者。
## 前置条件
- 已安装并跑通过 Claude Code（能正常对话、执行 `claude mcp list`）
- Python 3.11+ 与 `uv`（推荐）或 `pip`
- 了解函数装饰器、`async/await` 基本语法
- Node.js 18+（仅用于运行 MCP Inspector，非必须）
---
## 一、为什么大多数 MCP 工具是"黑盒"
很多人写完第一个 MCP 工具后的体验是这样的：工具被调用了，然后……Claude Code 转圈圈，几十秒到几分钟没有任何反馈，你不知道它是在正常干活、卡死了、还是在跑一个巨大的循环。
根因在于：MCP 工具默认是"一次性请求—一次性返回"模型。函数签名 `def foo(...) -> str`，调用方只能等它跑完拿到最终结果。这在秒级任务里没问题，但遇到**解析大日志、批量重命名、跑测试套件、爬取分页数据**这类几十秒到分钟级的任务，黑盒体验就直接劝退用户。
MCP 协议其实预留了解法——**服务端通知（notifications）**。FastMCP 把它封装成一个优雅的 `Context` 对象，只要在工具函数里多声明一个参数 `ctx: Context`，你就解锁了三件事：
| Context 能力 | 方法 | 效果 |
|---|---|---|
| 进度上报 | `await ctx.report_progress(current, total)` | 在 Claude Code 终端渲染进度条 |
| 日志回流 | `await ctx.info/debug/warning/error(msg)` | 把服务端日志推给客户端，用户可见 |
| 资源读取 | `await ctx.read_resource(uri)` | 在工具内部读取本服务暴露的资源 |
这三样恰恰是已有 MCP 教程最少覆盖的角落，也是把一个"玩具工具"升级为"生产可用工具"的关键一跃。
```mermaid
flowchart LR
U[用户:分析这批日志] --> CC[Claude Code]
CC -->|tools/call| T[analyze_log 工具]
T -->|ctx.report_progress| CC
T -->|ctx.info 日志| CC
T -->|ctx.read_resource| R[config 资源]
R --> T
T -->|每 1000 行回传进度| CC
T -->|最终结果| CC
CC --> U
style T fill:#fef3c7
style CC fill:#dbeafe
```
---
## 二、环境准备：30 秒起一个 FastMCP 项目
FastMCP 在 2026 年有两个等价入口：Anthropic 官方 `mcp` 包（`from mcp.server.fastmcp import FastMCP`），以及 PrefectHQ 维护、特性最全的独立 `fastmcp` 包。本教程用后者——它的 `Context` 文档与测试工具链最完善，且与官方 SDK API 完全一致，迁移零成本。
用 `uv` 创建一个隔离项目：
```bash
uv init log-analyzer && cd log-analyzer
uv add fastmcp pytest pytest-asyncio
```
如果你坚持用官方包，把下面的 `from fastmcp import ...` 换成 `from mcp.server.fastmcp import FastMCP, Context`，其余代码一字不改。
确认安装成功：
```bash
uv run python -c "import fastmcp; print(fastmcp.__version__)"
```
**预期结果**：打印版本号（如 `3.2.x`），无报错。
**常见问题**：
- `ModuleNotFoundError: fastmcp` —— 你直接用了系统 `python` 而非 `uv run`。所有命令都带 `uv run` 前缀。
- Python < 3.11 报 `TypeError: FastMCP` —— FastMCP 依赖 3.10+ 的类型语法，升级 Python。
---
## 三、最小可用：一个会报进度的工具
先写一个能体现核心价值的 20 行服务器 `server.py`：
```python
from fastmcp import FastMCP, Context
mcp = FastMCP("log-analyzer")
@mcp.tool
async def count_lines(file_path: str, ctx: Context) -> str:
"""逐行统计文件行数，并实时回报进度。"""
with open(file_path, encoding="utf-8") as f:
total = sum(1 for _ in f)
await ctx.info(f"开始统计 {file_path}，共约 {total} 行")
processed = 0
with open(file_path, encoding="utf-8") as f:
for line in f:
processed += 1
if processed % 20 == 0: # 每 20 行回传一次进度
await ctx.report_progress(processed, total)
await ctx.report_progress(total, total)
return f"完成：{file_path} 共 {processed} 行"
if __name__ == "__main__":
mcp.run(transport="stdio")
```
注意三个关键点：
1. `@mcp.tool` 不带括号也能用（FastMCP 2.x+ 装饰器零参数时省略括号）。
2. `ctx: Context` 是**按类型注入**的，调用方（Claude）不需要传它——框架自动填。
3. `report_progress` 和 `info` 都是**协程**，必须 `await`。漏写 `await` 是最高频 bug：代码"看起来跑通了"，但客户端永远收不到进度。
**预期结果**：先用内置 Client 自测，确认能收进度：
```python
# quick_check.py
import asyncio
from fastmcp import Client
from server import mcp
async def main():
async with Client(mcp) as client:
async for msg in client.call_tool("count_lines",
{"file_path": "server.py"},
progress_handler=print):
pass
asyncio.run(main())
```
```bash
uv run python quick_check.py
```
`progress_handler=print` 会把每一条进度通知打印成 `(progress, total, message)` 元组。
**常见问题**：
- 进度回调不触发 —— 检查 `await` 是否漏写，以及 `if processed % 20 == 0` 的步长是否过大（测试文件太小，20 行都没到就结束了）。
- `RuntimeError: no running event loop` —— 顶层没有 `asyncio.run(...)` 包裹。
---
## 四、实战项目：大日志分析服务器
现在构建一个完整、可交付的 MCP 服务器。它做一件真实的事：**分析服务器日志，按级别统计、抓取最近的 ERROR 上下文，并把结果交给 Claude 总结**。整个过程中用户能实时看到"已处理到第几行"。
项目结构：
```
log-analyzer/
├── server.py # MCP 服务器（tool + resource + prompt）
├── sample.log # 测试用日志
├── test_server.py # pytest 单测
├── .mcp.json # Claude Code 接入配置
└── pyproject.toml
```
### 4.1 完整 server.py
```python
from __future__ import annotations
from pathlib import Path
from collections import Counter
from fastmcp import FastMCP, Context
mcp = FastMCP("log-analyzer")
# 配置：通过 resource 暴露，工具内用 ctx.read_resource 读取
SETTINGS = {
"context_lines": "3", # ERROR 上下文窗口
"max_errors": "50", # 最多抓取多少条 ERROR
}
@mcp.tool
async def analyze_log(file_path: str, ctx: Context) -> str:
"""分析日志文件，统计各级别数量并抓取最近的 ERROR 上下文。
长时间运行，会通过进度通知实时回报处理进度。
"""
p = Path(file_path)
if not p.exists():
await ctx.warning(f"文件不存在：{file_path}")
return f"错误：找不到文件 {file_path}"
# —— 演示在工具内读取兄弟资源 ——
cfg = await ctx.read_resource("config://analyzer/context_lines")
context_lines = int(cfg[0].text) if cfg else 3
total = sum(1 for _ in p.open(encoding="utf-8", errors="replace"))
await ctx.info(f"开始分析 {file_path}，共 {total} 行，上下文窗口 {context_lines}")
counter: Counter[str] = Counter()
error_snippets: list[str] = []
window: list[str] = []
with p.open(encoding="utf-8", errors="replace") as f:
for i, line in enumerate(f, 1):
window.append(line.rstrip())
if len(window) > context_lines + 1:
window.pop(0)
for level in ("ERROR", "WARN", "INFO", "DEBUG"):
if level in line:
counter[level] += 1
if "ERROR" in line and len(error_snippets) < 50:
error_snippets.append("\n".join(window))
if i % 2000 == 0: # 每 2000 行回传进度
await ctx.report_progress(i, total,
message=f"已处理 {i}/{total}")
await ctx.report_progress(total, total, message="分析完成")
await ctx.debug(f"抓取到 {len(error_snippets)} 条 ERROR 片段")
summary = ", ".join(f"{k}={v}" for k, v in counter.most_common())
tail = "\n---\n".join(error_snippets[-3:]) or "（无 ERROR）"
return f"## 统计\n{summary}\n\n## 最近 ERROR 上下文\n{tail}"
@mcp.resource("config://analyzer/{key}")
def get_setting(key: str) -> str:
"""读取分析器运行配置（context_lines / max_errors）。"""
return SETTINGS.get(key, "未设置")
@mcp.prompt
def summarize_errors(log: str) -> str:
"""让 Claude 总结日志中的错误模式与可能根因。"""
return (
"你是一名资深 SRE。请分析下面日志的 ERROR 上下文，"
"归纳错误模式、给出 3 条最可能的根因假设，并按优先级排序：\n\n"
f"{log}"
)
if __name__ == "__main__":
mcp.run(transport="stdio")
```
### 4.2 准备测试日志
```bash
uv run python -c "
import random
lv = ['INFO','INFO','INFO','WARN','ERROR','DEBUG']
with open('sample.log','w') as f:
for i in range(12000):
lvl = random.choice(lv)
f.write(f'2026-06-19 10:{i%60:02d}:{i%60:02d} [{lvl}] event #{i} msg payload here\n')
print('sample.log 已生成 12000 行')
"
```
### 4.3 造一条真实 ERROR 便于验证
往 `sample.log` 末尾追加几行带 ERROR 的，确保 `analyze_log` 能抓到上下文：
```bash
printf '2026-06-19 11:00:00 [INFO] connection established\n2026-06-19 11:00:01 [ERROR] NullPointerException at com.app.Handler\n2026-06-19 11:00:02 [ERROR] DatabaseTimeout after 30s\n' >> sample.log
```
### 4.4 命令行自测（不进 Claude Code）
```bash
uv run fastmcp run server.py
```
这会以 stdio 启动并打印注册的 tools/resources/prompts 清单，确认 `analyze_log`、`config://analyzer/{key}`、`summarize_errors` 三个原语都在。
**预期结果**：终端列出 1 个 tool、1 个 resource、1 个 prompt，无异常退出。
**常见问题**：
- `read_resource` 返回空列表 —— 资源 URI 模板必须和 `@mcp.resource` 注册的完全一致（包括 `config://` 前缀和 `{key}` 占位符）。
- 进度在自测时看不到 —— stdio 模式下 stdout 被协议占用，进度走的是通知通道，需要在客户端（Inspector/Claude Code）里才能可视化。
---
## 五、用 FastMCP Client 写单元测试
MCP 服务器最大的测试优势是：**不需要起子进程，Client 和 Server 可以在同一个进程内用内存通道通信**，毫秒级往返、完全确定性。这也是 CI 里跑得最稳的形态。
`test_server.py`：
```python
import pytest
from fastmcp import Client
from server import mcp
@pytest.fixture
def client():
return Client(mcp) # 进程内内存传输
@pytest.mark.asyncio
async def test_lists_three_primitives(client):
async with client as c:
tools = await c.list_tools()
resources = await c.list_resources()
prompts = await c.list_prompts()
assert any(t.name == "analyze_log" for t in tools)
assert any(r.uri.startswith("config://analyzer") for r in resources)
assert any(p.name == "summarize_errors" for p in prompts)
@pytest.mark.asyncio
async def test_analyze_log_counts_levels(client):
async with client as c:
result = await c.call_tool("analyze_log", {"file_path": "sample.log"})
text = result.content[0].text
assert "统计" in text
assert "ERROR=" in text # 确实统计到了 ERROR
@pytest.mark.asyncio
async def test_analyze_log_missing_file(client):
async with client as c:
result = await c.call_tool("analyze_log", {"file_path": "nope.log"})
assert "找不到文件" in result.content[0].text
@pytest.mark.asyncio
async def test_progress_is_reported(client):
progress = []
async with client as c:
await c.call_tool(
"analyze_log",
{"file_path": "sample.log"},
progress_handler=lambda p, t, m=None: progress.append((p, t)),
)
assert progress, "应当至少回传一次进度"
assert progress[-1][0] == progress[-1][1] # 最后一次 = 100%
@pytest.mark.asyncio
async def test_resource_read(client):
async with client as c:
content = await c.read_resource("config://analyzer/context_lines")
assert content[0].text == "3"
```
运行：
```bash
uv run pytest -v
```
**预期结果**：5 个测试全部 `PASSED`，耗时通常 < 1 秒（因为没有进程开销）。
`test_progress_is_reported` 是这一节的核心——它用 `progress_handler` 断言进度通知真的被发出了，这是普通 MCP 测试教程不写的"反向验证"。
**常见问题**：
- `PytestUnraisableExceptionWarning: coroutine never awaited` —— 函数漏了 `async`，或漏了 `@pytest.mark.asyncio`。
- 测试卡住不动 —— 检查 `async with client as c` 是否成对，未关闭的 Client 会阻塞事件循环。
---
## 六、Inspector 可视化调试
单测覆盖了逻辑，但有时你需要"手动戳一下"看真实返回。MCP Inspector 提供浏览器 UI：
```bash
npx @modelcontextprotocol/inspector uv run server.py
```
打开 `http://localhost:6274`，你会看到：
- **Tools** 面板：填入 `file_path=sample.log`，点 *Run*，右侧实时显示 **进度条**（这是验证 `report_progress` 最直观的地方）和返回的 Markdown。
- **Resources** 面板：点 `config://analyzer/context_lines` 查看返回值。
- **Prompts** 面板：渲染 `summarize_errors` 模板。
如果 Inspector 里进度条不动，几乎可以锁定问题在服务端（漏 `await` 或步长太大），而不是 Claude Code 的问题。这是"先在 Inspector 复现，再回 Claude Code"的标准排障路径。
---
## 七、接入 Claude Code 并观察终端进度
### 7.1 注册到 Claude Code
```bash
claude mcp add log-analyzer -- uv run --directory $(pwd) server.py
```
`--directory` 很关键：MCP 服务器是 Claude Code 拉起的**子进程**，它的工作目录不一定是你的项目根目录，显式指定能避免"找不到 server.py / 依赖没装"。
用 `local` 作用域（默认）会把配置写进项目级 `.mcp.json`，可随仓库分享。等价的 `.mcp.json` 手写版：
```json
{
"mcpServers": {
"log-analyzer": {
"command": "uv",
"args": ["run", "--directory", "/绝对路径/log-analyzer", "server.py"],
"env": { "PYTHONUNBUFFERED": "1" }
}
}
}
```
`PYTHONUNBUFFERED=1` 让 stdout 不缓冲——进度通知能更及时地推送，避免"明明跑完了才一次性刷出来"。
### 7.2 验证连接
```bash
claude mcp list
```
**预期结果**：`log-analyzer` 显示为 `✓ connected`，并列出 1 tool / 1 resource / 1 prompt。
如果是 `✗ failed`，单独看详情：
```bash
claude mcp get log-analyzer
```
### 7.3 实际触发并看进度
在 Claude Code 里输入：
> 用 log-analyzer 分析 sample.log，告诉我 ERROR 都出现在哪。
Claude 会调用 `analyze_log`，此时**终端会渲染一条不断刷新的进度条**：`[████████░░░░] 8000/12000 已处理 8000/12000`，并穿插 `ctx.info` 推来的日志行。分析完成后，你可以紧接着让它用 `summarize_errors` prompt 做归因——同一个服务器，工具产出 → prompt 消费，闭环成立。
---
## 八、整体工作流程
```mermaid
flowchart TB
subgraph 开发期
A[写 server.py
tool/resource/prompt] --> B[uv run pytest
进程内单测]
B --> C[npx inspector
手动验证进度]
end
subgraph 接入期
C --> D[claude mcp add
或写 .mcp.json]
D --> E[claude mcp list
确认 connected]
end
subgraph 运行期
E --> F[Claude 调用 analyze_log]
F --> G[ctx.report_progress
终端进度条]
F --> H[ctx.info/debug
结构化日志回流]
F --> I[ctx.read_resource
读 config]
G --> J[用户实时可见]
H --> J
I --> K[最终 Markdown 结果]
K --> L[summarize_errors prompt 归因]
end
style F fill:#fef3c7
style G fill:#dcfce7
style H fill:#dcfce7
```
---
## 九、常见问题排错清单
| 现象 | 根因 | 解法 |
|---|---|---|
| Claude Code 转圈无反馈 | `report_progress` 漏 `await` | 检查所有 ctx 方法前都有 `await` |
| Inspector 进度条不动 | 文件太小、步长太大没触发 | 改 `if i % 50 == 0` 或加大测试文件 |
| `mcp list` 显示 failed | 子进程工作目录不对/依赖未装 | `.mcp.json` 加 `--directory`，确认 `uv add` 在同项目 |
| stdout 出现乱码 JSON | 在工具里 `print()` 污染了 stdio | 禁用 `print`，日志一律走 `ctx.info/debug` |
| 资源读取返回空 | URI 模板占位符不匹配 | `config://analyzer/{key}` 必须字字对应 |
| 测试随机失败 | 事件循环未正确关闭 | 确认 `async with client` 配对，用 `pytest-asyncio` 的 `auto` 模式 |
核心心法一句话：**stdio 模式下 stdout 是协议通道，任何用户可见的反馈都必须走 `ctx`，而不是 `print`。**
---
## 延伸阅读
- **FastMCP 官方文档 — Context 章节**：进度、日志、资源读取的完整 API，含 `ctx.session` 低级访问
- **MCP 协议规范 — Notifications**：理解 `notifications/progress` 在协议层如何编码，助于跨语言实现
- **Anthropic 官方 `mcp` Python SDK**：`from mcp.server.fastmcp import FastMCP` 的官方等价实现
- **FastMCP 测试模式**：进程内 Client/Server 测试、`progress_handler`、多传输层 fixture 写法
- **Claude Code MCP 管理 CLI**：`claude mcp add/get/list/remove` 与 `.mcp.json` 三种作用域的取舍
---
## 训练营适用性评估
- **适合类型**：MCP 工具开发训练营（进阶）、AI Agent 工具链训练营、Python 后端实战训练营
- **教学时长**：约 2.5–3 小时（讲解 45 分钟 + 跟做 90 分钟 + 排障与答疑 30 分钟）
- **难度定位**：⭐⭐⭐（中阶）。要求先有 MCP 基础认知和 Python 异步经验
- **可延伸作业**：(1) 把 `analyze_log` 改成支持正则过滤级别；(2) 加一个"分页读取资源"演示 `ctx.read_resource` 的批量用法；(3) 把传输从 stdio 切到 streamable-http，观察远程部署下进度是否仍能回流
- **不建议用于**：零基础训练营（应先跑通"调用现成 MCP 工具"再谈自建）、纯前端/无 Python 环境的场次
---
## 参考来源
1. [FastMCP Context 官方文档](https://fastmcp.wiki/en/servers/context) — Context 对象的 logging/progress/resource 完整 API 参考
2. [FastMCP Progress Reporting](https://gofastmcp.com/servers/progress) — `ctx.report_progress` 的进度通知机制与示例
3. [FastMCP Testing Patterns](https://fastmcp.wiki/en/patterns/testing) — 进程内 Client/Server 测试、pytest-asyncio 配置与 progress_handler 写法
4. [MCP Python SDK (modelcontextprotocol/python-sdk)](https://github.com/modelcontextprotocol/python-sdk) — 官方 `mcp` 包，`from mcp.server.fastmcp import FastMCP` 的源头实现
5. [FastMCP in Python: Build a Real MCP Server (2026 Guide)](https://www.danilchenko.dev/posts/fastmcp-mcp-server/) — FastMCP 3.2 生产级服务器实战（tools/resources/prompts 全流程）
> [!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 等场景全覆盖。
