<!-- title: MCP长任务进度通知与取消实战 -->

## 概述

绝大多数 MCP 教程只演示"工具调用—返回结果"这种瞬时交互。可一旦你的工具要做真正耗时的事——扫描几个 G 的日志、批量处理文件、跑一个慢脚本——客户端就会面对一个黑盒：进度多少了？还要等多久？能不能中途叫停？本文聚焦 MCP 协议里三个被低估的通知原语——`notifications/progress`（进度）、`notifications/cancelled`（取消）、`notifications/message`（日志）——带你从零写一个"会汇报、能打断"的 MCP 服务器，并用官方 MCP Inspector 做端到端调试。

**适用人群**：已经写过最简单的 stdio MCP 工具、想进阶到可观测、可中断长任务的开发者。

## 前置条件

- Node.js 18+ 和 npm/npx
- 会写基本的 TypeScript
- 了解 JSON-RPC 2.0 的"请求—响应—通知"概念
- 装好 Claude Code 用于最终联调（非必须，可用 Inspector 替代）

---

## 第一步：理解问题——为什么需要通知原语

先看一段"反例"。下面这个 MCP 工具会逐行扫描一个日志文件，统计 ERROR 出现的次数：

```ts
server.tool(
  "count_errors",
  { path: z.string() },
  async ({ path }) => {
    const content = await fs.readFile(path, "utf8");
    const lines = content.split("\n");
    let count = 0;
    for (const line of lines) lines.includes("ERROR") && count++;
    return { content: [{ type: "text", text: `共 ${count} 条 ERROR` }] };
  }
);
```

它功能没错，但有三类工程问题：

1. **黑盒等待**：客户端调用后，在拿到结果前看不到任何中间信号。一个 2GB 的日志可能要扫几十秒，用户体验等同于卡死。
2. **不可中断**：扫到一半发现路径选错了，没办法取消，CPU 白烧。
3. **无可观测性**：服务器内部走到哪一步、命中了什么异常，调用方一无所知，事后排障只能靠 `console.log`。

MCP 协议为此内置了三种**单向通知**（notification，JSON-RPC 里不需要回应的消息）。它们不改变工具的最终返回值，但让长任务变得透明、可控。理解这三者，是本文全部内容的核心。

```mermaid
sequenceDiagram
    participant Client as 客户端(Claude Code/Inspector)
    participant Server as MCP 服务器
    Client->>Server: tools/call (含 _meta.progressToken=42)
    Note over Server: 开始长任务
    Server-->>Client: notifications/progress (token=42, progress=30, total=100)
    Server-->>Client: notifications/message (level=info, "已扫描1000行")
    Server-->>Client: notifications/progress (token=42, progress=60)
    Client->>Server: notifications/cancelled (requestId=req-9, reason="用户中止")
    Server-->>Client: (请求被中止，不再返回结果)
    Note over Client,Server: 最终结果只发一次
```

## 第二步：搞清楚三个通知的参数

在写代码之前，把协议字段记牢，调试时才看得懂 Inspector 里的报文。

**进度通知 `notifications/progress`**：服务器主动发，参数为

- `progressToken`：客户端在原始请求 `_meta.progressToken` 里塞进来的标识（数字或字符串）
- `progress`：当前进度数值（如已处理行数）
- `total`（可选）：总量，客户端可据此算百分比
- `message`（可选）：人类可读说明

关键点：**progressToken 由客户端下发**。客户端不传，服务器就没法回报进度。Claude Code 和 Inspector 都会自动传。

**取消通知 `notifications/cancelled`**：任一方可发，参数为

- `requestId`：要取消的那个原始请求的 id
- `reason`（可选）：取消原因字符串

取消是**尽力而为**（best-effort）：对方收到后应停止工作并**不再返回结果或错误**。它不是强制指令，服务器可以"假装没听见"继续跑完。

**日志通知 `notifications/message`**：服务器发，参数为

