首页 › tutorials › 进度通知与日志：长任务可观测

概述

本教程聚焦 MCP（Model Context Protocol）里两条最常被忽视的通知原语——notifications/progress（进度通知）与 notifications/message（日志）。你将学会用 Python 官方 SDK（FastMCP）构建一个能让客户端"看得见进度、读得到日志"的长任务工具，并把它接入 Claude Code 实地观察。适合已经写过"第一个 MCP 工具"、想让工具更可观测、更易调试的中级开发者，预计上手 60–90 分钟。

前置条件

一、为什么你的 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. 请求结束、响应返回后，进度通知必须停止。
// 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": "扫描中..." }
}

规范里有几条硬约束值得记牢：

在 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：

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：

mkdir codebase-stats && cd codebase-stats
python -m venv .venv && source .venv/bin/activate
pip install "mcp[cli]"

写 server.py。这个工具 analyze_codebase 会递归扫描一个目录，统计文件数、总行数、按扩展名的分布，并在扫描过程中同时上报进度和日志：

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 传输

逐行看几个关键设计：

五、用 Inspector 把通知"看"出来

进度和日志是带外通知，跑 print 是看不到它们的——它们走的是 JSON-RPC 通知帧。最稳的观察方式是官方的 MCP Inspector：

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（路径用绝对路径）：

claude mcp add codebase-stats -- python /home/you/codebase-stats/server.py

在会话里触发：

帮我统计一下 /home/you/some-repo 的代码分布

Claude Code 会调用 analyze_codebase。此时两条通道分别这样落地：

claude --debug

--debug 会把 JSON-RPC 往来（含所有 notifications/progress 和 notifications/message）打印到终端，是排查"通知到底发没发"的第一现场。

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)

七、整体工作流

flowchart LR
    A[写 server.py<br/>注入 ctx.report_progress / ctx.info] --> B[Inspector<br/>勾 progressToken 调试]
    B --> C{通知都通?}
    C -- 否 --> A
    C -- 是 --> D[claude mcp add 注册]
    D --> E[claude --debug 观察现场]
    E --> F[限频 + 脱敏打磨]

八、调优与陷阱清单

延伸阅读

训练营适用性评估

参考来源

  1. Progress — Model Context Protocol Specification (2025-03-26) — progressToken 协商与 notifications/progress 单调递增、完成后停止等硬约束的权威出处。
  2. Logging — Model Context Protocol Specification (2025-03-26) — capabilities.logging、RFC 5424 八级严重度、logging/setLevel 与"日志不得含密钥/PII"安全红线。
  3. modelcontextprotocol/python-sdk — README — FastMCP Context 的 report_progress、info/debug/warning/error/log 方法签名与官方 tool_progress.py 代码片段。
  4. Notifications — FastMCP 文档 — 客户端侧如何订阅进度、处理 notifications/message，与本文 Inspector/Claude Code 观察方法互为印证。
  5. Add Progress Bars and Logging Notifications to Your MCP Server — zazencodes — 把进度条与日志落地到一个示例服务器的开发者实战教程。

[!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学通信知识星球海报-扫码加入

📎 本分类最新文章