MCP长任务三原语：进度日志与取消

首页 › tutorials › MCP长任务三原语：进度日志与取消

概述

本教程深入讲解 MCP（Model Context Protocol）中专门服务于长任务的三个通知原语：Logging（日志）、Progress（进度上报） 和 Cancellation（取消）。普通工具调用是"一问一答"的同步往返，但当你构建数据库迁移、批量文件分析、大规模爬取这类耗时数分钟甚至更久的工具时，模型和用户只能对着黑盒干等。这三个原语就是 MCP 用来打破黑盒、让服务器"会汇报、可中止"的标准化机制。

适用人群：已能用 Python/TypeScript SDK 跑通一个基础 MCP 工具（Tools 原语），现在想让工具支持长任务反馈与中止的开发者。

前置条件

已安装 Node.js ≥ 18（用于 MCP Inspector 调试）
已安装 Python ≥ 3.10 和 uv（pip install uv 或 pipx install uv）
能跑通一个最简 FastMCP 工具，了解 @mcp.tool() 装饰器与 stdio 传输
建议先装好 MCP Inspector：npx @modelcontextprotocol/inspector

一、为什么长任务需要"三个通知原语"

回顾 MCP 的通信模型：它基于 JSON-RPC 2.0，所有交互要么是 Request（带 id、期望响应），要么是 Notification（无 id、单向"发后即忘"）。

工具调用（tools/call）本质上是一个 Request：客户端发请求 → 服务端执行 → 服务端返回唯一的结果。问题在于，在请求和响应之间的这段"执行期"里，通道是沉默的。如果一个工具要跑 3 分钟，这 3 分钟内服务端无法对客户端说一句话，客户端也无法喊停。

MCP 用三个 Notification 来填补这段沉默：

原语	方向	方法名	解决什么
Logging	服务端 → 客户端	`notifications/message`	把过程信息（步骤、警告、错误）实时回传
Progress	双向（服务端 → 客户端为主）	`notifications/progress`	上报"已完成 X / 共 Y"的可量化进度
Cancellation	双向	`notifications/cancelled`	中止一个正在执行的请求

下面三步分别攻克它们，最后用一个完整的"批量日志分析服务器"把它们串起来。

二、步骤一：Logging 原语——让服务器开口说话

协议层要点

Logging 原语允许服务端向客户端发送结构化日志。关键规范（取自 MCP 2025-06-18 规范）：

能力声明：服务端要在 initialize 握手时声明 logging capability。
八个日志级别：完全遵循 RFC 5424 syslog 级别——debug、info、notice、warning、error、critical、alert、emergency。
级别控制：客户端通过 logging/setLevel 请求设置最低级别，服务端只回传不低于该级别的日志。
消息格式：用 notifications/message 发送，参数包含 level、可选的 logger（日志源名称）、以及任意可序列化的 data。

规范原文给出的日志消息结构：

{
  "jsonrpc": "2.0",
  "method": "notifications/message",
  "params": {
    "level": "error",
    "logger": "database",
    "data": {
      "error": "Connection failed",
      "details": { "host": "localhost", "port": 5432 }
    }
  }
}

代码实现（FastMCP）

FastMCP 的 Context 对象把八个级别各封装成一个异步方法，无需手写 JSON-RPC：

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("log-demo")

@mcp.tool()
async def fetch_weather(city: str, ctx: Context) -> str:
    """模拟一个带多级日志的天气查询工具"""
    ctx.info(f"开始查询 {city} 的天气")          # info 级别：常规信息
    ctx.debug("构造 API 请求参数")               # debug 级别：调试细节

    try:
        # ... 这里放真实的网络请求 ...
        ctx.notice("数据源返回成功，准备解析")     # notice 级别：显著但正常的事件
        return f"{city}: 晴，25°C"
    except Exception as e:
        ctx.error(f"查询失败: {e}")              # error 级别：操作失败
        ctx.warning("已回退到缓存数据")          # warning 级别：降级提示
        return f"{city}（缓存）: 晴，24°C"