- `level`：syslog 风格的级别，从低到高 `debug`/`info`/`notice`/`warning`/`error`/`critical`/`alert`/`emergency`
- `logger`（可选）：日志源名称
- `data`：任意结构化内容（字符串、对象都行）

客户端可通过 `logging/setLevel` 请求调节 verbosity，所以你大胆往里塞 `debug` 日志，不影响线上安静度。

## 第三步：搭一个能跑的项目骨架

新建目录并初始化：

```bash
mkdir mcp-longtask && cd mcp-longtask
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node tsx
npx tsc --init
```

在 `package.json` 里加 type 和启动脚本：

```json
{
  "type": "module",
  "scripts": {
    "dev": "tsx server.ts",
    "inspector": "npx @modelcontextprotocol/inspector npm run dev"
  }
}
```

新建 `server.ts`，先写一个会"假装慢"但还不报进度的工具，确保链路通：

```ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new McpServer({ name: "longtask-demo", version: "0.1.0" });

server.tool(
  "ping_slow",
  "模拟一个慢任务，固定等待3秒",
  {},
  async () => {
    await new Promise((r) => setTimeout(r, 3000));
    return { content: [{ type: "text", text: "pong" }] };
  }
);

await server.connect(new StdioServerTransport());
```

跑 `npm run inspector`，浏览器会自动打开 Inspector 面板。点 **Connect**，左侧能看到 `ping_slow` 工具，点 **Run Tool** 等 3 秒看到 "pong"。链路打通后，下面逐步加料。

## 第四步：先加日志——让服务器"会说话"

McpServer 高层 API 通过工具处理函数的**第二参数**暴露一个上下文对象，里面有 `sendNotification`、`meta`（含 progressToken）等。利用它发日志：

```ts
server.tool(
  "count_errors",
  "统计日志中 ERROR 行数，过程中输出日志",
  { path: z.string() },
  async ({ path }, ctx) => {
    const log = (level: any, data: unknown) =>
      ctx.sendNotification({
        method: "notifications/message",
        params: { level, logger: "scanner", data },
      });

    log("info", { msg: "开始扫描", path });
    const content = await fs.readFile(path, "utf8");
    const lines = content.split("\n");
    log("debug", { msg: "读取完毕", lines: lines.length });

    let count = 0;
    for (const line of lines) if (line.includes("ERROR")) count++;

    log("info", { msg: "扫描完成", count });
    return { content: [{ type: "text", text: `共 ${count} 条 ERROR` }] };
  }
);
```

回到 Inspector，运行工具时切到 **Notifications** 面板，能看到一条条 `notifications/message` 实时滚动。这就是服务器的"活日志"，比 `console.log` 强在：它走协议通道，**远程 MCP 也能收到**，而且客户端可以用 `logging/setLevel` 按级别过滤。

## 第五步：再加进度——让客户端画出进度条

进度通知需要一个 `progressToken`。高层 API 已经从客户端请求里帮你取出来，挂在 `ctx.meta.progressToken`。我们用它循环推送进度：

```ts
server.tool(
  "count_errors",
  "统计日志中 ERROR 行数，实时回报进度",
  { path: z.string() },
  async ({ path }, ctx) => {
    const content = await fs.readFile(path, "utf8");
    const lines = content.split("\n");
    const total = lines.length;
    let count = 0;

    for (let i = 0; i < total; i++) {
      if (lines[i].includes("ERROR")) count++;
      // 每 1000 行或最后一行报一次，避免通知风暴
      if (i % 1000 === 0 || i === total - 1) {
        ctx.sendNotification({
          method: "notifications/progress",
          params: {
            progressToken: ctx.meta?.progressToken,
            progress: i + 1,
            total,
            message: `已处理 ${i + 1}/${total} 行`,
          },
        });
        // 关键：用 await 让事件循环有机会把通知发出去、并响应取消
        await new Promise((r) => setImmediate(r));
      }
    }

    return { content: [{ type: "text", text: `共 ${count} 条 ERROR` }] };
  }
);
```

