MCP 事件模型：通知类型、消息格式、幂等性

1. 为什么需要 MCP 事件

到目前为止，ChatGPT 与你的后端之间的交互几乎都是 RPC：模型调用工具，工具做点事并返回结果——搞定。对于 200–500 ms、最多几秒的短操作，这很方便。

但一旦出现长耗时任务——例如为 GiftGenius 分析一份很大的员工偏好文件、从一堆外部 API 聚合推荐、重算一个庞大的 feed——事情就变得不那么美妙。HTTP 超时、函数重启、永远转的加载动画，用户坐在那里猜：“它还活着，还是已经挂了？”

这时事件模型就该登场了。与其维持一次长时间的工具调用，不如启动一个任务，拿到 jobId，接下来由服务器主动推送事件：已开始、进度更新、完成、失败。MCP 中的这些事件实现为 JSON-RPC notifications——单向消息，没有 id，不期待响应。

重要的一点：事件不是“线上 console.log”。它是具有固定模式的协议消息，你的 UI（小部件）和/或代理必须像处理工具调用结果那样严谨地处理它。

回顾：MCP 中的消息类型

在继续之前，先简要回顾一下 MCP 里到底有哪些消息类型。

撇开营销层不谈，MCP 基于 JSON-RPC 2.0。它有三类基础消息：请求、响应和通知。

不赘述列表，我们直接看一张小对比表：

MCP 事件正对应第三行：notifications。其特征：

• 顶层没有 id——不会有 result 或 error 作为回应；
• 发起方不等待 ACK——在协议层面“发射即忘”；
• 可靠性不是靠确认，而是靠处理器的幂等性与重试策略来构建。

重要限制：MCP 事件不会“随时随地在太空中乱飞”。它们存在于已建立的 MCP 连接之内，承载在具体的传输之上。最常见的是类似 SSE 的流（传输细节及其变体我们会在单独一讲中讨论）。

2. 什么是“实际的 MCP 事件”

从形式上说，MCP 事件就是 JSON-RPC notification，即如下对象：

{
  "jsonrpc": "2.0",
  "method": "notifications/job/progress",
  "params": {
    "jobId": "job_123",
    "percentage": 30,
    "stage": "在目录中查找备选",
    "eventId": "evt_abc123",
    "timestamp": "2025-11-21T10:15:00Z"
  }
}

这里有几个要点：

1. 在 method 字段中，我们编码事件类型及其“命名空间”。MCP 已经定义了一些标准的 notifications/... 方法用于日志、进度与资源变更，但你可以、也应该添加自己的业务专用方法，比如 notifications/job/progress 或 notifications/job/completed。
2. 所有业务数据位于 params。我们会在那里保存任务标识（jobId）、事件唯一 ID（eventId）、时间（timestamp）、人类可读的消息等。
3. 顶层没有 id 字段——这正是它是 notification 的原因。协议不提供对此的响应。如果服务器想知道“对方是否理解了”，可以再发一个事件，或等待客户端的后续反应（例如新的请求）。但就 JSON-RPC 而言，没有 ACK。

在心智模型上可以这样理解：工具调用 tools/call 是“你等待回复的信件”，而事件则是“来自 Slack 机器人的通知：“后台任务 #123 已完成””。

3. 事件分类：有哪些通知

如果只是放开“随便把 JSON 当 notifications 发”，两周后系统就会变成垃圾场：事件名称各不相同、字段到处漂移、UI 不知道如何处理。因此最好先约定一个小而清晰的分类法。

下面是一种与 MCP 规范以及 ChatGPT Apps 实际用例非常契合的分类方式。

作业生命周期事件（Job Lifecycle）

这类事件反映任务状态的关键跃迁。通常任务有一个状态机（state machine），如 pending → running → (completed | failed | canceled)。

典型事件：

• job.created —— 任务已注册；
• job.started —— worker 已开始执行；
• job.completed —— 任务成功完成；
• job.failed —— 任务失败并带错误；
• job.canceled —— 任务被用户取消。

GiftGenius 的 job.completed 示例：

{
  "jsonrpc": "2.0",
  "method": "notifications/job/completed",
  "params": {
    "eventId": "evt_gg_100",
    "jobId": "giftjob_42",
    "timestamp": "2025-11-21T10:20:00Z",
    "summary": "礼物推荐已完成",
    "resultResourceId": "resource:gifts:giftjob_42"
  }
}

