企业官网建设流程全解析

1. 为什么你的 MCP 工具一跑长任务就“卡死”

如果你用 FastMCP 写过工具，大概率遇到过这种场景：一个工具函数里要跑数据清洗、批量文件解析或者调用外部模型做推理，耗时从几十秒到几分钟不等。客户端一发请求，整个会话就挂在那里等，界面转圈，用户以为服务挂了，其实只是你的函数还在await asyncio.sleep()。

MCP 协议里，工具、资源、提示这些组件的交互默认都是阻塞式的。客户端发请求，服务端算完才回响应。对于秒级以内的操作这没问题，但一旦进入“分钟级”区间，体验就崩了。MCP 后台任务协议（SEP-1686）就是来解决这个问题的：客户端发起操作后立刻拿到一个任务 ID，然后可以轮询进度、等结果就绪再取。

FastMCP 把这套协议封装得很薄，核心动作只有一个——在装饰器里加task=True。但真正要把它用稳，光加个布尔值不够。你需要理解TaskConfig的三种执行模式、Docket后端的选择、轮询间隔的取舍，以及怎么验证后台执行确实生效了。这篇就按“能复制、能跑通、能排错”的路线，把 TaskConfig 与 Docket 的骨架搭出来。

适合谁看：已经在用 FastMCP 写工具、准备把耗时逻辑挪到后台的开发者；或者刚接触 MCP 后台任务、想先跑通一个最小可复现示例的人。下面所有代码都可以直接贴进项目里改。

2. TaoToken 在后台任务链路里的接入位置

后台任务跑起来之后，你的工具函数里大概率要调用模型能力——比如批量摘要、分类、生成报告。这时候如果每个任务都自己去管理 API Key、切换通道、处理限流，后台任务反而变成了新的复杂度来源。

我的做法是把模型调用统一走 TaoToken 的 API 通道。它提供统一的 Key 和 API 入口，后台任务里只需要拿一个 Key，就能调用不同模型，不用在任务代码里散落多套鉴权逻辑。官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 入口是 https://taotoken.net/api 。

具体接入位置：在你的 FastMCP 工具函数内部，当任务进入“需要模型推理”那一步时，用统一的 base_url 和 Key 发起请求。这样后台任务的重试、超时、并发控制都集中在 Docket 层，模型调用层保持干净。如果你还没拿 Key，可以先到控制台创建：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，然后在 API Keys 页面生成：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。

注意：后台任务里调用外部 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", tasks=True) @mcp.tool(task=TaskConfig(mode="optional", poll_interval=timedelta(seconds=2))) async def slow_computation(duration: int) -> str: """模拟一个耗时操作，每秒推进一步。""" for i in range(duration): await asyncio.sleep(1) return f"Completed in {duration} seconds" @mcp.tool(task=TaskConfig(mode="required")) async def must_be_background() -> str: """必须以后台方式执行，客户端不带 task 参数会报错。""" await asyncio.sleep(3) return "Only runs as a background task" @mcp.tool(task=TaskConfig(mode="forbidden")) async def sync_only() -> str: """不支持后台执行，永远同步返回。""" return "Never runs as background task"

这里三个工具分别对应三种模式。optional是task=True的等价写法，客户端带 task 参数就走后台，不带就同步；required强制后台，客户端不带 task 直接报错；forbidden是默认行为，不支持后台。

3.2 TaskConfig 参数对照

轮询间隔的取舍很直接：短间隔反馈快但服务器负载高，长间隔负载低但状态更新延迟。我一般给秒级任务配 2 秒，分钟级任务配 10 到 30 秒。

3.3 Docket 后端配置

默认走内存后端memory://，零配置，但重启丢任务、不支持水平扩展。生产环境换成 Redis：

export FASTMCP_DOCKET_URL=redis://localhost:6379

如果要加 worker 做水平扩展，用 CLI：

export FASTMCP_DOCKET_CONCURRENCY=20 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(task=True) 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(f"Processing {f}") await asyncio.sleep(0.5) await progress.increment() return f"Processed {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_address="http://localhost:8000", server_name="MyServer", ) try: resp = await client.call_tool( tool_name="slow_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 秒才返回，说明后台没生效，检查装饰器是不是漏了task=True或者 mode 配成了 forbidden。

4.3 日志回读

服务端启动时加上日志级别，能看到 worker 取任务的记录：

FASTMCP_LOG_LEVEL=DEBUG fastmcp run server.py

日志里会出现 worker 从队列取任务、执行、写回结果的条目。如果用的是 Redis 后端，还可以直接查队列长度确认任务有没有被消费。

5. 本篇常见错排查

报错一：ValueError: task=True requires an async function

后台任务必须用异步函数。把def改成async def，同步函数加task=True会在注册时直接抛错。

报错二：客户端带 task 参数调用 required 工具却报“task required”

检查客户端是不是漏传了task={"enabled": True}。mode="required"的工具，客户端不带 task 参数会直接返回错误，这是设计行为。

报错三：内存后端下加了 worker 但任务没被分担

内存后端只支持单进程，额外 worker 不生效。换FASTMCP_DOCKET_URL=redis://localhost:6379再试。

报错四：服务器重启后未完成任务全丢了

这是内存后端的特性，任务不持久化。生产环境必须换 Redis/Valkey。

报错五：进度一直不更新

检查Progress是不是作为带默认值的参数注入的，写成progress: Progress = Progress()，不要手动实例化传进去。

报错六：tasks=True全局开启后同步工具报错

全局开启后，同步工具需要显式设task=False来覆盖，否则注册时报错。

6. 把模型调用接进后台任务

后台任务骨架跑通后，下一步就是把实际的模型调用塞进工具函数。我的建议是：任务调度、重试、超时交给 Docket，模型调用统一走 TaoToken 的 API 通道。这样你的工具函数里只需要关心业务逻辑，鉴权和通道切换不散落在任务代码里。

如果你要长期跑编码类或 Agent 类任务，可以看 Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。想先验证模型对话效果，用模型对话入口：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，ClaudeCode 相关配置参考 https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode&utm_campaign=rewrite 。

一个实用技巧：在后台任务里调用模型时，把 Docket 的重试和模型调用的超时分开配。Docket 负责“任务级重试”，模型调用层负责“单次请求超时”。两者混在一起，排查问题时很难定位是任务调度挂了还是 API 请求卡了。