FastMCP 高级特性之Background Tasks：用 TaskConfig 与 Docket 搭建可复现的后台任务骨架

1. 为什么你的 MCP 工具一跑长任务就“卡死”如果你用 FastMCP 写过工具大概率遇到过这种场景一个工具函数里要跑数据清洗、批量文件解析或者调用外部模型做推理耗时从几十秒到几分钟不等。客户端一发请求整个会话就挂在那里等界面转圈用户以为服务挂了其实只是你的函数还在await asyncio.sleep()。MCP 协议里工具、资源、提示这些组件的交互默认都是阻塞式的。客户端发请求服务端算完才回响应。对于秒级以内的操作这没问题但一旦进入“分钟级”区间体验就崩了。MCP 后台任务协议SEP-1686就是来解决这个问题的客户端发起操作后立刻拿到一个任务 ID然后可以轮询进度、等结果就绪再取。FastMCP 把这套协议封装得很薄核心动作只有一个——在装饰器里加taskTrue。但真正要把它用稳光加个布尔值不够。你需要理解TaskConfig的三种执行模式、Docket后端的选择、轮询间隔的取舍以及怎么验证后台执行确实生效了。这篇就按“能复制、能跑通、能排错”的路线把 TaskConfig 与 Docket 的骨架搭出来。适合谁看已经在用 FastMCP 写工具、准备把耗时逻辑挪到后台的开发者或者刚接触 MCP 后台任务、想先跑通一个最小可复现示例的人。下面所有代码都可以直接贴进项目里改。2. TaoToken 在后台任务链路里的接入位置后台任务跑起来之后你的工具函数里大概率要调用模型能力——比如批量摘要、分类、生成报告。这时候如果每个任务都自己去管理 API Key、切换通道、处理限流后台任务反而变成了新的复杂度来源。我的做法是把模型调用统一走 TaoToken 的 API 通道。它提供统一的 Key 和 API 入口后台任务里只需要拿一个 Key就能调用不同模型不用在任务代码里散落多套鉴权逻辑。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。具体接入位置在你的 FastMCP 工具函数内部当任务进入“需要模型推理”那一步时用统一的 base_url 和 Key 发起请求。这样后台任务的重试、超时、并发控制都集中在 Docket 层模型调用层保持干净。如果你还没拿 Key可以先到控制台创建https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 然后在 API Keys 页面生成https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。注意后台任务里调用外部 API 时务必设置合理的超时和重试。Docket 本身支持重试策略但模型调用层的超时要单独配否则一个卡住的请求会占住 worker 槽位。3. 可复制的 TaskConfig 配置与 Docket 接入骨架3.1 最小可跑的服务端先装依赖FastMCP 的任务系统由 Docket 提供支持Docket 最初由 Prefect 开发用于支撑每天数百万并发任务的调度服务现在已经开源。安装时直接装 fastmcp 即可Docket 会作为依赖进来。pip install fastmcp服务端代码我把它拆成“工具定义”和“任务配置”两部分方便你对照改import asyncio from datetime import timedelta from fastmcp import FastMCP from fastmcp.server.tasks import TaskConfig mcp FastMCP(MyServer, tasksTrue) mcp.tool(taskTaskConfig(modeoptional, poll_intervaltimedelta(seconds2))) async def slow_computation(duration: int) - str: 模拟一个耗时操作每秒推进一步。 for i in range(duration): await asyncio.sleep(1) return fCompleted in {duration} seconds mcp.tool(taskTaskConfig(moderequired)) async def must_be_background() - str: 必须以后台方式执行客户端不带 task 参数会报错。 await asyncio.sleep(3) return Only runs as a background task mcp.tool(taskTaskConfig(modeforbidden)) async def sync_only() - str: 不支持后台执行永远同步返回。 return Never runs as background task这里三个工具分别对应三种模式。optional是taskTrue的等价写法客户端带 task 参数就走后台不带就同步required强制后台客户端不带 task 直接报错forbidden是默认行为不支持后台。3.2 TaskConfig 参数对照参数作用常用值mode执行模式optional / required / forbiddenpoll_interval建议客户端轮询间隔timedelta(seconds2) 到 30taskTrue布尔快捷方式等价于 modeoptionaltaskFalse布尔快捷方式等价于 modeforbidden轮询间隔的取舍很直接短间隔反馈快但服务器负载高长间隔负载低但状态更新延迟。我一般给秒级任务配 2 秒分钟级任务配 10 到 30 秒。3.3 Docket 后端配置默认走内存后端memory://零配置但重启丢任务、不支持水平扩展。生产环境换成 Redisexport FASTMCP_DOCKET_URLredis://localhost:6379如果要加 worker 做水平扩展用 CLIexport FASTMCP_DOCKET_CONCURRENCY20 fastmcp tasks worker server.py每个额外 worker 从同一个队列取任务。注意额外 worker 只在 Redis/Valkey 后端下有效内存后端只能单进程。3.4 进度上报与 Docket 依赖注入后台任务最怕“黑盒”用户不知道跑到哪了。FastMCP 提供Progress依赖注入后可以上报进度from fastmcp import FastMCP from fastmcp.dependencies import Progress, CurrentDocket, CurrentWorker from docket import Docket, Worker mcp FastMCP(MyServer) mcp.tool(taskTrue) async def process_files( files: list[str], progress: Progress Progress(), docket: Docket CurrentDocket(), worker: Worker CurrentWorker(), ) - str: await progress.set_total(len(files)) for f in files: await progress.set_message(fProcessing {f}) await asyncio.sleep(0.5) await progress.increment() return fProcessed {len(files)} files on {worker.name}CurrentDocket()让你能在任务里再调度其他后台任务把工作串联起来CurrentWorker()拿到 worker 元信息。进度 API 就三个set_total、increment、set_message即时执行和后台执行下都能用。4. 验证后台执行是否生效一次触发 日志回读4.1 客户端触发服务端起在 8000 端口后用客户端触发一次后台任务import asyncio from fastmcp import FastMCPClient async def main(): client FastMCPClient( server_addresshttp://localhost:8000, server_nameMyServer, ) try: resp await client.call_tool( tool_nameslow_computation, arguments{duration: 5}, task{enabled: True}, ) task_id resp.task_id print(f后台任务已启动任务 ID: {task_id}) while True: status await client.get_task_status(task_id) print(f当前状态: {status.status}) if status.status completed: print(f结果: {status.result}) break elif status.status failed: print(f失败: {status.error}) break await asyncio.sleep(1) finally: await client.close() if __name__ __main__: asyncio.run(main())4.2 成功结果长什么样跑通后你会看到类似输出后台任务已启动任务 ID: task_abc123 当前状态: running 当前状态: running 当前状态: completed 结果: Completed in 5 seconds关键验证点有两个一是call_tool立刻返回了 task_id没有等 5 秒二是轮询过程中状态从 running 变到 completed。如果call_tool卡了 5 秒才返回说明后台没生效检查装饰器是不是漏了taskTrue或者 mode 配成了 forbidden。4.3 日志回读服务端启动时加上日志级别能看到 worker 取任务的记录FASTMCP_LOG_LEVELDEBUG fastmcp run server.py日志里会出现 worker 从队列取任务、执行、写回结果的条目。如果用的是 Redis 后端还可以直接查队列长度确认任务有没有被消费。5. 本篇常见错排查报错一ValueError: taskTrue requires an async function后台任务必须用异步函数。把def改成async def同步函数加taskTrue会在注册时直接抛错。报错二客户端带 task 参数调用 required 工具却报“task required”检查客户端是不是漏传了task{enabled: True}。moderequired的工具客户端不带 task 参数会直接返回错误这是设计行为。报错三内存后端下加了 worker 但任务没被分担内存后端只支持单进程额外 worker 不生效。换FASTMCP_DOCKET_URLredis://localhost:6379再试。报错四服务器重启后未完成任务全丢了这是内存后端的特性任务不持久化。生产环境必须换 Redis/Valkey。报错五进度一直不更新检查Progress是不是作为带默认值的参数注入的写成progress: Progress Progress()不要手动实例化传进去。报错六tasksTrue全局开启后同步工具报错全局开启后同步工具需要显式设taskFalse来覆盖否则注册时报错。6. 把模型调用接进后台任务后台任务骨架跑通后下一步就是把实际的模型调用塞进工具函数。我的建议是任务调度、重试、超时交给 Docket模型调用统一走 TaoToken 的 API 通道。这样你的工具函数里只需要关心业务逻辑鉴权和通道切换不散落在任务代码里。如果你要长期跑编码类或 Agent 类任务可以看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。想先验证模型对话效果用模型对话入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite ClaudeCode 相关配置参考 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite 。一个实用技巧在后台任务里调用模型时把 Docket 的重试和模型调用的超时分开配。Docket 负责“任务级重试”模型调用层负责“单次请求超时”。两者混在一起排查问题时很难定位是任务调度挂了还是 API 请求卡了。