预期结果

在 MCP Inspector 里调用 fetch_weather 时，Notifications 面板会按时间顺序出现多条 notifications/message，每条带 level 和 data 字段。在 Claude Code 里，info 以上级别的日志会以浅色提示行的形式穿插在工具执行过程中，让用户感知到"它在干活"。

常见问题与解决

日志完全不显示？ 检查客户端是否通过 logging/setLevel 把级别设成了 warning 或更高。Inspector 默认显示所有级别；某些客户端默认只显示 info 及以上。debug 级别最容易被"静音"。
日志刷屏？ 规范明确要求服务端 SHOULD 实现速率限制（rate limiting）。在循环里发日志时，改为"每 N 次迭代发一条"或"每秒最多一条"，避免把客户端淹没。
日志泄露了密钥？ 规范的 Security 小节强制要求：日志 MUST NOT 包含凭证、密钥、个人隐私信息。发日志前务必脱敏。

三、步骤二：Progress 原语——上报"完成度"

协议层要点

Progress 用于流式上报长操作的进度。它比 Logging 更"结构化"——Logging 是文字过程，Progress 是可量化的百分比/计数器。核心机制是 progressToken（进度令牌）：

客户端发起追踪：在请求的 _meta 里塞一个 progressToken（字符串或整数），表示"我想接收这个请求的进度"：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "some_method",
  "params": {
    "_meta": { "progressToken": "abc123" }
  }
}

服务端上报：用同一个 token 发 notifications/progress，携带 progress（当前值）、可选 total（目标值）、可选 message（人类可读说明）：

{
  "jsonrpc": "2.0",
  "method": "notifications/progress",
  "params": {
    "progressToken": "abc123",
    "progress": 50,
    "total": 100,
    "message": "Reticulating splines..."
  }
}

三条硬性规则（摘自规范）： - progress 值必须单调递增，即使 total 未知。 - progress 和 total 可以是浮点数。 - 完成后必须停止发送进度通知。

代码实现（FastMCP）

progressToken 的细节被 SDK 完全屏蔽——你只需调用 ctx.report_progress()，SDK 会自动取出当前请求关联的 token 并封装好通知：

import asyncio
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("progress-demo")

@mcp.tool()
async def batch_translate(items: list[str], ctx: Context) -> str:
    """批量翻译，并逐项上报进度"""
    total = len(items)
    results = []
    for i, item in enumerate(items):
        # ... 实际翻译逻辑 ...
        await asyncio.sleep(0.3)  # 模拟耗时
        results.append(f"[已翻译] {item}")
        # SDK 自动绑定当前请求的 progressToken
        await ctx.report_progress(
            progress=i + 1,
            total=total,
        )
    return f"完成 {total} 项:\n" + "\n".join(results)

预期结果

在 Cursor 等支持进度条的客户端里，工具执行期间会出现一条动态推进的进度条（如 3/10）。在 Inspector 的 Notifications 面板会看到多条 notifications/progress，progress 字段从 1 递增到 total。Claude Code 会把进度以"完成 30%"之类的简要行回显。

常见问题与解决

进度条没出现？ 进度通知是可选的——规范说接收方"MAY 选择完全不发进度通知"。另外，并非所有客户端都把 progress 渲染成进度条（Claude Code 目前以文字行呈现，Cursor 渲染成进度条）。token 一定是客户端给的：服务端不能自造 token，没有 token 时 report_progress 会被安全忽略。
不知道总数怎么办？ 直接省略 total，只传 progress 递增值。客户端会显示成"已处理 N 项"而非百分比。
要不要每条都发？ 不必。规范说频率由发送方自行决定。对 10000 项的任务，每 100 项发一次即可，避免通知洪水。

四、步骤三：Cancellation 原语——优雅中止

