企业官网建设流程全解析

如何在 FastMCP 中把长时间运行的工具配置为后台任务并让客户端透明取回结果？

【免费下载链接】fastmcp🚀 The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp

MCP 中工具调用默认是阻塞的：客户端发出请求后要一直等工具返回。当某个工具要跑几十秒甚至几分钟时，这种体验很差。FastMCP 实现了 MCP 后台任务扩展（SEP-2663，io.modelcontextprotocol/tasks），让服务端立刻返回一个任务 ID，工具在后台 worker 中执行，客户端再按自己的节奏轮询、完成时取回结果。本文要完成的任务是：把一个长时间运行的工具配置为后台任务，并让客户端用普通的client.call_tool(...)透明地拿到最终结果。整个功能依赖fastmcp-tasks包，服务端和客户端进程都需要安装它，且该功能需要 FastMCP 4.0.0（对应文档页面标注的版本）。

准备：安装 tasks 扩展包

pip install "fastmcp[tasks]"

这一步服务端和客户端都要做：服务端靠它提供任务执行引擎，客户端靠它声明自己支持 tasks 能力。安装完可以用fastmcp version确认版本（文档中的示例输出显示 FastMCP version 4.0.0，见 安装文档）。

服务端：注册 TasksExtension 并标记工具

按 服务端后台任务文档，启用只需两步：给服务器注册TasksExtension，再给工具装饰器加task=True。

import asyncio from fastmcp import FastMCP from fastmcp_tasks import TasksExtension mcp = FastMCP("MyServer") mcp.add_extension(TasksExtension()) @mcp.tool(task=True) async def slow_computation(duration: int) -> str: """A long-running operation.""" for i in range(duration): await asyncio.sleep(1) return f"Completed in {duration} seconds"

有三个必须知道的行为约束：

• task=True只是声明该工具可以后台执行，真正执行它的是TasksExtension。注册了task=True工具却没注册扩展的服务器会在启动时直接报错，而不是静默降级为同步执行；
• 后台任务只支持async函数，把task=True用在同步函数上会在注册时抛出ValueError；
• 只有工具能带task=，resources、resource templates 和 prompts 都不带。

用 TaskConfig 控制执行模式和轮询间隔

task=True等价于TaskConfig(mode="optional")。需要更细的控制时用TaskConfig替代布尔值：

from datetime import timedelta from fastmcp import FastMCP from fastmcp.utilities.tasks import TaskConfig from fastmcp_tasks import TasksExtension mcp = FastMCP("MyServer") mcp.add_extension(TasksExtension()) # 短任务：建议客户端 2 秒轮询一次 @mcp.tool(task=TaskConfig(mode="optional", poll_interval=timedelta(seconds=2))) async def quick_task() -> str: return "Done quickly" # 要求必须后台执行，未 opt-in 的客户端会收到错误 @mcp.tool(task=TaskConfig(mode="required")) async def must_be_background() -> str: return "Only runs as a background task"

三种执行模式的语义（来自文档中的表格）：

poll_interval是服务器建议客户端回查频率的上限而不是精确节拍：FastMCP 客户端一开始轮询得快，随后回退到该间隔，所以快任务仍然几乎立刻被观察到完成。间隔越短客户端反馈越快，但服务端负载越大。

如果希望默认让服务器上所有工具都支持后台任务，可以给构造器传tasks=True（单个装饰器仍可用task=False覆盖）。注意：如果服务器里还有同步工具，必须显式写task=False，否则注册会报错。

后端与环境变量

TasksExtension()无参调用时读取FASTMCP_DOCKET_*环境变量，未设置时落到进程内的memory://后端，零配置开箱即用：

也可以在构造时直接传参，例如mcp.add_extension(TasksExtension(url="redis://localhost:6379/0", concurrency=20))。两种后端的取舍和横向扩展见本文"可选分支"一节。

客户端：导入 fastmcp_tasks 后透明取回结果

客户端文档 强调两点前提：

1. 客户端任务支持是 opt-in 的：在客户端进程里导入fastmcp_tasks（你用它调call_tool_task时自然会导入），进程中所有Client实例都会声明 tasks 能力；不导入的话Client永远不声明该能力，服务端就把所有调用当同步请求执行。
2. tasks 只在 modern 协议上协商：能力在2026-07-28连接上协商。客户端默认的mode="auto"会自动协商到它；mode="legacy"的客户端永远不会触发后台任务，工具对它始终同步执行。

开启之后，透明调用就是普通的call_tool：

import fastmcp_tasks # enables client task support from fastmcp import Client async with Client(server, mode="auto") as client: result = await client.call_tool("slow_computation", {"duration": 10}) print(result.data)

如果服务端把这次调用作为后台任务执行，call_tool会在底层轮询到完成，然后返回与同步调用相同形状的结果——调用代码不需要知道这次调用被任务化了。文档明确说这是大多数代码的推荐默认：无论服务端是否真的把调用任务化，这段代码都能工作。

可选：显式任务句柄

需要在任务运行期间做别的事、查看状态或取消它时，改用call_tool_task，它立即返回一个ToolTask句柄而不等待完成：

from fastmcp import Client from fastmcp_tasks import call_tool_task async with Client(server, mode="auto") as client: task = await call_tool_task(client, "slow_computation", {"duration": 10}) print(f"Task started: {task.task_id}") # 任务运行期间可以做其他工作…… result = await task.result()

