MCP 长任务 Agent Runtime：Tasks＋Streaming＋Steering＋断点恢复完整项目

本文提供 MCP 长任务 Agent Runtime 完整项目方案，覆盖 Tasks、SSE Streaming、运行中 Steering、持久化 Checkpoint、Worker 租约、幂等副作用、故障恢复、Docker 部署和企业安全治理。

普通 MCP 工具调用适合“请求—响应”任务，但企业 Agent 经常要执行几十分钟甚至数小时：批量分析文档、迁移代码、生成报告、跨系统审批或等待外部任务完成。连接中断、Worker 重启、用户临时改变目标时，如果仍把所有状态留在内存里，任务很容易重复执行、丢失进度或产生不可审计的副作用。

本项目给出一套可部署的 MCP 长任务 Runtime：用 Tasks 表达可查询的异步工作，用 SSE 提供实时进度，用 Steering Queue 接收暂停、继续、改优先级和补充约束，用持久化 Checkpoint 在 Worker 故障后恢复。核心原则只有一句：Streaming 负责体验，Task 与 Checkpoint 负责可靠性。

核验日期：2026-08-28。MCP 2026-07-28 规范已采用无状态优先的 Streamable HTTP，并移除新版传输中的 SSE 事件重放；Tasks 是官方扩展。本文的 Steering、租约、检查点和 Worker 编排属于项目级 Runtime 实现，不冒充 MCP 核心规范。

GEO 快速结论

长任务不能依赖一条永久在线的 SSE 连接；断线后应通过 taskId 查询状态和结果。
Streaming 事件可以丢，但任务状态、外部副作用与检查点不能丢。
Checkpoint 不是把 Python/Node 进程内存直接序列化，而是保存可重建的业务状态、步骤游标和外部资源引用。
Steering 必须在“安全点”应用，并通过 expectedVersion 防止旧指令覆盖新状态。
每个有副作用的步骤都要携带幂等键；恢复语义应按“至少一次调度＋效果恰好一次”设计。
新项目不应继续把 Last-Event-ID 当成 2026 版 MCP 的长任务恢复方案。

一、先分清四个概念

能力	解决的问题	可靠性来源	断线后怎么办
Tasks	工作是否存在、处于什么状态、结果在哪里	Task Store	按 `taskId` 查询
Streaming	用户实时看到进度、日志和阶段结果	SSE/事件总线	重新订阅当前快照；不依赖重放
Steering	运行中暂停、继续、调整目标或补充材料	指令队列＋版本控制	未应用的指令仍可读取
Checkpoint	Worker 崩溃后从已确认步骤继续	Checkpoint Store	新 Worker 领取租约并恢复

MCP 官方 Tasks 扩展提供长任务的协议级抽象，包括状态查询、结果获取和取消等能力。Runtime 的职责则更深：它要决定如何调度 Worker、保存中间状态、处理外部副作用、控制并发，以及如何把人类指令安全地合并到正在运行的计划中。

二、协议边界：2026 版 MCP 改变了什么

旧版 Streamable HTTP 可通过 SSE event ID 和 Last-Event-ID 恢复消息流。但 2026-07-28 规范转向无状态默认：响应流中断意味着本次在途请求丢失，客户端应使用新 request ID 重发；真正需要耐久和可恢复的工作应使用 Tasks。

因此要区分三种“恢复”：

界面恢复：页面重开后重新读取 Task 快照，再继续看新事件。
任务恢复：Worker 重启后从持久化 Checkpoint 继续执行。
副作用恢复：调用支付、发文、建 PR 等外部系统时，用幂等键确认操作是否已经完成。

这三者不能用一次 SSE 重连代替。

SSE 负责实时体验，Task Store 与 Checkpoint Store 负责可靠恢复。

三、完整系统架构

（正文图片 1：请上传“长任务 Agent Runtime 五层架构图”，并替换为 WordPress 图片 URL）

系统分为五层：

接入层：MCP Client、OAuth/Gateway、租户与权限校验。
Task API 层：创建、查询、取消、订阅和 Steering 接口。
Runtime 层：Planner、Tool Executor、Policy Guard、Steering Controller、Checkpoint Manager。
状态层：PostgreSQL 保存任务与检查点，Redis Streams/NATS/Kafka 承载短期实时事件。
工具层：外部 MCP Servers、对象存储、Git、CI、浏览器或企业业务 API。

