首页 › tutorials › MCP长任务进度通知与取消实战

概述

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

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

前置条件

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

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

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 里不需要回应的消息）。它们不改变工具的最终返回值，但让长任务变得透明、可控。理解这三者，是本文全部内容的核心。

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 由客户端下发。客户端不传，服务器就没法回报进度。Claude Code 和 Inspector 都会自动传。

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

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

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

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

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

新建目录并初始化：

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 和启动脚本：

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

新建 server.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）等。利用它发日志：

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。我们用它循环推送进度：

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 上挂监听。一个干净的做法是用一个共享的"取消集合"：

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

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

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

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

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 上调方便得多。启动：

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

{
  "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 截断、可取消：

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() 立刻释放文件句柄。

工作流程总览

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[客户端日志面板]

进阶建议

延伸阅读

训练营适用性评估

参考来源

  1. Progress - Model Context Protocol — 进度通知 notifications/progress 的官方协议定义与字段规范
  2. Cancellation - Model Context Protocol — 取消通知 notifications/cancelled 的语义与"尽力而为"约定
  3. Add Progress Bars and Logging Notifications to Your MCP Server — 实战向：给 MCP 服务器加进度条与日志通知的代码示例
  4. GitHub - modelcontextprotocol/inspector — 官方可视化调试工具，本文 Inspector 端到端调试章节的依据
  5. Logging, Progress Tracking, and Cancellation (Quarkus MCP) — 三原语配套讲解，可与 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学通信知识星球海报-扫码加入

📎 本分类最新文章