注意call_tool_task要求服务端真的把这次调用作为任务执行——工具不是task=True、或服务端没注册扩展时会抛ToolError。只想要结果时用call_tool，确实需要句柄时再用它。

句柄上的常用操作：

status = await task.status() print(f"{status.status}: {status.status_message}") # status.status 取值："working"、"input_required"、"completed"、"failed" 或 "cancelled" # 轮询到终态（或指定状态）；不会替你回答任务中途提出的输入 status = await task.wait(timeout=30.0) status = await task.wait(state="input_required", timeout=30.0) # 取回结果（会顺带回答任务中途的输入请求）；await task 是等价简写 result = await task.result() # 取消（协作式：任务可能在服务端注意到请求前就跑完） await task.cancel()

默认情况下失败或被取消的任务会让result()抛ToolError；给call_tool_task传raise_on_error=False可以改为拿到错误结果。

如果任务会中途向客户端提问，给Client传一个elicitation_handler，call_tool和task.result()都会自动回答；没有 handler 时，提出输入请求的任务会抛ToolError而不是挂住。服务端工具如何发起提问（返回InputRequiredResult的 guard 模式）见 服务端文档。

用仓库示例端到端验证

仓库里有一对可直接运行的 示例服务器 和 示例客户端（运行说明见 examples/tasks/README.md），默认走memory://后端，除了两个进程外什么都不用装。

在 fastmcp 仓库根目录：

uv sync # 只需一次 python examples/tasks/server.py # 监听 http://127.0.0.1:8000/mcp

示例服务器用一行启用任务扩展（mcp = FastMCP("Tasks Example")后mcp.add_extension(TasksExtension())），暴露一个slow_computation工具：task=TaskConfig(poll_interval=timedelta(seconds=1))，按duration（1–60 秒）每秒上报一次进度，返回{label} finished in {duration}s。

再开一个终端驱动它：

# 透明模式 —— 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 2

每个模式的判断标准（均为文档和示例代码给出的行为）：

• 透明模式：打印工具返回文本（label为transparent，即形如transparent finished in 8s）和耗时elapsed …s；
• 句柄模式：先打印Task started: {task.task_id}，然后循环打印still {status.status}: {status.status_message}直到状态进入completed/failed/cancelled，最后打印工具返回文本（label为handle）；
• 并行模式：同时启动 4 个任务（默认时长 5 4 3 2，可用参数覆盖），逐个打印完成时间与+Ns偏移，最终打印总耗时和最长单任务时长——总墙钟时间跟随最长任务而不是时长之和，这是 worker 并发执行的直接证据（worker 默认并发 10）。

可选分支：Redis 后端与独立 worker 进程

默认memory://后端适合单机验证：无外部依赖、单进程即可。缺点是临时性（服务器重启后所有未完成任务丢失）、拾取延迟约 250ms、无法横向扩展。生产部署按文档改用 Redis（或 Valkey）：

mcp.add_extension(TasksExtension(url="redis://localhost:6379/0"))

Redis 后端带来：任务在服务器重启后仍然存在、个位数毫秒的拾取延迟、可以加 worker 跨进程甚至跨机器分摊负载。

示例目录自带 docker-compose.yml（redis:7-alpine，映射到宿主机24242端口），分布式跑法：

cd examples/tasks docker compose up -d export FASTMCP_DOCKET_URL=redis://localhost:24242/0 # 或：direnv allow python server.py # 一个终端 python -m fastmcp_tasks.worker_cli worker server.py # 另外的终端各跑一个额外 worker

每个额外 worker 从同一队列拉任务。worker 并发用环境变量控制：export FASTMCP_DOCKET_CONCURRENCY=20。两个边界要记住：额外 worker 只在 Redis/Valkey 后端下有效，memory://是单进程的；带任务能力的工具必须在服务器启动时就定义好，动态后加的工具不会被 worker 注册，无法后台执行。

限制与排查

• 工具没后台执行而是同步跑完了：通常是客户端进程没有import fastmcp_tasks（未声明 tasks 能力），或客户端用了mode="legacy"。此时服务端按同步模式执行该调用。
• 服务器启动报错：task=True工具存在但TasksExtension未注册；或在同步函数上用了task=True（注册期ValueError）；或开了tasks=True全局默认但存在未显式task=False的同步工具。
• call_tool_task抛ToolError：服务端没有把这次调用真正任务化（工具非task=True或扩展未注册）。
• 任务中途要输入却报错：Client没有配置elicitation_handler时，提出输入请求的任务抛ToolError而不是挂住。
• 后台任务里不能await ctx.elicit(...)：命令式 elicitation 会阻塞 worker 整个客户端往返的时间；任务里要返回InputRequiredResult（guard 模式），由客户端回答后重入，从带任务的工具里调ctx.elicit()会抛出指向该模式的错误。
• 进度上报两种方式通用：Progress依赖（set_total/increment/set_message）在同步和后台两种执行模式下都能用，同一份工具代码无需区分客户端如何调用。

更完整的参考：服务端 Background Tasks（含任务中途采集输入、FASTMCP_TASKS_ENCRYPTION_KEY对任务上下文快照加密等小节）、客户端 Tasks 和 fastmcp-tasks 包说明。

【免费下载链接】fastmcp🚀 The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp