<!-- title: 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 规范）：

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

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

```json
{
  "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：

```python
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（进度令牌）**：

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

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

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

```json
{
  "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 并封装好通知：

```python
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`：

```json
{
  "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` 捕获取消：

```python
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`）

```python
"""
批量日志分析 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`）里注册：

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

```bash
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` 的完整往返。

---

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

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

```mermaid
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，形成可演示的作业。

---

## 参考来源

1. [Progress — Model Context Protocol](https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/progress) — 官方进度上报规范，progressToken、notifications/progress 的权威定义（2025-06-18 版本）
2. [Logging — Model Context Protocol](https://modelcontextprotocol.io/specification/2025-06-18/server/utilities/logging) — 官方日志规范，RFC 5424 八级、logging/setLevel、notifications/message、安全要求
3. [Cancellation — Model Context Protocol](https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/cancellation) — 官方取消规范，notifications/cancelled、竞态处理、best-effort 语义
4. [Progress & Notifications — python-sdk DeepWiki](https://deepwiki.com/modelcontextprotocol/python-sdk/5.4-progress-and-notifications) — Python SDK 中 report_progress 与日志方法的使用详解
5. [Add Progress Bars and Logging Notifications to Your MCP Server — zazencodes](https://zazencodes.substack.com/p/add-progress-bars-and-logging-notifications) — 用 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学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)