如何在 FastMCP 中把长时间运行的工具配置为后台任务并让客户端透明取回结果【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcpMCP 中工具调用默认是阻塞的客户端发出请求后要一直等工具返回。当某个工具要跑几十秒甚至几分钟时这种体验很差。FastMCP 实现了 MCP 后台任务扩展SEP-2663io.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再给工具装饰器加taskTrue。import asyncio from fastmcp import FastMCP from fastmcp_tasks import TasksExtension mcp FastMCP(MyServer) mcp.add_extension(TasksExtension()) mcp.tool(taskTrue) async def slow_computation(duration: int) - str: A long-running operation. for i in range(duration): await asyncio.sleep(1) return fCompleted in {duration} seconds有三个必须知道的行为约束taskTrue只是声明该工具可以后台执行真正执行它的是TasksExtension。注册了taskTrue工具却没注册扩展的服务器会在启动时直接报错而不是静默降级为同步执行后台任务只支持async函数把taskTrue用在同步函数上会在注册时抛出ValueError只有工具能带taskresources、resource templates 和 prompts 都不带。用 TaskConfig 控制执行模式和轮询间隔taskTrue等价于TaskConfig(modeoptional)。需要更细的控制时用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(taskTaskConfig(modeoptional, poll_intervaltimedelta(seconds2))) async def quick_task() - str: return Done quickly # 要求必须后台执行未 opt-in 的客户端会收到错误 mcp.tool(taskTaskConfig(moderequired)) async def must_be_background() - str: return Only runs as a background task三种执行模式的语义来自文档中的表格mode客户端不带 tasks 能力客户端带 tasks 能力forbiddentaskFalse的默认值同步执行同步执行永不任务化optionaltaskTrue的默认值同步执行作为后台任务执行required报错missing required capability作为后台任务执行poll_interval是服务器建议客户端回查频率的上限而不是精确节拍FastMCP 客户端一开始轮询得快随后回退到该间隔所以快任务仍然几乎立刻被观察到完成。间隔越短客户端反馈越快但服务端负载越大。如果希望默认让服务器上所有工具都支持后台任务可以给构造器传tasksTrue单个装饰器仍可用taskFalse覆盖。注意如果服务器里还有同步工具必须显式写taskFalse否则注册会报错。后端与环境变量TasksExtension()无参调用时读取FASTMCP_DOCKET_*环境变量未设置时落到进程内的memory://后端零配置开箱即用环境变量默认值说明FASTMCP_DOCKET_URLmemory://后端 URLmemory://或redis://host:port/dbFASTMCP_DOCKET_NAMEfastmcp队列名共享队列名和 URL 的服务器与 worker 共用一个队列FASTMCP_DOCKET_CONCURRENCY10每个 worker 的最大并发任务数也可以在构造时直接传参例如mcp.add_extension(TasksExtension(urlredis://localhost:6379/0, concurrency20))。两种后端的取舍和横向扩展见本文可选分支一节。客户端导入 fastmcp_tasks 后透明取回结果客户端文档 强调两点前提客户端任务支持是 opt-in 的在客户端进程里导入fastmcp_tasks你用它调call_tool_task时自然会导入进程中所有Client实例都会声明 tasks 能力不导入的话Client永远不声明该能力服务端就把所有调用当同步请求执行。tasks 只在 modern 协议上协商能力在2026-07-28连接上协商。客户端默认的modeauto会自动协商到它modelegacy的客户端永远不会触发后台任务工具对它始终同步执行。开启之后透明调用就是普通的call_toolimport fastmcp_tasks # enables client task support from fastmcp import Client async with Client(server, modeauto) 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, modeauto) as client: task await call_tool_task(client, slow_computation, {duration: 10}) print(fTask started: {task.task_id}) # 任务运行期间可以做其他工作…… result await task.result()注意call_tool_task要求服务端真的把这次调用作为任务执行——工具不是taskTrue、或服务端没注册扩展时会抛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(timeout30.0) status await task.wait(stateinput_required, timeout30.0) # 取回结果会顺带回答任务中途的输入请求await task 是等价简写 result await task.result() # 取消协作式任务可能在服务端注意到请求前就跑完 await task.cancel()默认情况下失败或被取消的任务会让result()抛ToolError给call_tool_task传raise_on_errorFalse可以改为拿到错误结果。如果任务会中途向客户端提问给Client传一个elicitation_handlercall_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工具taskTaskConfig(poll_intervaltimedelta(seconds1))按duration1–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或 Valkeymcp.add_extension(TasksExtension(urlredis://localhost:6379/0))Redis 后端带来任务在服务器重启后仍然存在、个位数毫秒的拾取延迟、可以加 worker 跨进程甚至跨机器分摊负载。示例目录自带 docker-compose.ymlredis:7-alpine映射到宿主机24242端口分布式跑法cd examples/tasks docker compose up -d export FASTMCP_DOCKET_URLredis://localhost:24242/0 # 或direnv allow python server.py # 一个终端 python -m fastmcp_tasks.worker_cli worker server.py # 另外的终端各跑一个额外 worker每个额外 worker 从同一队列拉任务。worker 并发用环境变量控制export FASTMCP_DOCKET_CONCURRENCY20。两个边界要记住额外 worker 只在 Redis/Valkey 后端下有效memory://是单进程的带任务能力的工具必须在服务器启动时就定义好动态后加的工具不会被 worker 注册无法后台执行。限制与排查工具没后台执行而是同步跑完了通常是客户端进程没有import fastmcp_tasks未声明 tasks 能力或客户端用了modelegacy。此时服务端按同步模式执行该调用。服务器启动报错taskTrue工具存在但TasksExtension未注册或在同步函数上用了taskTrue注册期ValueError或开了tasksTrue全局默认但存在未显式taskFalse的同步工具。call_tool_task抛ToolError服务端没有把这次调用真正任务化工具非taskTrue或扩展未注册。任务中途要输入却报错Client没有配置elicitation_handler时提出输入请求的任务抛ToolError而不是挂住。后台任务里不能await ctx.elicit(...)命令式 elicitation 会阻塞 worker 整个客户端往返的时间任务里要返回InputRequiredResultguard 模式由客户端回答后重入从带任务的工具里调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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 如何在 FastMCP 中把长时间运行的工具配置为后台任务并让客户端透明取回结果？
如何在 FastMCP 中把长时间运行的工具配置为后台任务并让客户端透明取回结果【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcpMCP 中工具调用默认是阻塞的客户端发出请求后要一直等工具返回。当某个工具要跑几十秒甚至几分钟时这种体验很差。FastMCP 实现了 MCP 后台任务扩展SEP-2663io.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再给工具装饰器加taskTrue。import asyncio from fastmcp import FastMCP from fastmcp_tasks import TasksExtension mcp FastMCP(MyServer) mcp.add_extension(TasksExtension()) mcp.tool(taskTrue) async def slow_computation(duration: int) - str: A long-running operation. for i in range(duration): await asyncio.sleep(1) return fCompleted in {duration} seconds有三个必须知道的行为约束taskTrue只是声明该工具可以后台执行真正执行它的是TasksExtension。注册了taskTrue工具却没注册扩展的服务器会在启动时直接报错而不是静默降级为同步执行后台任务只支持async函数把taskTrue用在同步函数上会在注册时抛出ValueError只有工具能带taskresources、resource templates 和 prompts 都不带。用 TaskConfig 控制执行模式和轮询间隔taskTrue等价于TaskConfig(modeoptional)。需要更细的控制时用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(taskTaskConfig(modeoptional, poll_intervaltimedelta(seconds2))) async def quick_task() - str: return Done quickly # 要求必须后台执行未 opt-in 的客户端会收到错误 mcp.tool(taskTaskConfig(moderequired)) async def must_be_background() - str: return Only runs as a background task三种执行模式的语义来自文档中的表格mode客户端不带 tasks 能力客户端带 tasks 能力forbiddentaskFalse的默认值同步执行同步执行永不任务化optionaltaskTrue的默认值同步执行作为后台任务执行required报错missing required capability作为后台任务执行poll_interval是服务器建议客户端回查频率的上限而不是精确节拍FastMCP 客户端一开始轮询得快随后回退到该间隔所以快任务仍然几乎立刻被观察到完成。间隔越短客户端反馈越快但服务端负载越大。如果希望默认让服务器上所有工具都支持后台任务可以给构造器传tasksTrue单个装饰器仍可用taskFalse覆盖。注意如果服务器里还有同步工具必须显式写taskFalse否则注册会报错。后端与环境变量TasksExtension()无参调用时读取FASTMCP_DOCKET_*环境变量未设置时落到进程内的memory://后端零配置开箱即用环境变量默认值说明FASTMCP_DOCKET_URLmemory://后端 URLmemory://或redis://host:port/dbFASTMCP_DOCKET_NAMEfastmcp队列名共享队列名和 URL 的服务器与 worker 共用一个队列FASTMCP_DOCKET_CONCURRENCY10每个 worker 的最大并发任务数也可以在构造时直接传参例如mcp.add_extension(TasksExtension(urlredis://localhost:6379/0, concurrency20))。两种后端的取舍和横向扩展见本文可选分支一节。客户端导入 fastmcp_tasks 后透明取回结果客户端文档 强调两点前提客户端任务支持是 opt-in 的在客户端进程里导入fastmcp_tasks你用它调call_tool_task时自然会导入进程中所有Client实例都会声明 tasks 能力不导入的话Client永远不声明该能力服务端就把所有调用当同步请求执行。tasks 只在 modern 协议上协商能力在2026-07-28连接上协商。客户端默认的modeauto会自动协商到它modelegacy的客户端永远不会触发后台任务工具对它始终同步执行。开启之后透明调用就是普通的call_toolimport fastmcp_tasks # enables client task support from fastmcp import Client async with Client(server, modeauto) 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, modeauto) as client: task await call_tool_task(client, slow_computation, {duration: 10}) print(fTask started: {task.task_id}) # 任务运行期间可以做其他工作…… result await task.result()注意call_tool_task要求服务端真的把这次调用作为任务执行——工具不是taskTrue、或服务端没注册扩展时会抛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(timeout30.0) status await task.wait(stateinput_required, timeout30.0) # 取回结果会顺带回答任务中途的输入请求await task 是等价简写 result await task.result() # 取消协作式任务可能在服务端注意到请求前就跑完 await task.cancel()默认情况下失败或被取消的任务会让result()抛ToolError给call_tool_task传raise_on_errorFalse可以改为拿到错误结果。如果任务会中途向客户端提问给Client传一个elicitation_handlercall_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工具taskTaskConfig(poll_intervaltimedelta(seconds1))按duration1–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或 Valkeymcp.add_extension(TasksExtension(urlredis://localhost:6379/0))Redis 后端带来任务在服务器重启后仍然存在、个位数毫秒的拾取延迟、可以加 worker 跨进程甚至跨机器分摊负载。示例目录自带 docker-compose.ymlredis:7-alpine映射到宿主机24242端口分布式跑法cd examples/tasks docker compose up -d export FASTMCP_DOCKET_URLredis://localhost:24242/0 # 或direnv allow python server.py # 一个终端 python -m fastmcp_tasks.worker_cli worker server.py # 另外的终端各跑一个额外 worker每个额外 worker 从同一队列拉任务。worker 并发用环境变量控制export FASTMCP_DOCKET_CONCURRENCY20。两个边界要记住额外 worker 只在 Redis/Valkey 后端下有效memory://是单进程的带任务能力的工具必须在服务器启动时就定义好动态后加的工具不会被 worker 注册无法后台执行。限制与排查工具没后台执行而是同步跑完了通常是客户端进程没有import fastmcp_tasks未声明 tasks 能力或客户端用了modelegacy。此时服务端按同步模式执行该调用。服务器启动报错taskTrue工具存在但TasksExtension未注册或在同步函数上用了taskTrue注册期ValueError或开了tasksTrue全局默认但存在未显式taskFalse的同步工具。call_tool_task抛ToolError服务端没有把这次调用真正任务化工具非taskTrue或扩展未注册。任务中途要输入却报错Client没有配置elicitation_handler时提出输入请求的任务抛ToolError而不是挂住。后台任务里不能await ctx.elicit(...)命令式 elicitation 会阻塞 worker 整个客户端往返的时间任务里要返回InputRequiredResultguard 模式由客户端回答后重入从带任务的工具里调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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考