🧠 图解记忆: 长任务要 call now、fetch later,用协议化 task_id 管状态、结果、取消和清理。
💡 答案要点
背景问题:长任务处理的困境
普通 MCP 工具调用是一次请求—响应;可以报告进度,但调用方通常仍要等待最终响应。现实中还有大量需要“先提交、后取结果”的 长时间运行任务:
- 药物分子分析(数小时)
- 代码迁移(分钟到小时)
- 测试套件执行(数千个测试用例)
- 深度研究(多轮搜索和推理)
传统 workaround(糟糕的解决方案):
python
# 把一个工具拆成三个工具——这是灾难!
tool `start_job`: 开始任务,返回 job_id
tool `get_status(job_id)`: 查询状态
tool `get_result(job_id)`: 获取结果问题:
- Agent 可能hallucinate job_id(亚马逊团队真实踩坑)
- Agent 可能不会正确轮询
- 每个 MCP Server 自己实现这套约定,不通用
Tasks 提案的核心解决方案:
"引入 task primitive 和 task ID,客户端可以主动查询任务状态和结果,有效期由服务器定义(可长达数天)。"
关键概念:
| 概念 | 说明 |
|---|---|
| Task | 面向长时间工作流的协议化任务抽象 |
| Task ID | 任务的唯一标识符,客户端可用它查询状态和获取结果 |
| call-now, fetch-later | 发起任务时不阻塞,后续主动拉取结果 |
Tasks vs 传统工具调用的本质区别:
| 维度 | 传统工具调用 | Tasks 原语 |
|---|---|---|
| 执行模式 | 同步等待结果 | 异步,发起后不阻塞 |
| 状态查询 | 无(只能等) | task ID 主动查询 |
| 结果获取 | 一次性 | 可延迟,多次尝试 |
| 适用场景 | 短任务(<1分钟) | 长任务(分钟到天) |
| 客户端控制 | 服务器驱动 | 客户端主导轮询 |
MCP Tasks 生命周期:
┌─────────────────────────────────────────────────────────┐
│ MCP Tasks 生命周期 │
├─────────────────────────────────────────────────────────┤
│ │
│ 1. 发起任务(返回 task_id,不阻塞) │
│ → tools/call 返回 { taskId: "task_abc123" } │
│ │
│ 2. 查询状态(客户端主动) │
│ → tasks/get { taskId: "task_abc123" } │
│ ← { status: "processing", 进度: "45%" } │
│ │
│ 3. 获取结果(任务完成后) │
│ → tasks/result { taskId: "task_abc123" } │
│ ← { status: "completed", result: {...} } │
│ │
│ 4. 删除任务(清理资源) │
│ → tasks/delete { taskId: "task_abc123" } │
└─────────────────────────────────────────────────────────┘适合评估的企业场景:
| 场景 | 挑战 | Tasks 解决方案 |
|---|---|---|
| 医疗生命科学 | 药物分析任务耗时数小时 | 并发发起 + 主动轮询状态 |
| 企业自动化平台 | SDLC 流程跨部门协调 | 后台任务 + 不阻塞 Agent |
| 代码迁移 | 分钟到小时的分析迁移 | start → poll → getResult |
| 测试执行 | 数千测试用例,小时级别 | 流式日志 + 最终结果 |
| 深度研究 | 多轮搜索推理 | 后台运行 + 完成通知 |
伪代码(不要当作当前 Python SDK 的固定接口):
python
# 发起一个长时间任务(不阻塞)
result = client.call_tool(
"analyze_drug_interactions",
args={"molecules": [...]},
timeout=3600 # 期望执行时间,但不是阻塞上限
)
# result 是 TaskResult,不是最终结果
print(result.task_id) # "task_abc123"
print(result.status) # "processing"
# 客户端主动轮询状态
import time
while result.status == "processing":
status = client.request("tasks/get", {"taskId": result.task_id})
print(f"进度: {status.progress}%")
time.sleep(30)
# 获取最终结果(任务完成后)
if result.status == "completed":
final_result = client.request("tasks/result", {"taskId": result.task_id})
print(final_result.data)
# 清理资源
client.request("tasks/delete", {"taskId": result.task_id})规范状态与面试边界:
- SEP-1686 描述了 Tasks 的状态机和
tasks/get、tasks/result、tasks/list、tasks/delete等消息; - Tasks 是否位于核心规范、独立扩展或某个 SDK 的实验命名空间会随发布版本变化,必须同时检查目标协议版本、SDK 文档和初始化 capabilities;
- 不支持 Tasks 时,仍可在业务层使用
start/status/result工具约定,但任务 ID、租户隔离、过期清理、轮询退避和幂等必须由应用治理; - 因此面试重点应是长任务生命周期与兼容性设计,而不是背一段可能尚未由所用 SDK 实现的代码。
为什么值得关注:
- 解决企业级刚需:长时间任务(分钟到天)是企业 AI 的核心场景
- 标准化工作流集成:AWS Step Functions、Google Workflows、CI/CD 都可以用 MCP 包装
- 多 Agent 并行:slow Agent 不再阻塞整个系统,其他 Agent 可以继续工作
- 客户端主导:Host Application 控制轮询,而不是让 Agent 自己管理(避免 hallucination)
面试话术:
"Tasks 解决的是 call-now、fetch-later:长任务先返回 task ID,再查询状态和结果。设计时我会重点回答状态机、幂等、租户隔离、结果保留期、取消、轮询退避和不支持扩展时的降级。由于 Tasks 的核心规范、扩展和 SDK 支持仍可能不同,落地前必须核对目标版本,不能只凭 SEP 名称假设客户端已经实现。"
