FastMCP 后台任务实战基于 SEP-2663 扩展的异步工具调用全指南【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp本指南以仓库内 examples/tasks/README.md 中可运行的客户端/服务端示例为骨架系统讲解 FastMCP 如何通过io.modelcontextprotocol/tasks扩展SEP-2663实现后台任务服务端如何用一行代码启用任务扩展、如何用taskTrue声明可后台化的工具并上报进度客户端如何以「透明等待、显式句柄、并行并发」三种方式驱动后台任务以及如何用 Redis/Docket 将 worker 扩展到独立进程。读完你将能直接运行该示例并理解其底层实现原理能力协商、tasks/get/tasks/update/tasks/cancel三个请求方法、客户端轮询回退策略与在任务中应答 elicitation 的机制。示例概览一个开箱即用的客户端/服务端任务对examples/tasks/目录包含四个文件构成一个完整的 SEP-2663 后台任务演示文件作用server.py服务端暴露一个taskTrue工具执行时每秒上报一次进度client.py客户端用三种方式驱动后台任务docker-compose.yml可选启动 Redis用于跨进程分发 workerREADME.md运行说明本文骨架示例默认运行在内存后端memory://上除了启动服务端和客户端两个进程外不需要安装任何额外组件、不需要启动 Redis。这也是理解 FastMCP 后台任务最直接的入口。运行示例三种驱动方式1. 启动服务端在 fastmcp 仓库根目录下先完成依赖同步再启动服务端uv sync # from the fastmcp root, once python examples/tasks/server.py # listens on http://127.0.0.1:8000/mcp服务端监听在http://127.0.0.1:8000/mcp以 HTTP transport 运行server.py 中mcp.run(transporthttp, host127.0.0.1, port8000)。2. 客户端三种调用方式在另一个终端中运行客户端# 透明模式 —— call_tool 在后台执行任务并直接返回最终结果 python examples/tasks/client.py --duration 8 # 显式句柄模式 —— 立即返回由你自行轮询状态最后收集结果 python examples/tasks/client.py handle --duration 6 # 并行模式 —— 同时发射多个任务观察它们重叠执行 python examples/tasks/client.py parallel python examples/tasks/client.py parallel 8 6 4 2parallel是最值得观察的运行方式四个时长递减的任务同时启动总墙钟时间趋近于最长那个任务而非各任务之和因为 worker 并发执行了它们。默认并发为[5, 4, 3, 2]也可传入任意数量的时长参数每个参数对应一个任务。服务端实现一行代码启用任务扩展服务端启用后台任务的核心只有两行server.pymcp FastMCP(Tasks Example) mcp.add_extension(TasksExtension())TasksExtension来自fastmcp_tasks包是 SEP-2663 任务扩展的服务端线缆适配器。无参数构造时它会读取FASTMCP_DOCKET_*环境变量并回退到进程内的memory://worker——这正是示例「无需安装任何东西」的原因。声明一个可后台化的工具taskTrue服务端的关键工具是slow_computationserver.pymcp.tool(taskTaskConfig(poll_intervaltimedelta(seconds1))) async def slow_computation( label: Annotated[str, A name for this run, echoed back in progress logs], duration: Annotated[int, How many seconds the computation should take (1-60)], progress: Progress Progress(), ) - str: Spend duration seconds working, reporting progress once per second. if not 1 duration 60: raise ValueError(duration must be between 1 and 60 seconds) await progress.set_total(duration) for elapsed in range(1, duration 1): await asyncio.sleep(1) await progress.increment() await progress.set_message(f{label}: {elapsed}/{duration}s) return f{label} finished in {duration}s这里有几个值得注意的细节task接收的是TaskConfig对象而非裸布尔值。TaskConfig定义在 fastmcp_slim/fastmcp/utilities/tasks.py包含mode与poll_interval两个字段mode取值forbidden/optional/required默认optional。forbidden表示不支持任务执行optional表示同步与任务两种方式都支持required表示必须作为任务执行客户端未声明能力时报-32021。布尔写法taskTrue会被from_bool转换为modeoptional。poll_interval是服务端建议的轮询间隔核心默认值为timedelta(seconds5)见 tasks.py。示例将其缩短为 1 秒让客户端能在约 1 秒内观察到任务完成5 秒是为真实负载调优的默认值。任务的工具函数必须是 async 函数。TaskConfig.validate_function会校验这一点对声明了任务执行但使用同步函数的工具直接抛出ValueErrortasks.py。progress参数通过依赖注入获得。fastmcp.dependencies.Progress提供了set_total/increment/set_message三个进度上报原语服务端每秒上报一次客户端可以在轮询时读到status_message。两个关注点的刻意分离从源码结构看FastMCP 将「工具能否任务化」与「服务端是否运行任务」做了严格分离taskTrue是每个组件上的意图声明这个工具可以作为任务运行mcp.add_extension(TasksExtension(...))是服务端级启用与配置这个服务端运行任务配置如下。add_extension是必需的不会因为存在taskTrue标志而被自动探测。设计上这样做的原因记录于 dev-docs/v4-notes/background-tasks.md是扩展的配置后端 URL、worker 并发、TTL 等需要有一个明确的存放位置同时保证能力广告的诚实性——服务端仅在扩展已注册时才对外声明tasks能力从而避免生产环境忘了配 Redis、工具静默跑在内存后端这类最危险的误用。如果使用了taskTrue却没有注册扩展会得到一个响亮的构建期错误。服务端内部扩展注册了哪些能力TasksExtensionfastmcp_tasks/fastmcp_tasks/extension.py向服务端贡献了三部分内容协商能力扩展标识符为io.modelcontextprotocol/tasks常量TASKS_EXTENSION_ID定义于 tasks.py三个附加请求方法methods()返回的MethodBinding列表extension.pytasks/get、tasks/update、tasks/cancel且仅在 2026-07-28 协议时代可用_TASK_METHOD_VERSIONS MODERN_PROTOCOL_VERSIONS扩展机制本身是分时代的旧时代下这些方法会被报告为 not found一个tools/call拦截器intercept_tool_call决定本次调用是否作为任务执行。此外还有一个lifespan启动 Docket 后端/worker并安装 worker 侧的Context钩子进度、server 解析等保证 worker 进程里的ctx可用。钩子是进程级且引用计数的一个进程内有多个服务端各跑一个 TasksExtension 时钩子直到最后一个扩展关闭才被清除避免一个服务端退出连累另一个服务端在途的 workerextension.py。任务判定逻辑required/optional/forbidden拦截器intercept_tool_callextension.py的决策流程为解析工具含版本解析避免对旧版本调用错误地任务化最高版本若工具不存在或task_config.supports_tasks()为假即forbidden直接透传给普通工具执行路径检查客户端是否在当前请求的_meta中声明了 tasks 能力仅在现代协议时代有效旧协议时代的_meta声明视为不存在mode required且客户端未声明 → 抛-32021MISSING_REQUIRED_CLIENT_CAPABILITYmode required且已声明或mode optional且已声明 → 走create_task任务化路径其余情况 → 透传普通调用。同时tasks/*三个方法在被处理前都会经过两道门禁_check_task_request一是要求客户端在请求_meta中声明了 tasks 能力否则返回-32021二是校验 HTTP 传输下的Mcp-Name路由头与请求体内的taskId一致SEP-2243 路由镜像检查不一致时返回HEADER_MISMATCH错误extension.py。客户端实现一次 import 启用全进程任务能力客户端启用任务支持的机制非常轻量——只需导入fastmcp_tasksclient.pyfrom fastmcp_tasks import call_tool_task # importing enables client task support这个 import 会触发 fastmcp_tasks/init.py 中的register_internal_client_extension_factory(_build_tasks_client_extension)将客户端任务扩展注册到进程内的每一个Client。没有这个 importClient永远不会对外广告 tasks 能力服务端就会把调用同步执行。这是有意设计的隐式但响亮激活模式导入即启用不导入则不启用没有静默降级。从源码结构看客户端任务扩展TasksClientExtensionclient.py做两件事广告 tasks 能力告诉服务端本客户端能驱动任务因此taskTrue的工具可能被任务化这是同意不是请求声明ResultClaim对resultType: task的声称结果CreateTaskResult声明解析权。解析器在底层轮询tasks/get直到完成把工具的真实结果包装回CallToolResult——调用call_tool的代码完全感知不到这次调用被任务化了。三种客户端模式的源码级解读透明模式默认命令client.py 中transparent命令直接调用client.call_tool(slow_computation, {...})。服务端将调用任务化并返回CreateTaskResult声称结果客户端的ResultClaim解析器随即在底层轮询到完成最终返回工具的字符串结果。调用方代码与普通同步工具调用别无二致——call_tool是否被任务化对调用者是透明的。显式句柄模式handle命令client.py使用call_tool_task立即获得一个ToolTask句柄然后自行轮询task await call_tool_task( client, slow_computation, {label: handle, duration: duration} ) console.print(fTask started: {task.task_id}) while True: status await task.status() if status.status in (completed, failed, cancelled): break console.print(fstill {status.status}: {status.status_message}) await asyncio.sleep(1) result await task.result()ToolTaskclient.py是快速返回逃生舱的句柄对象提供task_id服务端生成的而非客户端生成的任务 IDstatus()通过tasks/get获取当前状态wait(stateNone, timeout300.0)轮询直到到达指定状态或任一终态wait(stateinput_required)可观察任务暂停等待输入的中间态result()驱动到完成并返回解析后的结果失败/取消时默认抛ToolErrorraise_on_errorFalse可改为返回错误结果结果会缓存cancel()通过tasks/cancel请求协作式取消直接await句柄等价于result()。call_tool_taskclient.py与call_tool的差异在于它使用allow_claimedTrue调用底层call_tool因此服务端一接受任务就返回拿到ClientCreateTaskResult即构造ToolTask。如果服务端没有任务化该调用例如工具未声明taskTrue或连接不是现代协议它会抛出ToolError提示确保工具声明了taskTrue且连接为modeauto。并行模式parallel命令client.py的核心技巧是在 await 任何一个任务之前先把所有call_tool_task都发起完毕每个都立即返回然后统一asyncio.gather收集结果tasks [ await call_tool_task( client, slow_computation, {label: ftask-{i}({d}s), duration: d}, ) for i, d in enumerate(durations) ] await asyncio.gather(*(collect(task) for task in tasks))由于每个任务在 worker 中并发执行默认并发 10四个任务的总耗时接近最长单个任务时长如8 6 4 2约等于 8 秒而非 864220 秒——这正是后台任务并行执行的直观证明。客户端轮询的工程细节从源码看客户端轮询并非固定频率的简单循环而是精心调校过指数退避_next_poll_delayclient.py从 20msMIN_POLL_INTERVAL起步每轮翻倍直至上限。快速任务约 20ms 即可观察到完成长任务则收敛到服务端建议的节奏两边都不浪费上限取服务端建议与客户端设置较小语义_poll_ceiling服务端广告的pollIntervalMs是对服务端负载的刻意声明作为退避上限被尊重未广告时回退到客户端设置FASTMCP_TASKS_CLIENT_POLL_INTERVAL默认 0.5 秒上限永远不会低于 20ms防止恶意0让客户端空转整个驱动的总超时timeout_seconds是整段驱动的一个截止时间而非单请求超时与同步路径在tools/call超时后中止一致。每轮轮询与睡眠都受剩余时间约束client.py任务中需要输入时走 elicitation状态翻转为input_required时客户端通过既有的elicitation_handler应答任务中挂起的请求再用tasks/update送达然后继续轮询。若客户端没有配置 elicitation handler则抛出ToolError提示传入elicitation_handler。现代协议下只有 elicitation 受支持sampling 与 roots 已弃用client.py。客户端侧还有一个可调设置FASTMCP_TASKS_CLIENT_POLL_INTERVALTasksClientSettings.poll_intervalsettings.py仅当服务端未广告pollIntervalMs时作为退避上限生效服务端广告时严格遵循服务端间隔。分布式 worker可选把任务执行搬出服务端进程默认memory://后端把 worker 运行在服务端进程内。要把 worker 变成独立进程需要把 Docket 指向 Redis 并先启动它cd examples/tasks docker compose up -d export FASTMCP_DOCKET_URLredis://localhost:24242/0 # or: direnv allow python server.py # in one terminal python -m fastmcp_tasks.worker_cli worker server.py # extra worker(s) in othersdocker-compose.yml 启动一个映射到宿主机24242端口的redis:7-alpine容器内 6379并带健康检查。注意环境变量指向的是宿主机端口24242。后端与 worker 形态的对应关系如下BackendWorkersmemory://in-process only默认redis://…distributed across processespython -m fastmcp_tasks.worker_cli worker server.py启动一个额外的独立 worker 进程。从 worker_cli.py 的源码看该命令会通过load_and_merge_config加载服务端模块读取已注册的TasksExtension的解析后配置而非仅凭环境变量猜测——代码里显式注释指出构造函数里配置的 Redis URL 在服务端加载前环境是看不到的这是 #4603 review 修正过的坑检查后端是否为分布式若是memory://直接报错退出并提示安装 Redis/Valkey、配置FASTMCP_DOCKET_URLredis://localhost:6379/0的完整步骤内存后端只能在单进程内工作进入服务端 lifespan 常驻循环处理队列中的任务。Docket 后端配置项TasksExtension构造函数的全部可选参数extension.py都对应DocketSettings字段settings.py未传的参数回退到FASTMCP_DOCKET_*环境变量再回退到默认值参数 / 环境变量默认值含义url/FASTMCP_DOCKET_URLmemory://后端 URL。memory://单进程redis://host:port/db分布式多进程name/FASTMCP_DOCKET_NAMEfastmcp队列名。所有共享同一名称与后端 URL 的服务端/worker 共享同一任务队列worker_name/FASTMCP_DOCKET_WORKER_NAME自动生成worker 名称concurrency/FASTMCP_DOCKET_CONCURRENCY10worker 可并发处理的最大任务数redelivery_timeout/FASTMCP_DOCKET_REDELIVERY_TIMEOUT300s任务重投递超时worker 超时未完成则转投其他 workerreconnection_delay/FASTMCP_DOCKET_RECONNECTION_DELAY5sworker 与后端断连后的重连间隔minimum_check_interval/FASTMCP_DOCKET_MINIMUM_CHECK_INTERVAL50msworker 轮询新任务的频率。调低减少任务拾取延迟但增加 CPU高吞吐长任务场景可调高另外还有两类独立的环境变量前缀FASTMCP_TASKS_ENCRYPTION_KEYTasksSettings.encryption_key用于加密任务上下文快照的密钥。快照携带提交调用者的访问令牌与 HTTP 头写入 Docket 后端保留到 TTL。共享同一任务队列的所有服务端和 worker 必须设置相同的密钥worker 无法解密快照时会直接失败该任务而不是以匿名调用者身份运行。未设置时快照以明文 JSON 存储。密钥由 PBKDF2 从该值派生 Fernet 密钥任意非空字符串可用但建议至少 32 个随机字符settings.py。FASTMCP_DOCKET_*配置与核心 FastMCP 设置共用.env或FASTMCP_ENV_FILE指定的 env 文件settings.py。fastmcp-tasks的额外依赖在 fastmcp_tasks/pyproject.toml 中声明fastmcp-slim[server]、cryptography43.0.0用于FASTMCP_TASKS_ENCRYPTION_KEY的 Fernet 加密与pydocket0.24.1Docket 执行引擎。协议与架构背景SEP-2663 与 FastMCP 的实现现状本示例是 SEP-2663Tasks Extension的参考实现。该扩展基于 SEP-2133 扩展机制属于能力协商型特性。仓库设计文档 dev-docs/v4-notes/background-tasks.md 记录了完整的演进脉络核心要点如下线缆形态6 步交互流程客户端在请求_meta中广告 tasks 能力——这是同意不是请求客户端发出普通tools/call由服务端决定是否任务化若任务化服务端返回CreateTaskResultresultType: task的声称结果并携带服务端生成的taskId客户端轮询tasks/get直到终态结果内联在该响应中任务执行中的输入请求elicitation 等是轮询式的状态翻转为input_required未决请求出现在inputRequests映射中客户端通过tasks/update应答tasks/cancel是协作式的。可选推送notifications/tasks经subscriptions/listen存在但服务端无需发送。与旧协议SEP-1686的关键差异任务 ID 改为服务端生成tasks/list、tasks/delete被移除枚举风险 / 依赖 TTL结果检索从独立的tasks/result内联进tasks/get任务创建改为持久化创建 MUST任务中输入从推送中继改为轮询状态机从 7 种缩减为 5 种可增广请求仅限tools/call负载均衡路由由Mcp-Name: taskId头承担。FastMCP 的实现取舍fastmcp-tasks包执行引擎与线缆适配器分离Docket 执行引擎队列、worker、结果存储、TTL、memory:///redis://后端是线缆无关的包内新增的是薄的 SEP-2663 线缆适配器工具专属task仅限 tools 表面prompt/resource 的任务脊柱被丢弃不引领 SDK 的可增广请求类型客户端默认透明完成友好接口call_tool在底层驱动轮询并返回真实结果底层call_tool_mcp暴露原始CreateTaskResultcall_tool_task提供快速返回的ToolTask句柄扩展必须显式注册add_extension(TasksExtension(...))是启用任务的必要条件是后端配置的唯一归属地也是能力广告的诚实来源任务 ID 隔离与授权tasks/get/update/cancel通过服务端生成的、按调用者隔离的复合键实现授权——比规范中taskIds 可作为 bearer token的建议更强。该设计文档标注为 Status: Shipped即上述设计已落地为当前仓库中的fastmcp-tasks包与示例代码。小结从示例到生产的关键要点服务端mcp.add_extension(TasksExtension())一行启用工具用taskTrue或taskTaskConfig(mode..., poll_interval...)声明任务函数必须是 async且可注入Progress上报进度。客户端import fastmcp_tasks一次全进程所有Client即具备任务驱动能力call_tool透明等待、call_tool_task显式句柄、asyncio.gather并行收集。默认零依赖memory://后端进程内执行适合开发与单进程部署。生产分发docker compose up -d启动 Redis设置FASTMCP_DOCKET_URLredis://...再用python -m fastmcp_tasks.worker_cli worker server.py按需扩展独立 worker多实例共享队列时务必统一配置FASTMCP_TASKS_ENCRYPTION_KEY以加密任务上下文快照。【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
FastMCP 后台任务实战：基于 SEP-2663 扩展的异步工具调用全指南
FastMCP 后台任务实战基于 SEP-2663 扩展的异步工具调用全指南【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp本指南以仓库内 examples/tasks/README.md 中可运行的客户端/服务端示例为骨架系统讲解 FastMCP 如何通过io.modelcontextprotocol/tasks扩展SEP-2663实现后台任务服务端如何用一行代码启用任务扩展、如何用taskTrue声明可后台化的工具并上报进度客户端如何以「透明等待、显式句柄、并行并发」三种方式驱动后台任务以及如何用 Redis/Docket 将 worker 扩展到独立进程。读完你将能直接运行该示例并理解其底层实现原理能力协商、tasks/get/tasks/update/tasks/cancel三个请求方法、客户端轮询回退策略与在任务中应答 elicitation 的机制。示例概览一个开箱即用的客户端/服务端任务对examples/tasks/目录包含四个文件构成一个完整的 SEP-2663 后台任务演示文件作用server.py服务端暴露一个taskTrue工具执行时每秒上报一次进度client.py客户端用三种方式驱动后台任务docker-compose.yml可选启动 Redis用于跨进程分发 workerREADME.md运行说明本文骨架示例默认运行在内存后端memory://上除了启动服务端和客户端两个进程外不需要安装任何额外组件、不需要启动 Redis。这也是理解 FastMCP 后台任务最直接的入口。运行示例三种驱动方式1. 启动服务端在 fastmcp 仓库根目录下先完成依赖同步再启动服务端uv sync # from the fastmcp root, once python examples/tasks/server.py # listens on http://127.0.0.1:8000/mcp服务端监听在http://127.0.0.1:8000/mcp以 HTTP transport 运行server.py 中mcp.run(transporthttp, host127.0.0.1, port8000)。2. 客户端三种调用方式在另一个终端中运行客户端# 透明模式 —— call_tool 在后台执行任务并直接返回最终结果 python examples/tasks/client.py --duration 8 # 显式句柄模式 —— 立即返回由你自行轮询状态最后收集结果 python examples/tasks/client.py handle --duration 6 # 并行模式 —— 同时发射多个任务观察它们重叠执行 python examples/tasks/client.py parallel python examples/tasks/client.py parallel 8 6 4 2parallel是最值得观察的运行方式四个时长递减的任务同时启动总墙钟时间趋近于最长那个任务而非各任务之和因为 worker 并发执行了它们。默认并发为[5, 4, 3, 2]也可传入任意数量的时长参数每个参数对应一个任务。服务端实现一行代码启用任务扩展服务端启用后台任务的核心只有两行server.pymcp FastMCP(Tasks Example) mcp.add_extension(TasksExtension())TasksExtension来自fastmcp_tasks包是 SEP-2663 任务扩展的服务端线缆适配器。无参数构造时它会读取FASTMCP_DOCKET_*环境变量并回退到进程内的memory://worker——这正是示例「无需安装任何东西」的原因。声明一个可后台化的工具taskTrue服务端的关键工具是slow_computationserver.pymcp.tool(taskTaskConfig(poll_intervaltimedelta(seconds1))) async def slow_computation( label: Annotated[str, A name for this run, echoed back in progress logs], duration: Annotated[int, How many seconds the computation should take (1-60)], progress: Progress Progress(), ) - str: Spend duration seconds working, reporting progress once per second. if not 1 duration 60: raise ValueError(duration must be between 1 and 60 seconds) await progress.set_total(duration) for elapsed in range(1, duration 1): await asyncio.sleep(1) await progress.increment() await progress.set_message(f{label}: {elapsed}/{duration}s) return f{label} finished in {duration}s这里有几个值得注意的细节task接收的是TaskConfig对象而非裸布尔值。TaskConfig定义在 fastmcp_slim/fastmcp/utilities/tasks.py包含mode与poll_interval两个字段mode取值forbidden/optional/required默认optional。forbidden表示不支持任务执行optional表示同步与任务两种方式都支持required表示必须作为任务执行客户端未声明能力时报-32021。布尔写法taskTrue会被from_bool转换为modeoptional。poll_interval是服务端建议的轮询间隔核心默认值为timedelta(seconds5)见 tasks.py。示例将其缩短为 1 秒让客户端能在约 1 秒内观察到任务完成5 秒是为真实负载调优的默认值。任务的工具函数必须是 async 函数。TaskConfig.validate_function会校验这一点对声明了任务执行但使用同步函数的工具直接抛出ValueErrortasks.py。progress参数通过依赖注入获得。fastmcp.dependencies.Progress提供了set_total/increment/set_message三个进度上报原语服务端每秒上报一次客户端可以在轮询时读到status_message。两个关注点的刻意分离从源码结构看FastMCP 将「工具能否任务化」与「服务端是否运行任务」做了严格分离taskTrue是每个组件上的意图声明这个工具可以作为任务运行mcp.add_extension(TasksExtension(...))是服务端级启用与配置这个服务端运行任务配置如下。add_extension是必需的不会因为存在taskTrue标志而被自动探测。设计上这样做的原因记录于 dev-docs/v4-notes/background-tasks.md是扩展的配置后端 URL、worker 并发、TTL 等需要有一个明确的存放位置同时保证能力广告的诚实性——服务端仅在扩展已注册时才对外声明tasks能力从而避免生产环境忘了配 Redis、工具静默跑在内存后端这类最危险的误用。如果使用了taskTrue却没有注册扩展会得到一个响亮的构建期错误。服务端内部扩展注册了哪些能力TasksExtensionfastmcp_tasks/fastmcp_tasks/extension.py向服务端贡献了三部分内容协商能力扩展标识符为io.modelcontextprotocol/tasks常量TASKS_EXTENSION_ID定义于 tasks.py三个附加请求方法methods()返回的MethodBinding列表extension.pytasks/get、tasks/update、tasks/cancel且仅在 2026-07-28 协议时代可用_TASK_METHOD_VERSIONS MODERN_PROTOCOL_VERSIONS扩展机制本身是分时代的旧时代下这些方法会被报告为 not found一个tools/call拦截器intercept_tool_call决定本次调用是否作为任务执行。此外还有一个lifespan启动 Docket 后端/worker并安装 worker 侧的Context钩子进度、server 解析等保证 worker 进程里的ctx可用。钩子是进程级且引用计数的一个进程内有多个服务端各跑一个 TasksExtension 时钩子直到最后一个扩展关闭才被清除避免一个服务端退出连累另一个服务端在途的 workerextension.py。任务判定逻辑required/optional/forbidden拦截器intercept_tool_callextension.py的决策流程为解析工具含版本解析避免对旧版本调用错误地任务化最高版本若工具不存在或task_config.supports_tasks()为假即forbidden直接透传给普通工具执行路径检查客户端是否在当前请求的_meta中声明了 tasks 能力仅在现代协议时代有效旧协议时代的_meta声明视为不存在mode required且客户端未声明 → 抛-32021MISSING_REQUIRED_CLIENT_CAPABILITYmode required且已声明或mode optional且已声明 → 走create_task任务化路径其余情况 → 透传普通调用。同时tasks/*三个方法在被处理前都会经过两道门禁_check_task_request一是要求客户端在请求_meta中声明了 tasks 能力否则返回-32021二是校验 HTTP 传输下的Mcp-Name路由头与请求体内的taskId一致SEP-2243 路由镜像检查不一致时返回HEADER_MISMATCH错误extension.py。客户端实现一次 import 启用全进程任务能力客户端启用任务支持的机制非常轻量——只需导入fastmcp_tasksclient.pyfrom fastmcp_tasks import call_tool_task # importing enables client task support这个 import 会触发 fastmcp_tasks/init.py 中的register_internal_client_extension_factory(_build_tasks_client_extension)将客户端任务扩展注册到进程内的每一个Client。没有这个 importClient永远不会对外广告 tasks 能力服务端就会把调用同步执行。这是有意设计的隐式但响亮激活模式导入即启用不导入则不启用没有静默降级。从源码结构看客户端任务扩展TasksClientExtensionclient.py做两件事广告 tasks 能力告诉服务端本客户端能驱动任务因此taskTrue的工具可能被任务化这是同意不是请求声明ResultClaim对resultType: task的声称结果CreateTaskResult声明解析权。解析器在底层轮询tasks/get直到完成把工具的真实结果包装回CallToolResult——调用call_tool的代码完全感知不到这次调用被任务化了。三种客户端模式的源码级解读透明模式默认命令client.py 中transparent命令直接调用client.call_tool(slow_computation, {...})。服务端将调用任务化并返回CreateTaskResult声称结果客户端的ResultClaim解析器随即在底层轮询到完成最终返回工具的字符串结果。调用方代码与普通同步工具调用别无二致——call_tool是否被任务化对调用者是透明的。显式句柄模式handle命令client.py使用call_tool_task立即获得一个ToolTask句柄然后自行轮询task await call_tool_task( client, slow_computation, {label: handle, duration: duration} ) console.print(fTask started: {task.task_id}) while True: status await task.status() if status.status in (completed, failed, cancelled): break console.print(fstill {status.status}: {status.status_message}) await asyncio.sleep(1) result await task.result()ToolTaskclient.py是快速返回逃生舱的句柄对象提供task_id服务端生成的而非客户端生成的任务 IDstatus()通过tasks/get获取当前状态wait(stateNone, timeout300.0)轮询直到到达指定状态或任一终态wait(stateinput_required)可观察任务暂停等待输入的中间态result()驱动到完成并返回解析后的结果失败/取消时默认抛ToolErrorraise_on_errorFalse可改为返回错误结果结果会缓存cancel()通过tasks/cancel请求协作式取消直接await句柄等价于result()。call_tool_taskclient.py与call_tool的差异在于它使用allow_claimedTrue调用底层call_tool因此服务端一接受任务就返回拿到ClientCreateTaskResult即构造ToolTask。如果服务端没有任务化该调用例如工具未声明taskTrue或连接不是现代协议它会抛出ToolError提示确保工具声明了taskTrue且连接为modeauto。并行模式parallel命令client.py的核心技巧是在 await 任何一个任务之前先把所有call_tool_task都发起完毕每个都立即返回然后统一asyncio.gather收集结果tasks [ await call_tool_task( client, slow_computation, {label: ftask-{i}({d}s), duration: d}, ) for i, d in enumerate(durations) ] await asyncio.gather(*(collect(task) for task in tasks))由于每个任务在 worker 中并发执行默认并发 10四个任务的总耗时接近最长单个任务时长如8 6 4 2约等于 8 秒而非 864220 秒——这正是后台任务并行执行的直观证明。客户端轮询的工程细节从源码看客户端轮询并非固定频率的简单循环而是精心调校过指数退避_next_poll_delayclient.py从 20msMIN_POLL_INTERVAL起步每轮翻倍直至上限。快速任务约 20ms 即可观察到完成长任务则收敛到服务端建议的节奏两边都不浪费上限取服务端建议与客户端设置较小语义_poll_ceiling服务端广告的pollIntervalMs是对服务端负载的刻意声明作为退避上限被尊重未广告时回退到客户端设置FASTMCP_TASKS_CLIENT_POLL_INTERVAL默认 0.5 秒上限永远不会低于 20ms防止恶意0让客户端空转整个驱动的总超时timeout_seconds是整段驱动的一个截止时间而非单请求超时与同步路径在tools/call超时后中止一致。每轮轮询与睡眠都受剩余时间约束client.py任务中需要输入时走 elicitation状态翻转为input_required时客户端通过既有的elicitation_handler应答任务中挂起的请求再用tasks/update送达然后继续轮询。若客户端没有配置 elicitation handler则抛出ToolError提示传入elicitation_handler。现代协议下只有 elicitation 受支持sampling 与 roots 已弃用client.py。客户端侧还有一个可调设置FASTMCP_TASKS_CLIENT_POLL_INTERVALTasksClientSettings.poll_intervalsettings.py仅当服务端未广告pollIntervalMs时作为退避上限生效服务端广告时严格遵循服务端间隔。分布式 worker可选把任务执行搬出服务端进程默认memory://后端把 worker 运行在服务端进程内。要把 worker 变成独立进程需要把 Docket 指向 Redis 并先启动它cd examples/tasks docker compose up -d export FASTMCP_DOCKET_URLredis://localhost:24242/0 # or: direnv allow python server.py # in one terminal python -m fastmcp_tasks.worker_cli worker server.py # extra worker(s) in othersdocker-compose.yml 启动一个映射到宿主机24242端口的redis:7-alpine容器内 6379并带健康检查。注意环境变量指向的是宿主机端口24242。后端与 worker 形态的对应关系如下BackendWorkersmemory://in-process only默认redis://…distributed across processespython -m fastmcp_tasks.worker_cli worker server.py启动一个额外的独立 worker 进程。从 worker_cli.py 的源码看该命令会通过load_and_merge_config加载服务端模块读取已注册的TasksExtension的解析后配置而非仅凭环境变量猜测——代码里显式注释指出构造函数里配置的 Redis URL 在服务端加载前环境是看不到的这是 #4603 review 修正过的坑检查后端是否为分布式若是memory://直接报错退出并提示安装 Redis/Valkey、配置FASTMCP_DOCKET_URLredis://localhost:6379/0的完整步骤内存后端只能在单进程内工作进入服务端 lifespan 常驻循环处理队列中的任务。Docket 后端配置项TasksExtension构造函数的全部可选参数extension.py都对应DocketSettings字段settings.py未传的参数回退到FASTMCP_DOCKET_*环境变量再回退到默认值参数 / 环境变量默认值含义url/FASTMCP_DOCKET_URLmemory://后端 URL。memory://单进程redis://host:port/db分布式多进程name/FASTMCP_DOCKET_NAMEfastmcp队列名。所有共享同一名称与后端 URL 的服务端/worker 共享同一任务队列worker_name/FASTMCP_DOCKET_WORKER_NAME自动生成worker 名称concurrency/FASTMCP_DOCKET_CONCURRENCY10worker 可并发处理的最大任务数redelivery_timeout/FASTMCP_DOCKET_REDELIVERY_TIMEOUT300s任务重投递超时worker 超时未完成则转投其他 workerreconnection_delay/FASTMCP_DOCKET_RECONNECTION_DELAY5sworker 与后端断连后的重连间隔minimum_check_interval/FASTMCP_DOCKET_MINIMUM_CHECK_INTERVAL50msworker 轮询新任务的频率。调低减少任务拾取延迟但增加 CPU高吞吐长任务场景可调高另外还有两类独立的环境变量前缀FASTMCP_TASKS_ENCRYPTION_KEYTasksSettings.encryption_key用于加密任务上下文快照的密钥。快照携带提交调用者的访问令牌与 HTTP 头写入 Docket 后端保留到 TTL。共享同一任务队列的所有服务端和 worker 必须设置相同的密钥worker 无法解密快照时会直接失败该任务而不是以匿名调用者身份运行。未设置时快照以明文 JSON 存储。密钥由 PBKDF2 从该值派生 Fernet 密钥任意非空字符串可用但建议至少 32 个随机字符settings.py。FASTMCP_DOCKET_*配置与核心 FastMCP 设置共用.env或FASTMCP_ENV_FILE指定的 env 文件settings.py。fastmcp-tasks的额外依赖在 fastmcp_tasks/pyproject.toml 中声明fastmcp-slim[server]、cryptography43.0.0用于FASTMCP_TASKS_ENCRYPTION_KEY的 Fernet 加密与pydocket0.24.1Docket 执行引擎。协议与架构背景SEP-2663 与 FastMCP 的实现现状本示例是 SEP-2663Tasks Extension的参考实现。该扩展基于 SEP-2133 扩展机制属于能力协商型特性。仓库设计文档 dev-docs/v4-notes/background-tasks.md 记录了完整的演进脉络核心要点如下线缆形态6 步交互流程客户端在请求_meta中广告 tasks 能力——这是同意不是请求客户端发出普通tools/call由服务端决定是否任务化若任务化服务端返回CreateTaskResultresultType: task的声称结果并携带服务端生成的taskId客户端轮询tasks/get直到终态结果内联在该响应中任务执行中的输入请求elicitation 等是轮询式的状态翻转为input_required未决请求出现在inputRequests映射中客户端通过tasks/update应答tasks/cancel是协作式的。可选推送notifications/tasks经subscriptions/listen存在但服务端无需发送。与旧协议SEP-1686的关键差异任务 ID 改为服务端生成tasks/list、tasks/delete被移除枚举风险 / 依赖 TTL结果检索从独立的tasks/result内联进tasks/get任务创建改为持久化创建 MUST任务中输入从推送中继改为轮询状态机从 7 种缩减为 5 种可增广请求仅限tools/call负载均衡路由由Mcp-Name: taskId头承担。FastMCP 的实现取舍fastmcp-tasks包执行引擎与线缆适配器分离Docket 执行引擎队列、worker、结果存储、TTL、memory:///redis://后端是线缆无关的包内新增的是薄的 SEP-2663 线缆适配器工具专属task仅限 tools 表面prompt/resource 的任务脊柱被丢弃不引领 SDK 的可增广请求类型客户端默认透明完成友好接口call_tool在底层驱动轮询并返回真实结果底层call_tool_mcp暴露原始CreateTaskResultcall_tool_task提供快速返回的ToolTask句柄扩展必须显式注册add_extension(TasksExtension(...))是启用任务的必要条件是后端配置的唯一归属地也是能力广告的诚实来源任务 ID 隔离与授权tasks/get/update/cancel通过服务端生成的、按调用者隔离的复合键实现授权——比规范中taskIds 可作为 bearer token的建议更强。该设计文档标注为 Status: Shipped即上述设计已落地为当前仓库中的fastmcp-tasks包与示例代码。小结从示例到生产的关键要点服务端mcp.add_extension(TasksExtension())一行启用工具用taskTrue或taskTaskConfig(mode..., poll_interval...)声明任务函数必须是 async且可注入Progress上报进度。客户端import fastmcp_tasks一次全进程所有Client即具备任务驱动能力call_tool透明等待、call_tool_task显式句柄、asyncio.gather并行收集。默认零依赖memory://后端进程内执行适合开发与单进程部署。生产分发docker compose up -d启动 Redis设置FASTMCP_DOCKET_URLredis://...再用python -m fastmcp_tasks.worker_cli worker server.py按需扩展独立 worker多实例共享队列时务必统一配置FASTMCP_TASKS_ENCRYPTION_KEY以加密任务上下文快照。【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