这里的 resultResourceId 可以指向一个 MCP 资源，后续由小部件或代理读取。

进度事件（Progress Updates）

这些是生命周期中的“小步伐”：它们不改变最终状态，但能让用户感知到“事情在发生”。

典型的 job.progress 事件：

{
  "jsonrpc": "2.0",
  "method": "notifications/job/progress",
  "params": {
    "eventId": "evt_gg_101",
    "jobId": "giftjob_42",
    "timestamp": "2025-11-21T10:18:30Z",
    "percentage": 40,
    "stage": "按预算筛选礼物",
    "etaSeconds": 25
  }
}

这里要注意，percentage 应当合理地单调增加，趋向 100，而不是来回跳。为“进度”字段选一个名字（例如 percentage）并在所有事件中保持一致。MCP 官方的进度工具也有规则：进度只会增长。

数据/资源更新事件（Resource/Data events）

有时用户甚至不关心具体的 jobId。更重要的是某个实体发生了变化：商品 feed 更新了、生成了新的报表快照、重新生成了个人画像。

MCP 已提供标准级别的通知 resources/updated、resources/list_changed 等，告知客户端“重读资源列表，有变化”。

对 GiftGenius 来说，可能如下：

{
  "jsonrpc": "2.0",
  "method": "resources/updated",
  "params": {
    "eventId": "evt_feed_17",
    "timestamp": "2025-11-21T09:00:00Z",
    "resourceId": "resource:product-feed",
    "changeType": "snapshot_ready"
  }
}

小部件收到此类事件后，可以例如高亮“更新礼物列表”的按钮。

UX 与系统级事件

还有一些不严格属于业务，但对 UX 或诊断很重要的事件：

• 日志消息 logging/message —— MCP 的标准日志通知；
• heartbeat/ping —— 服务器的周期性“我还活着”；
• 退化告警：例如“外部 API 当前较慢，结果可能延迟”。

这类事件有助于监控与调试；有时也可在 UI 中以更温和的方式呈现，让用户知道系统没死，只是忙。

4. 事件结构：必填字段与负载（payload）

事件也是一种 API 对象，和工具请求一样需要设计。一个好习惯是先约定一组基础字段。

从概念上，将事件划分为三部分更有用：元数据、关联信息和有效载荷。

通用形式示例：

{
  "jsonrpc": "2.0",
  "method": "notifications/job/progress",
  "params": {
    "eventId": "evt_gg_103",
    "type": "job.progress",
    "timestamp": "2025-11-21T10:19:00Z",
    "jobId": "giftjob_42",
    "payload": {
      "percentage": 60,
      "stage": "对比评价",
      "etaSeconds": 15
    }
  }
}

在这套结构中可以辨识出：

• eventId —— 事件的唯一标识，用于客户端去重；
• type —— 事件的逻辑名称（可与 method 对应/规范化）；
• timestamp —— 事件由服务器生成的时间；
• jobId 或其他 correlation-id —— 用于确定事件归属；
• payload —— 实际数据，不同事件类型有不同的结构。

在真实系统中，你几乎肯定会通过 JSON Schema 或至少 TypeScript 类型形式化描述这些结构，方便服务端与客户端校验消息。有些团队会采用受 CloudEvents 启发的格式：其中也有标准字段 id、source、type、time 等。

但核心思想很简单：事件应当机器可读并且一致——不要出现“有时叫 jobId，有时叫 job_id，有时干脆没有”的惊喜。

在下面的示例中，为避免代码过于臃肿，我们更常使用“扁平化”版本：事件数据直接位于 params 中，而不再嵌套 payload，并且如果 method 已承担其角色，则有时省略 type。原则不变：每个事件都应有稳定的元数据（eventId、jobId、timestamp）和可预测的有效载荷。

5. 事件的幂等性：为何以及如何实现

现在到本讲最重要的词——幂等性。

事件处理器的幂等性意味着：无论同一事件被处理一次还是十次，系统的最终状态都保持正确。在伴随网络与重试的分布式系统中，这几乎是生死攸关的问题。

为什么同一个事件会多次到达？