两个工程细节容易被忽略，值得单独说：

**节流（throttle）很重要。** 如果每行都发一条 `notifications/progress`，一个百万行的文件会狂发百万条消息，把 stdio 管道或 SSE 连接打爆。上面的代码每 1000 行才报一次，是务实的折中。原则是：**进度更新的频率，匹配人眼感知**（每秒几次即可），而不是匹配数据粒度。

**`await new Promise(r => setImmediate(r))` 不是摆设。** Node 的事件循环里，一个纯同步的 `for` 循环会把整个进程占死，通知根本发不出去，取消也无法响应。`setImmediate` 把控制权交还事件循环一次，让通知有机会 flush、让取消信号有机会被读到。

## 第六步：最后加取消——让长任务可中断

取消是双向通知：客户端发 `notifications/cancelled`，服务器要**监听**它并主动停止。高层 API 没有直接暴露取消事件，我们需要在连接的 transport/server 上挂监听。一个干净的做法是用一个共享的"取消集合"：

```ts
// 模块级：记录被取消的 requestId
const cancelled = new Set<string | number>();

server.server.onerror = console.error;
// 监听客户端发来的通知
server.server.sendNotification; // 占位，真正监听用下面方式

// 在创建 transport 前，拦截 setRequestHandler 的通知
// 更简洁：用 transport 的 onmessage，或直接用底层 Server
```

高层 API 对取消的支持不够顺，因此涉及取消时，**降到底层 Server API 更清晰**。下面是完整改造：

```ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

const lowServer = new Server(
  { name: "longtask-demo", version: "0.1.0" },
  { capabilities: { tools: {} } }
);

// 注册"取消"处理器
lowServer.setRequestHandler(
  { method: "notifications/cancelled" } as any,
  async (req) => {
    const id = (req.params as any)?.requestId;
    if (id !== undefined) cancelled.add(id);
  }
);

lowServer.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "count_errors",
      description: "统计日志中 ERROR 行数，可取消",
      inputSchema: {
        type: "object",
        properties: { path: { type: "string" } },
        required: ["path"],
      },
    },
  ],
}));

lowServer.setRequestHandler(CallToolRequestSchema, async (req) => {
  if (req.params.name !== "count_errors") {
    throw new Error("未知工具");
  }
  const path = (req.params.arguments as any).path;
  const token = (req.params as any)._meta?.progressToken;
  const reqId = req.id as string | number;

  const content = await fs.readFile(path, "utf8");
  const lines = content.split("\n");
  const total = lines.length;
  let count = 0;

  for (let i = 0; i < total; i++) {
    // 每次循环都检查取消标志
    if (cancelled.has(reqId)) {
      // 协议要求：被取消后不要再返回结果，直接 throw 中断即可
      throw new Error("__CANCELLED__");
    }
    if (lines[i].includes("ERROR")) count++;
    if (i % 1000 === 0 || i === total - 1) {
      await lowServer.notification({
        method: "notifications/progress",
        params: { progressToken: token, progress: i + 1, total },
      });
      await new Promise((r) => setImmediate(r));
    }
  }
  cancelled.delete(reqId);
  return {
    content: [{ type: "text", text: `共 ${count} 条 ERROR` }],
  };
});

await lowServer.connect(new StdioServerTransport());
```

注意 `throw new Error("__CANCELLED__")` 这一行。**协议规定：请求被取消后，服务器不应再返回正常结果或错误。** 上面用抛异常只是为了跳出循环；严谨做法是直接 `return` 一个空响应或用一个标志让 transport 层丢弃。实际工程中，更稳妥的是把"被取消"作为一种控制流而非异常，配合一个 `AbortController` 传递到下游真正的 I/O（如 fetch、子进程）。

## 第七步：用 MCP Inspector 做端到端调试