协议层要点

任何一方都可以取消一个进行中的请求，方法是发一条 notifications/cancelled，带上 requestId 和可选 reason：

{
  "jsonrpc": "2.0",
  "method": "notifications/cancelled",
  "params": {
    "requestId": "123",
    "reason": "User requested cancellation"
  }
}

几个容易踩坑的规则： - 只能取消本方向发出的请求，且要求"认为仍在进行中"。 - initialize 请求禁止被取消。 - 接收方应该停止处理、释放资源、并且不再返回响应。 - 取消是 best-effort（尽力而为）：处理可能已经完成，甚至响应可能已经发出。 - 必须处理竞态：由于网络延迟，取消通知可能在处理完成后才到达，双方都要优雅应对（"发后即忘"，忽略对已完成/未知请求的取消）。

代码实现（FastMCP）

SDK 把协议层的 cancelled 通知映射到底层异步取消语义：收到取消时，正在 await 的工具协程会收到 asyncio.CancelledError。最佳实践是用 try/finally 保证资源释放，用 try/except 捕获取消：

import asyncio
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMcp("cancel-demo")

@mcp.tool()
async def long_running_job(minutes: int, ctx: Context) -> str:
    """模拟一个长任务：每秒处理一帧，可被中途取消"""
    ctx.info(f"启动任务，预计 {minutes} 分钟")
    processed = 0
    try:
        total_frames = minutes * 60
        for f in range(total_frames):
            # 真实场景：逐帧渲染 / 逐行处理
            await asyncio.sleep(1)        # <- 取消信号会在这里抛出 CancelledError
            processed += 1
            await ctx.report_progress(progress=processed, total=total_frames)
        return f"全部完成，处理 {processed} 帧"
    except asyncio.CancelledError:
        # 用户/客户端取消了任务
        ctx.warning(f"任务被取消，已完成 {processed} 帧，正在保存断点")
        # ... 这里写断点保存逻辑 ...
        raise   # 重要：重新抛出，让 SDK 完成取消协议交互，不返回正常响应
    finally:
        # 无论正常完成还是取消，都释放资源
        ctx.info("资源已释放")

预期结果

在 Inspector 里发起 long_running_job 后，从客户端侧再发一条 notifications/cancelled（Inspector 提供"Cancel Request"按钮），服务器日志会打印"任务被取消…资源已释放"，且不会再返回正常响应。在支持取消的客户端（如某些 IDE）中，用户点击停止按钮即可触发同样的流程。

常见问题与解决

取消后还收到了工具结果？ 这是竞态的正常表现——处理可能在取消通知到达前已完成。规范明确要求发送方应该忽略取消之后收到的响应。
服务器根本不理会取消？ 取消是 best-effort，规范允许接收方在"请求无法取消"时忽略。如果你的长任务卡在不可中断的外部调用（如同步阻塞 IO）上，取消就难以生效——尽量用 asyncio.sleep / asyncio.wait_for 这类可中断的 await 点。
忘记 raise CancelledError？ 如果只 except 不 raise，函数会返回一个正常结果，破坏取消语义。除非你有意要"软停止"（返回部分结果），否则应在保存断点后重新抛出。

五、实战案例：批量日志分析服务器

现在把三原语融合成一个完整、可运行的服务器。场景：分析一个目录下的所有日志文件，统计 ERROR 级别条目。这个任务可能涉及成百上千个文件，天然需要进度反馈、过程日志、以及"中途不想等了"的取消能力。

完整代码（`log_analyzer.py`）

"""
批量日志分析 MCP 服务器
- Logging：记录分析阶段、警告、错误
- Progress：上报文件级进度
- Cancellation：支持中途取消并保存断点
"""
import asyncio
import json
from pathlib import Path

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("log-analyzer")