原因很多：从连接断开与重连这种常见情况，到服务器端“以防万一”重复发送通知的重试策略。在使用流式协议时（例如服务器向打开的连接中主动推送事件，类似 SSE——我们会在单独的传输篇中详谈），这是经典场景：客户端带着 Last-Event-ID 重连，服务器会补发缺失事件，客户端可能第二次看到其中的一些。

如果处理器不具备幂等性，就会出现奇怪现象：

• job.completed 事件导致重复发放奖励或把订单状态改了两次；
• resource.updated 事件让小部件每次都“追加”卡片，造成 UI 重复；
• 重复的 job.progress 会让用户困惑，比如进度条来回跳动。

正确策略分两层：服务端生成事件的方式，以及客户端处理事件的方式。

服务端：稳定的 ID 与状态机

服务器应当：

• 为每个逻辑事件生成唯一的 eventId；
• 保证同一 jobId 的事件形成有效的状态序列：不能在 job.completed 之后再发 job.failed，也不能发两个结果不同的 job.completed。

也就是说你实际上有一个任务状态机，每个事件都是一个允许的转换。

客户端：去重与“温和”更新

客户端（小部件、代理或其他组件）应当：

• 在至少当前连接/会话的生命周期内，保存已处理的 eventId 集合；
• 处理前检查：如果 eventId 已见过，则直接忽略或仅重绘 UI 而不产生副作用；
• 当收到更改任务状态的事件（job.completed、job.failed）时，确保转换是允许的：例如如果任务已标记为 completed，重复的 job.completed 不应改变任何状态，而 failed 则更应作为不合法而忽略。

commerce 领域的经典例子：处理支付确认的 webhook。同一个 order.paid 很可能到达两次；因此后端会保存 paymentId 以及“已入账”的标志。即使 webhook 第二次到来，订单状态也不会改变。MCP 事件应当用同样的思维来设计。

6. 示例：为 GiftGenius 设计事件

把这些应用到我们的教学案例 GiftGenius。设想一个长流程：用户上传一份包含员工及其兴趣的大 CSV，要求“为所有人挑选礼物创意”。这个操作可能需要几十秒。

一个合理的事件模型可以这样描述：

1. 用户调用工具 start_bulk_gift_analysis。工具返回 jobId："bulk_2025_001"。
2. MCP 服务器创建任务，并几乎立刻发送 job.started 携带简短说明。
3. 随执行过程发送若干 job.progress 阶段：
   • 10% —— “解析文件并校验格式”；
   • 40% —— “提取兴趣与部门”；
   • 70% —— “按类别匹配礼物”；
   • 100% —— 临近完成。
4. 最后收到 job.completed，其中包含指向最终推荐结果资源的链接。
5. 如果出错——则不是 completed，而是带错误码的 job.failed，并可附带修复建议。

非正式地说流程就是这样，但我们把两类关键事件 job.progress 和 job.completed 固化成 JSON 模式。伪 JSON Schema（简化版）：

{
  "job.progress": {
    "type": "object",
    "properties": {
      "eventId": { "type": "string" },
      "jobId": { "type": "string" },
      "timestamp": { "type": "string", "format": "date-time" },
      "percentage": { "type": "number", "minimum": 0, "maximum": 100 },
      "stage": { "type": "string" },
      "etaSeconds": { "type": "number" }
    },
    "required": ["eventId", "jobId", "timestamp", "percentage", "stage"]
  }
}

{
  "job.completed": {
    "type": "object",
    "properties": {
      "eventId": { "type": "string" },
      "jobId": { "type": "string" },
      "timestamp": { "type": "string", "format": "date-time" },
      "summary": { "type": "string" },
      "resultResourceId": { "type": "string" }
    },
    "required": ["eventId", "jobId", "timestamp", "resultResourceId"]
  }
}

你不必立刻实现完整的模式校验，但在脑海中保持这种结构很有帮助：它能避免把字段“抹”到不同格式、也能提醒你别忘了关键元数据。

7. 迷你实战：发送 MCP 事件的服务器

现在把理论与一小段 TypeScript 伪代码连起来。我们不会深入真实的 MCP 库（其一，它们仍在演进；其二，这一讲的重点是模型），而是画一个结构骨架。

假设我们的 MCP 服务器里有一个抽象的 sendNotification，它能把 JSON-RPC notification 发回 ChatGPT。伪接口：