Inspector 是协议级调试的神器，比直接挂在 Claude Code 上调方便得多。启动：

```bash
npm run inspector
# 等价于: npx @modelcontextprotocol/inspector npm run dev
```

调试长任务的标准动作：

1. **Connect** 连上服务器，确认 `count_errors` 出现在工具列表。
2. 顶部 **Configuration** 区可设环境变量、超时、传输类型（stdio / SSE / HTTP）。
3. 点开工具，填 `path` 指向一个较大的日志文件，点 **Run Tool**。
4. 切到 **Notifications** 面板：能实时看到 `notifications/progress`（progress/total 在涨）和 `notifications/message` 日志。
5. **测取消**：再次运行，趁进度没跑完，点 Inspector 上的取消按钮（或手动从另一个连接发 `notifications/cancelled`）。观察服务器是否在几秒内停下、不再返回结果。

> **常见问题**：运行后一条 `notifications/progress` 都收不到？多半是没 `await setImmediate`，通知全堵在缓冲区里，要等整个循环结束才一次性 flush。先加上那行 `await`，90% 的"进度不显示"问题就消失了。

> **常见问题**：Inspector 提示连接超时？检查 `tsconfig.json` 的 `module` 设为 `NodeNext`、`target` 设为 `ES2022`，以及 `package.json` 的 `"type": "module"`，否则 `tsx` 加载 ESM 包会报错。

## 第八步：接入 Claude Code 做联调

Inspector 调通后，把它接到真实客户端。编辑 `~/.claude.json`（或项目级 `.mcp.json`）：

```json
{
  "mcpServers": {
    "longtask": {
      "command": "npm",
      "args": ["run", "dev"],
      "cwd": "/绝对路径/mcp-longtask"
    }
  }
}
```

在 Claude Code 里输入 `/mcp` 应能看到 `longtask` 服务器为 connected。然后自然语言调用："帮我统计 /var/log/app.log 里有多少 ERROR"。Claude Code 会自动把 `progressToken` 下发，调用过程中你会在界面看到进度，中途说"停下"即可触发取消。

---

## 实战案例：可中断的大日志分析服务器

把前面三步合起来，做一个稍完整的例子——`log-analyzer`，它不只数 ERROR，还按级别汇总、支持 `maxLines` 截断、可取消：

```ts
lowServer.setRequestHandler(CallToolRequestSchema, async (req) => {
  const { path, maxLines = Infinity } = req.params.arguments as any;
  const token = (req.params as any)._meta?.progressToken;
  const reqId = req.id as number;

  const stream = fs.createReadStream(path, { encoding: "utf8" });
  const rl = readline.createInterface({ input: stream });

  const stats: Record<string, number> = {};
  let processed = 0;
  const total = (await fs.stat(path)).size; // 用字节数做总量参考

  for await (const line of rl) {
    if (cancelled.has(reqId)) {
      stream.destroy();
      return { content: [{ type: "text", text: "已取消" }] };
    }
    if (processed >= maxLines) break;

    const m = line.match(/\b(ERROR|WARN|INFO|DEBUG)\b/);
    if (m) stats[m[1]] = (stats[m[1]] ?? 0) + 1;

    if (processed % 5000 === 0) {
      await lowServer.notification({
        method: "notifications/progress",
        params: {
          progressToken: token,
          progress: processed,
          total: total,
          message: `已处理 ${processed} 行`,
        },
      });
      await lowServer.notification({
        method: "notifications/message",
        params: { level: "debug", logger: "analyzer", data: { processed, stats } },
      });
      await new Promise((r) => setImmediate(r));
    }
    processed++;
  }

  cancelled.delete(reqId);
  return {
    content: [
      { type: "text", text: `扫描 ${processed} 行，统计：\n${JSON.stringify(stats, null, 2)}` },
    ],
  };
});
```