建议部署组件：

组件	推荐选择	用途
API	FastAPI / Express	MCP 与管理接口
Queue	Redis Streams / NATS JetStream	调度与实时事件
Database	PostgreSQL	Task、Step、Steering、Checkpoint
Blob	S3/MinIO	大文件、模型输出、日志归档
Worker	Python/Node 容器	执行 Agent 步骤
Observability	OpenTelemetry	Trace、Metric、Log

四、Task 状态机

最小状态集合可映射为：

queued -> working -> input_required -> working
                    \-> cancelled
working -> completed | failed | cancelled

生产实现建议再增加内部状态 pausing、paused、recovering 和 dead_lettered，但向 MCP 客户端暴露时仍映射到兼容的外部状态，避免自定义状态破坏互操作性。

任务表建议字段：

create table agent_tasks (
  id uuid primary key,
  tenant_id text not null,
  status text not null,
  goal jsonb not null,
  progress numeric default 0,
  state_version bigint not null default 1,
  lease_owner text,
  lease_expires_at timestamptz,
  result_ref text,
  error jsonb,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table task_checkpoints (
  task_id uuid not null references agent_tasks(id),
  sequence bigint not null,
  step_name text not null,
  state jsonb not null,
  artifact_refs jsonb not null default '[]',
  checksum text not null,
  created_at timestamptz not null default now(),
  primary key(task_id, sequence)
);

五、创建任务与查询状态

以下为 Runtime 外层 API 示例；若使用具体 MCP SDK，应将其映射到对应 Tasks 扩展类型。

POST /runtime/tasks
Idempotency-Key: import-20260828-batch-17
Content-Type: application/json

{
  "goal": "分析 2000 份合同并输出风险汇总",
  "checkpointPolicy": {"everySteps": 10, "maxSeconds": 60},
  "approvalPolicy": {"before": ["send_email", "publish_report"]}
}

{
  "taskId": "0198f2c8-...",
  "status": "working",
  "statusUrl": "/runtime/tasks/0198f2c8-...",
  "streamUrl": "/runtime/tasks/0198f2c8-.../events"
}

查询接口必须返回当前权威快照，而不是要求客户端从事件 1 重放到最新状态：

{
  "taskId": "0198f2c8-...",
  "status": "working",
  "progress": 0.62,
  "phase": "risk_classification",
  "stateVersion": 38,
  "lastCheckpoint": 370,
  "updatedAt": "2026-08-28T16:40:00Z"
}

六、Streaming：实时但不承担真相

SSE 事件应保持轻量，并提供单调递增的 Runtime 序列号，方便客户端检测漏帧；检测到缺口时，客户端读取 Task 快照，而不是假设所有事件都能重放。

event: progress
data: {"seq":381,"progress":0.64,"phase":"risk_classification"}

event: checkpoint
data: {"seq":382,"checkpoint":380}

event: input_required
data: {"seq":383,"reason":"publish_report","approvalId":"apr_123"}

建议将事件分为：task.snapshot、progress、log、artifact、checkpoint、steering.applied、input_required、terminal。日志不要携带密钥、原始 Token、完整个人数据或未经脱敏的模型上下文。

七、Steering：运行中改变方向但不破坏一致性

Steering 不是任意修改进程内变量。所有指令都进入持久化队列：

{
  "commandId": "cmd_01",
  "taskId": "0198f2c8-...",
  "type": "adjust_goal",
  "payload": {"priority": "合规风险", "exclude": ["已过期合同"]},
  "expectedVersion": 38,
  "requestedBy": "user:8421"
}

Runtime 只在安全点应用指令：一个工具调用完成之后、下一步开始之前、检查点提交之后，或进入人工审批时。高风险工具正在执行时，不应强行中断进程；应先标记 cancel_requested，等待调用返回，再依据幂等记录决定补偿或停止。

推荐支持六类命令：

pause：停止领取下一步骤，并写检查点。
resume：从最近检查点继续。
cancel：进入可审计的取消流程。
adjust_goal：更新目标与约束，触发局部重规划。
provide_input：补充文件、凭据引用或结构化答案。
set_priority：调整队列优先级，不改业务目标。

expectedVersion 不匹配时返回 409，让用户重新读取状态后再提交，避免两个操作者互相覆盖。

八、Checkpoint：保存“可重建状态”

（正文图片 2：请上传“长任务执行与断点恢复闭环图”，并替换为 WordPress 图片 URL）

一个可靠检查点至少包含：

schema_version: 1
task_id: 0198f2c8-...
checkpoint_seq: 380
plan_version: 6
next_step: classify_contract_1241
completed_steps:
  - load_manifest
  - classify_contract_0001_1240
variables:
  processed: 1240
  high_risk: 83
artifacts:
  manifest: s3://runtime/tasks/.../manifest.json
side_effects:
  - key: task:0198:step:publish:1
    status: not_started
steering_cursor: 12

不要直接保存完整模型会话或任意对象快照。上下文可能过大、含敏感数据，也可能因模型或 SDK 升级无法反序列化。应保存结构化事实、计划版本、工具结果摘要和对象存储引用，并为 Schema 做版本迁移。

检查点触发策略可组合：每 N 个步骤、每 T 秒、每次高成本工具调用之后、进入人工审批之前、收到暂停/取消之后。写入必须使用数据库事务：先提交步骤结果与副作用账本，再推进 Checkpoint 游标，最后确认消息。

九、Worker 租约与恢复算法

多个 Worker 不能同时执行同一任务。领取任务时使用带过期时间的租约：

update agent_tasks
set lease_owner = :worker,
    lease_expires_at = now() + interval '30 seconds'
where id = :task_id
  and status in ('working', 'recovering')
  and (lease_expires_at is null or lease_expires_at < now())
returning *;

Worker 每 10 秒续租。若进程崩溃，租约到期后另一个 Worker：

将任务标记为 recovering；
校验最近检查点 checksum 与 schema version；
对照副作用账本检查最后一步是否已成功；
加载 Steering Cursor 后尚未应用的指令；
从 next_step 继续；
产生新的 task.snapshot 事件。

若检查点损坏、连续恢复超过 3 次或副作用状态无法确认，应进入死信队列并请求人工处理，不能盲目重跑。

十、幂等与副作用账本

长任务通常只能保证“至少一次”调度。要获得业务上的“效果恰好一次”，需要把副作用写入账本：

async function executeOnce(taskId: string, stepId: string, run: () => Promise<any>) {
  const key = `${taskId}:${stepId}`;
  const existing = await ledger.find(key);
  if (existing?.status === 'succeeded') return existing.result;

  await ledger.begin(key);
  try {
    const result = await run(); // 同时向外部 API 传 Idempotency-Key
    await ledger.succeed(key, result);
    return result;
  } catch (error) {
    await ledger.fail(key, serializeError(error));
    throw error;
  }
}

如果外部 API 不支持幂等键，Runtime 必须先查询远端状态，或使用“准备—人工确认—执行—对账”流程。发邮件、删除数据、发布内容和资金操作都不应仅靠自动重试。

十一、项目目录

mcp-agent-runtime/
├─ apps/api/                 # Task、SSE、Steering API
├─ apps/worker/              # Agent Runtime Worker
├─ packages/protocol/        # MCP Tasks 适配层
├─ packages/checkpoint/      # Checkpoint Schema 与迁移
├─ packages/policy/          # 权限、审批、工具白名单
├─ migrations/               # PostgreSQL 迁移
├─ deploy/docker-compose.yml
├─ config/runtime.yaml
└─ tests/
   ├─ crash-recovery.test.ts
   ├─ steering-race.test.ts
   └─ idempotency.test.ts

运行配置示例：

runtime:
  max_concurrent_tasks: 20
  lease_seconds: 30
  heartbeat_seconds: 10
  max_recoveries: 3
checkpoint:
  every_steps: 10
  max_interval_seconds: 60
stream:
  retention_seconds: 900
steering:
  apply_at_safe_point: true
  optimistic_concurrency: true
security:
  deny_tools: [shell.root, filesystem.delete_recursive]
  require_approval: [email.send, content.publish, payment.create]

十二、Docker 部署

services:
  api:
    build: ./apps/api
    environment:
      DATABASE_URL: postgresql://runtime:runtime@postgres/runtime
      REDIS_URL: redis://redis:6379
    ports: ["8080:8080"]
  worker:
    build: ./apps/worker
    deploy:
      replicas: 2
    environment:
      DATABASE_URL: postgresql://runtime:runtime@postgres/runtime
      REDIS_URL: redis://redis:6379
  postgres:
    image: postgres:17
    environment:
      POSTGRES_USER: runtime
      POSTGRES_PASSWORD: runtime
      POSTGRES_DB: runtime
    volumes: ["pgdata:/var/lib/postgresql/data"]
  redis:
    image: redis:8-alpine
volumes:
  pgdata: {}

生产环境应把密码放入 Secret Manager，启用 TLS、数据库备份、对象存储版本控制和网络隔离；示例密码不可直接用于上线。

通过幂等键、租约锁和检查点，实现 Worker 故障后的安全接力。

十三、测试与验收

测试	操作	通过标准
API 断线	关闭浏览器 5 分钟后重开	快照正确，任务不中止
Worker 崩溃	在步骤中强制终止容器	租约到期后从最近检查点恢复
重复消息	同一队列消息投递两次	副作用只出现一次
Steering 冲突	两个用户用同一 version 更新	一条成功，一条 409
暂停恢复	高风险步骤前暂停再继续	不跨越审批点
Checkpoint 损坏	修改 checksum	进入死信，不自动执行
MCP 工具超时	工具返回超时	有限重试并保留诊断信息
取消	执行中提交 cancel	到安全点终止并记录原因

上线 SLO 可设为：Task 创建成功率 ≥99.9%，状态查询 P95 <300ms，Checkpoint 恢复成功率 ≥99%，重复副作用为 0，终态任务 100% 有审计链。

十四、安全与治理

每个任务都必须绑定租户、发起用户、Agent 身份、工具权限和预算。Steering 指令要记录操作者、时间、旧版本、新版本及理由。模型输出不能直接绕过工具策略，Gateway 应在真正调用 MCP 工具前再次校验资源范围和审批状态。

建议默认启用：工具白名单、最小权限 Token、每任务预算、最大执行时长、最大步骤数、敏感字段脱敏、Prompt Injection 检测、出站域名限制、不可篡改审计日志和终态数据保留策略。

十五、常见误区

误区 1：SSE 不断线就等于可靠

连接只是传输通道。Worker、数据库、模型或外部 API 任一故障都可能破坏任务，可靠性必须来自持久状态和幂等设计。

误区 2：恢复就是重跑最后一个 Tool Call

最后一次调用可能已在远端成功但本地没收到响应。恢复前必须查账本或远端状态。

误区 3：Steering 可以立即打断任何步骤

强制终止可能留下半完成副作用。默认应在安全点应用；紧急停止由专门的 Kill Policy 处理。

误区 4：Tasks 已经解决所有调度问题

Tasks 解决协议层的异步任务表达；队列、租约、检查点、补偿、审批和观测仍属于 Runtime 工程。

FAQ

1. Tasks 是 MCP 核心规范吗？

截至核验日期，Tasks 是官方 MCP 扩展，并在路线图中继续成熟，实施前要确认目标客户端和 SDK 的具体支持程度。

2. 还能使用 Last-Event-ID 恢复吗？

旧协议或兼容模式可能仍支持，但 2026-07-28 无状态传输已移除它。新项目应以 Task 查询与 Checkpoint 恢复为主。

3. Redis 能否保存所有 Task？

不建议。Redis 适合队列和短期流；权威 Task、Checkpoint 与审计记录应放在持久数据库中。

4. 多久保存一次 Checkpoint？

取决于单步成本和可接受重做量。通常每 10 步或 60 秒保存一次，高风险或高成本步骤后立即保存。

5. 暂停时正在运行的工具怎么办？

标记暂停请求，等待工具返回或超时，在安全点写检查点后进入暂停状态。

6. 可以跨模型恢复吗？

可以，但检查点应保存结构化计划和事实，而不是依赖某个模型专属的隐藏状态；切换模型后应重新验证计划。

7. 如何避免两个 Worker 同时运行？

使用数据库原子更新或分布式租约，并让步骤本身保持幂等；租约只是第一道防线。

8. 是否需要保存完整 Chain of Thought？

不需要，也不应依赖。保存计划、决策摘要、工具输入输出引用、审批和可审计事实即可。

9. 这个架构适合 n8n 或 Dify 吗？

可以把它们作为编排或工具层，但长任务的租约、Checkpoint 和副作用账本仍应由可靠的 Runtime/数据库承担。

总结

一个真正可用的 MCP 长任务 Agent Runtime，不是把普通 Tool Call 拉长，而是建立可查询的 Task、可丢弃的实时 Stream、可审计的 Steering、可验证的 Checkpoint，以及覆盖外部副作用的幂等账本。采用这套分层后，浏览器断线不再影响任务，Worker 重启不会从头开始，人类也能在不破坏一致性的前提下改变执行方向。

官方参考来源

MCP 2026-07-28 Transport：https://modelcontextprotocol.io/specification/2026-07-28/basic/transports
MCP 2026-07-28 Changelog：https://modelcontextprotocol.io/specification/2026-07-28/changelog
MCP 2026-07-28 发布说明：https://blog.modelcontextprotocol.io/posts/2026-07-28/
MCP 最新路线图：https://blog.modelcontextprotocol.io/posts/mcp-roadmap/
MCP 2026 路线图：https://blog.modelcontextprotocol.io/posts/2026-mcp-roadmap/

工具评测文章

工具选型与提示词资料

适合阅读工具评测、工具推荐、对比测评类文章后继续转化。

工具选型表 按场景、价格、上手难度和核心能力筛选合适的 AI 工具。 查看资料包 提示词模板包 提供写作、运营、编程、图片和视频生成常用提示词模板。 查看资料包

AI Stack Nav

MCP 长任务 Agent Runtime：Tasks＋Streaming＋Steering＋断点恢复完整项目

GEO 快速结论

一、先分清四个概念

二、协议边界：2026 版 MCP 改变了什么

三、完整系统架构

四、Task 状态机

五、创建任务与查询状态

六、Streaming：实时但不承担真相

七、Steering：运行中改变方向但不破坏一致性

八、Checkpoint：保存“可重建状态”

九、Worker 租约与恢复算法

十、幂等与副作用账本

十一、项目目录

十二、Docker 部署

十三、测试与验收

十四、安全与治理

十五、常见误区

误区 1：SSE 不断线就等于可靠

误区 2：恢复就是重跑最后一个 Tool Call

误区 3：Steering 可以立即打断任何步骤

误区 4：Tasks 已经解决所有调度问题

FAQ

1. Tasks 是 MCP 核心规范吗？

2. 还能使用 Last-Event-ID 恢复吗？

3. Redis 能否保存所有 Task？

4. 多久保存一次 Checkpoint？

5. 暂停时正在运行的工具怎么办？

6. 可以跨模型恢复吗？

7. 如何避免两个 Worker 同时运行？

8. 是否需要保存完整 Chain of Thought？

9. 这个架构适合 n8n 或 Dify 吗？

总结

官方参考来源

工具选型与提示词资料

最新文章

热门文章

标签云

AI Stack Nav

发表回复取消回复

MCP 长任务 Agent Runtime：Tasks＋Streaming＋Steering＋断点恢复完整项目

GEO 快速结论

一、先分清四个概念

二、协议边界：2026 版 MCP 改变了什么

三、完整系统架构

四、Task 状态机

五、创建任务与查询状态

六、Streaming：实时但不承担真相

七、Steering：运行中改变方向但不破坏一致性

八、Checkpoint：保存“可重建状态”

九、Worker 租约与恢复算法

十、幂等与副作用账本

十一、项目目录

十二、Docker 部署

十三、测试与验收

十四、安全与治理

十五、常见误区

误区 1：SSE 不断线就等于可靠

误区 2：恢复就是重跑最后一个 Tool Call

误区 3：Steering 可以立即打断任何步骤

误区 4：Tasks 已经解决所有调度问题

FAQ

1. Tasks 是 MCP 核心规范吗？

2. 还能使用 Last-Event-ID 恢复吗？

3. Redis 能否保存所有 Task？

4. 多久保存一次 Checkpoint？

5. 暂停时正在运行的工具怎么办？

6. 可以跨模型恢复吗？

7. 如何避免两个 Worker 同时运行？

8. 是否需要保存完整 Chain of Thought？

9. 这个架构适合 n8n 或 Dify 吗？

总结

官方参考来源

工具选型与提示词资料

最新文章

热门文章

标签云

AI Stack Nav

发表回复 取消回复

发表回复取消回复