// 用于发送 MCP 通知的工具函数
async function sendNotification(
  method: string,
  params: Record<string, unknown>
) {
  // 在这里序列化 JSON 并通过活动的 MCP 连接发送出去
}

现在实现工具 start_bulk_gift_analysis 的处理器。它注册任务，返回 jobId，同时在后台“滴答作响”并发送进度。在真实世界中，这会由 worker 和队列承担，这里先用定时器代替。

type Job = {
  id: string;
  status: "pending" | "running" | "completed" | "failed";
};

const jobs = new Map<string, Job>();

export async function startBulkGiftAnalysisTool() {
  const jobId = `bulk_${Date.now()}`;
  jobs.set(jobId, { id: jobId, status: "pending" });

  // 立即发送 job.started
  await sendNotification("notifications/job/started", {
    eventId: `evt_${jobId}_started`,
    jobId,
    timestamp: new Date().toISOString(),
    summary: "已启动大名单礼物分析"
  });

  simulateJob(jobId); // 在“后台”启动任务

  return { jobId };
}

任务模拟本身：

async function simulateJob(jobId: string) {
  jobs.set(jobId, { id: jobId, status: "running" });

  const stages = [
    { percent: 10, stage: "解析 CSV" },
    { percent: 40, stage: "分析兴趣" },
    { percent: 70, stage: "挑选礼物" },
    { percent: 100, stage: "生成结果" }
  ];

  for (const s of stages) {
    await sendNotification("notifications/job/progress", {
      eventId: `evt_${jobId}_${s.percent}`,
      jobId,
      timestamp: new Date().toISOString(),
      percentage: s.percent,
      stage: s.stage
    });
    await new Promise(r => setTimeout(r, 1000));
  }

  jobs.set(jobId, { id: jobId, status: "completed" });

  await sendNotification("notifications/job/completed", {
    eventId: `evt_${jobId}_done`,
    jobId,
    timestamp: new Date().toISOString(),
    summary: "礼物分析已完成",
    resultResourceId: `resource:gifts:${jobId}`
  });
}

代码故意保持简单，但它清楚地展示了：

• 我们采用 started → progress* → completed 的事件序列；
• 每个事件都有唯一的 eventId；
• 所有事件都绑定到同一个 jobId。

未来当你加入真正的队列与 worker 时，事件结构基本不变——只是 sendNotification 被调用的位置会改变。

8. 客户端：最简单的幂等事件处理器

在客户端（比如你的 Apps SDK 小部件）需要学会接收这些事件，把它们与当前任务关联起来，并且在遇到重复时不“发疯”。

先不深入传输层（后面再讲），设想有一个 onMcpNotification 函数：每当收到 notification，你的 MCP 客户端层就会调用它。

加上最简单的去重：

const processedEvents = new Set<string>();

function handleNotification(method: string, params: any) {
  const eventId = params.eventId as string | undefined;
  if (!eventId) return; // 这点很有争议，但示例里就先这样

  if (processedEvents.has(eventId)) {
    // 重复——忽略或温和地更新 UI
    return;
  }
  processedEvents.add(eventId);

  if (method === "notifications/job/progress") {
    updateJobProgress(params.jobId, params.percentage, params.stage);
  } else if (method === "notifications/job/completed") {
    markJobCompleted(params.jobId, params.resultResourceId);
  }
}

updateJobProgress 与 markJobCompleted 的实现就是纯粹的 React/UI 代码：

function updateJobProgress(jobId: string, percent: number, stage: string) {
  // 例如，写入 Zustand/Redux/React state
  console.log(`Job ${jobId}: ${percent}% — ${stage}`);
}

function markJobCompleted(jobId: string, resourceId: string) {
  console.log(`Job ${jobId} 已完成，资源: ${resourceId}`);
}

这样的处理器：

• 即使事件到达两次也不会出错；
• 不会产生副作用（比如“第二次弹出‘完成！’的模态框”）；
• 为更复杂的逻辑铺路，例如校验允许的状态转换（不允许在已 completed 的状态上再应用 failed）。

在生产代码中，你很可能会在与 MCP 服务器重连时清空 processedEvents，并且不仅保存 eventId，还会保存每个 jobId 的当前状态，以便在出现奇怪的事件序列时更合理地处理。