async def _scan_file(path: Path) -> int:
    """统计单个文件中的 ERROR 行数，带可中断点"""
    count = 0
    with path.open("r", encoding="utf-8", errors="ignore") as fh:
        for line in fh:
            if "ERROR" in line:
                count += 1
            # 每 2000 行让出一次控制权，给取消信号一个触发点
            if count and count % 2000 == 0:
                await asyncio.sleep(0)
    return count


@mcp.tool()
async def analyze_logs(directory: str, ctx: Context) -> str:
    """
    扫描目录下所有 .log 文件，统计 ERROR 条目数。
    支持进度上报与中途取消。
    """
    root = Path(directory)
    if not root.is_dir():
        ctx.error(f"路径不是目录: {directory}")
        return json.dumps({"error": "invalid directory"})

    files = sorted(root.rglob("*.log"))
    total = len(files)
    if total == 0:
        ctx.warning("目录下没有 .log 文件")
        return json.dumps({"error_count": 0, "scanned": 0})

    ctx.info(f"发现 {total} 个日志文件，开始分析")
    results: dict[str, int] = {}
    processed = 0

    try:
        for fp in files:
            # —— Progress：按文件粒度上报 ——
            await ctx.report_progress(progress=processed + 1, total=total)
            # —— Logging：每个大文件单独记一条 notice ——
            if fp.stat().st_size > 1_000_000:
                ctx.notice(f"处理大文件: {fp.name}")
            try:
                n = await _scan_file(fp)   # 可中断的 await 点
                if n:
                    results[str(fp)] = n
            except Exception as e:
                ctx.error(f"读取 {fp.name} 失败: {e}")
            processed += 1

        total_errors = sum(results.values())
        ctx.info(f"分析完成：共 {total_errors} 条 ERROR")
        return json.dumps(
            {"error_count": total_errors, "scanned": processed, "top": results},
            ensure_ascii=False, indent=2,
        )

    except asyncio.CancelledError:
        # —— Cancellation：保存断点后让 SDK 完成取消交互 ——
        ctx.warning(f"已取消，已扫描 {processed}/{total}，保存断点到 .analyzer-state")
        ctx.session  # 占位，表示可在此访问会话
        (root / ".analyzer-state").write_text(
            json.dumps({"scanned": processed, "total": total})
        )
        raise


if __name__ == "__main__":
    mcp.run(transport="stdio")

接入 Claude Code 测试

在 ~/.claude.json（或项目的 .mcp.json）里注册：

{
  "mcpServers": {
    "log-analyzer": {
      "command": "uv",
      "args": ["run", "--with", "mcp", "python", "log_analyzer.py"]
    }
  }
}

启动 Claude Code，输入提示让它调用工具：

请用 log-analyzer 分析 /var/log/myapp 目录下的 ERROR，并告诉我最多的文件。

你会观察到：执行过程中穿插出现 info（发现 N 个文件）、notice（处理大文件）等日志行，以及"扫描进度 X/Y"的进度反馈。若中途按下 Esc 取消，服务器会写入 .analyzer-state 断点。

用 Inspector 实测三原语

更精准的观察方式是用 MCP Inspector：

npx @modelcontextprotocol/inspector uv run --with mcp python log_analyzer.py

打开浏览器控制台后： 1. 在 Tools 面板选中 analyze_logs，填入 directory，点 Run。 2. 右侧 Notifications 流里会实时滚动 notifications/message（日志）和 notifications/progress（进度）。 3. 在工具还在跑时，点 Cancel Request，观察服务端是否打印"已取消"且不再返回结果——这就是 notifications/cancelled 的完整往返。

六、工作流程：一次长任务调用的完整交互

下图把三原语串成一次完整的长任务交互时序：