这里把"逐行 `readFile` 拆数组"升级成**流式 `createReadStream` + readline**，内存占用从 O(文件大小) 降到 O(单行)，这才是处理大日志的正道。进度用行数、总量用字节数（单位不匹配但不影响"在动"的观感），取消时 `stream.destroy()` 立刻释放文件句柄。

## 工作流程总览

```mermaid
flowchart TD
    A[客户端调用工具<br/>下发 progressToken] --> B{服务器循环处理}
    B -->|每 N 步| C[发 notifications/progress]
    B -->|命中关键点| D[发 notifications/message 日志]
    B --> E{收到 cancelled?}
    E -->|是| F[destroy 资源<br/>停止、不返回结果]
    E -->|否| G{处理完?}
    G -->|否| B
    G -->|是| H[返回最终结果]
    C --> I[客户端画进度条]
    D --> J[客户端日志面板]
```

## 进阶建议

- **配合 `AbortController`**：把取消信号转成 `AbortSignal` 传给 `fetch`、子进程、数据库查询，让取消真正生效到底层 I/O，而不只是跳出 JS 循环。
- **远程 MCP 的取消语义**：SSE/HTTP 传输下，取消靠同一个 JSON-RPC 通道下发，逻辑一致，但要注意连接断开本身就应被视为隐式取消，服务器要监听 transport 的 `close` 事件。
- **通知不要带敏感数据**：`notifications/message` 的 `data` 会被客户端记录（Inspector 会全文显示），日志里别塞 token、密钥。
- **给工具标注 readOnlyHint**：纯统计类工具用注解声明 `readOnlyHint: true`，客户端会自动放行、少弹确认，但注意这只声明不强制。

## 延伸阅读

- MCP 官方规范——Progress / Cancellation / Logging 三节，字段最权威
- `@modelcontextprotocol/inspector` 仓库，了解面板新特性与传输调试
- Anthropic 的 "MCP for Developers" 系列中关于长任务与流式的章节
- `modelcontextprotocol/servers` 仓库里的 `server-everything` 参考实现，演示了几乎所有进阶特性

## 训练营适用性评估

- **适合的营型**：中高级 MCP 开发训练营、AI Agent 工具链工程化实战营。不适合纯入门营，需学员先有"能跑通的最简 MCP 工具"经验。
- **预计教学时长**：理论 + 跟做约 2.5 小时（含 Inspector 调试环节）；完整跑通实战案例再加 1 小时。
- **动手比重**：高。每个步骤都有可运行代码和 Inspector 验证点，学员能即时看到进度条/日志/取消的真实效果，反馈闭环短、成就感强。
- **可拓展作业**：让学员把取消信号接 `AbortController` 透传到一个真实 `fetch` 子任务；或把流式统计改成增量返回（每 5000 行发一次 `notifications/message` 的中间摘要），体会"通知 vs 分块结果"的区别。

## 参考来源

1. [Progress - Model Context Protocol](https://modelcontextprotocol.io/specification/2025-03-26/basic/utilities/progress) — 进度通知 `notifications/progress` 的官方协议定义与字段规范
2. [Cancellation - Model Context Protocol](https://modelcontextprotocol.io/specification/2025-03-26/basic/utilities/cancellation) — 取消通知 `notifications/cancelled` 的语义与"尽力而为"约定
3. [Add Progress Bars and Logging Notifications to Your MCP Server](https://zazencodes.substack.com/p/add-progress-bars-and-logging-notifications) — 实战向：给 MCP 服务器加进度条与日志通知的代码示例
4. [GitHub - modelcontextprotocol/inspector](https://github.com/modelcontextprotocol/inspector) — 官方可视化调试工具，本文 Inspector 端到端调试章节的依据
5. [Logging, Progress Tracking, and Cancellation (Quarkus MCP)](https://docs.quarkiverse.io/quarkus-mcp-server/dev/guides-logging-progress-cancellation.html) — 三原语配套讲解，可与 TS 实现互相印证字段含义

> [!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)