接下来要理解这些 MCP 事件如何穿过代理/小部件，变成具体的用户体验：进度条、阶段展示、最终结果出现。我们继续看事件与 run/workflow 以及 UX 的关联。

9. 事件、run/workflow 与 UX 的联动

尽管我们已经有一讲专门讨论 workflow 与代理，但现在你会看到完整画面。我们已经引入了事件家族（job.*、resource.*、系统事件）；接下来看看它们如何穿过代理/小部件与 ChatGPT，最终变成具体的用户体验。

一个典型的长任务场景是：ChatGPT 调用 MCP 工具得到 jobId；随后服务器围绕该 jobId 推送进度、完成或失败事件；你的小部件或代理逻辑据此更新 UI 并做出决策。

用时序图可以这样表示：

sequenceDiagram
    participant User as 用户
    participant GPT as ChatGPT（模型）
    participant App as GiftGenius MCP 服务器
    participant Widget as GiftGenius 小部件

    User->>GPT: "为 2000 名员工挑选礼物"
    GPT->>App: tools.call start_bulk_gift_analysis
    App-->>GPT: response { jobId: "bulk_2025_001" }

    GPT->>Widget: ToolOutput { jobId }
    Widget->>Widget: 显示进度条

    App-->>GPT: notification job.started
    App-->>GPT: notification job.progress (10%, 40%, 70%, 100%)
    App-->>GPT: notification job.completed { resultResourceId }

    GPT->>Widget: 把事件/数据传递给小部件
    Widget->>User: 更新进度并展示结果
    

实际中的时序会更复杂一些，但核心思想很简单：MCP 事件是你的后台操作与用户体验之间的“神经系统”。

10. 使用 MCP 事件的常见错误

错误 1：“事件 = 生产格式的日志”。
有时开发者会把以前写到 console.log 的内容直接通过 MCP 转发。结果事件中既没有 eventId、也没有 jobId 和规范的 timestamp，只有一些半诗意的消息“我们快好了”。这种做法让系统脆弱：难以解析、无法去重、UI 也不知道消息属于哪一个任务。最好从一开始就把事件作为正式契约来设计：清晰的方法名、稳定的字段集合、合理的 payload。

错误 2：没有幂等性和唯一的 eventId。
许多团队最初会想：“事件不就来一次么”。一周后问题来了：客户端重连导致通知重复，用户收到两份相同内容，电商后端重复发放奖励。没有唯一的 eventId 和最起码的客户端去重，你迟早会踩到严重 bug。在分布式系统中应基于“至少一次投递”的模型：重复不可避免。

错误 3：把系统事件与业务事件搅成一锅粥。
例如，在同一条流里同时灌入 logging/message、job.progress、job.completed、resources/updated，且没有清晰的 type/method 区分。最终 UI 层开始写奇怪的 if (message.includes("完成")) 来判断任务是否结束。最好明确划分：系统通知（日志、heartbeat）与业务事件（job.*、resource.*）应有严格描述的模式。

错误 4：任务状态转换不一致。
有时服务器在同一事件流中先发 job.completed，之后又来 job.progress，然后再发 job.failed。这通常是因为缺少显式状态机以及在发出事件时的检查。客户端将无从判断实际发生了什么。更正确的做法是描述一个有限状态自动机，并禁止发布违反它的事件：例如在 completed 之后最多只能发额外的信息事件，不能把任务再切回 running。

错误 5：过度绑定当前规范版本中的 MCP 方法名。
MCP 规范仍在发展中。如果你把一切都绑定到当前版本里的系统方法名，而不使用自己的命名空间，一旦协议变更你可能需要重写半个系统。更好的方式是把事件当作你在 MCP 之上的迷你规范：可以基于现有方法（notifications/progress、resources/updated），但业务事件（notifications/job/*）应设计在你自己的命名空间中，并保持相对独立。

错误 6：事件与 UX 脱节。
有时团队在后端做了漂亮的事件模型，却没有把它落实到小部件上：job.progress 只存在于日志中，而 UI 上 40 秒只有一个孤零零的转圈。用户在这种场景下既不信 MCP，也不信 AI。设计事件时始终要想着你想要的具体 UI 效果：进度条、阶段、局部结果。MCP 事件不是为了协议而存在，而是为了明确的应用行为。