sequenceDiagram
    participant C as Client (Claude Code / Inspector)
    participant S as MCP Server (log-analyzer)

    C->>S: tools/call (analyze_logs) + _meta.progressToken
    Note over S: 开始执行长任务

    loop 每个文件
        S-->>C: notifications/message (level=notice, 大文件提示)
        S-->>C: notifications/progress (progress=k, total=N)
    end

    alt 用户中途取消
        C-->>S: notifications/cancelled (requestId, reason)
        Note over S: 触发 CancelledError
        S-->>C: notifications/message (level=warning, 保存断点)
        Note over S: 释放资源，不返回响应
    else 正常完成
        S->>C: tools/call 结果 (JSON 结果)
    end

关键点：所有 -->> 的虚线箭头都是 Notification（无 id、无响应），它们穿插在唯一的 tools/call 请求-响应对之间，正是它们把"沉默的执行期"变成了可观测、可控的过程。

七、三原语选型决策

在实际开发中，三者并非每次都要全用，按下表按需取用：

场景特征	选 Logging	选 Progress	选 Cancellation
想记录"正在做什么"的文字过程	✅
任务有明确的可量化进度		✅
任务可能跑很久、用户可能反悔			✅
任务秒级完成	可选	不必	不必
调试期排查问题	✅（debug 级）

一个经验法则：凡是超过 5 秒的任务，至少给 Progress 或 Logging 之一；凡是超过 1 分钟的任务，三者都建议上。

八、延伸阅读

MCP 规范三剑客：Logging、Progress、Cancellation 三页是本文的权威依据，规范里的所有 MUST/SHOULD 都来自这里。
Python SDK 文档：Context 对象的 info/debug/report_progress 等方法签名与示例，是代码实现的直接来源。
MCP Inspector：观察 notification 流的最佳工具，建议每次开发带通知的服务器都用它实测一遍。
RFC 5424：Logging 的八个级别定义出自这份 syslog 标准，理解级别语义有助于选对级别。

训练营适用性评估

适合的训练营类型：MCP 服务器开发进阶营、AI Agent 工具构建实战营、企业级 LLM 工具链工程营。要求学员已具备基础 MCP/Tools 经验，属于"中级→高级"的内容。
不适合：零基础入门营（需先掌握 Tools 原语和 SDK 基本用法）。
预计教学时长：约 90–120 分钟。其中协议讲解 20 分钟、三个分步练习各 20 分钟、综合实战案例 30 分钟、Inspector 实测与答疑 20 分钟。
配套练习建议：让学员改造自己已有的某个长耗时工具，分别加上 Logging、Progress，最后支持 Cancellation，形成可演示的作业。

参考来源

Progress — Model Context Protocol — 官方进度上报规范，progressToken、notifications/progress 的权威定义（2025-06-18 版本）
Logging — Model Context Protocol — 官方日志规范，RFC 5424 八级、logging/setLevel、notifications/message、安全要求
Cancellation — Model Context Protocol — 官方取消规范，notifications/cancelled、竞态处理、best-effort 语义
Progress & Notifications — python-sdk DeepWiki — Python SDK 中 report_progress 与日志方法的使用详解
Add Progress Bars and Logging Notifications to Your MCP Server — zazencodes — 用 Inspector 实测 notification/message 与进度条的可视化实战博客

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

概述

前置条件

一、为什么长任务需要"三个通知原语"

二、步骤一：Logging 原语——让服务器开口说话

协议层要点

代码实现（FastMCP）

预期结果

常见问题与解决

三、步骤二：Progress 原语——上报"完成度"

协议层要点

代码实现（FastMCP）

预期结果

常见问题与解决

四、步骤三：Cancellation 原语——优雅中止

协议层要点

代码实现（FastMCP）

预期结果

常见问题与解决

五、实战案例：批量日志分析服务器

完整代码（log_analyzer.py）

接入 Claude Code 测试

用 Inspector 实测三原语

六、工作流程：一次长任务调用的完整交互

七、三原语选型决策

八、延伸阅读

训练营适用性评估

参考来源

📎 本分类最新文章

完整代码（`log_analyzer.py`）