使用 Temporal 作为 mcp-agent 工作流执行引擎从部署到五种生产级工作流模式实战【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent本篇指南围绕 mcp-agent 仓库中 examples/temporal 的完整示例集展开系统讲解如何将 Temporal 配置为 mcp-agent 的持久化执行引擎并依次剖析 basic、evaluator-optimizer、orchestrator、parallel、router 五种工作流模式的源码实现与运行方式。读完本文你将掌握从启动 Temporal Server、注册 Worker到用app.workflow定义可暂停、可恢复、可重试的工作流再到产出graded_report.md分级报告的完整闭环能力。为什么选择 Temporal 作为执行引擎mcp-agent 原生支持两种执行模式asyncio与temporal二者的切换只需修改 mcp_agent.config.yaml 中的execution_engine配置项一行配置即可完成引擎替换。Temporal 是微服务编排平台为工作流提供了**持久化执行durable execution**能力工作流可以长时间运行不受进程重启影响支持暂停、恢复、重试等运维操作由 Temporal 平台统一保障相同的编排能力虽然可以通过进程内in-proc的asyncio模式实现但官方文档明确建议生产环境的 mcp-agent 部署应使用工作流编排后端即 Temporal。从源码角度看TemporalExecutor是这一能力的核心实现位于 src/mcp_agent/executor/temporal/init.py其类注释将其职责定义为“将workflow作为 Temporal 工作流、将workflow_tasks作为 Temporal 活动activities运行的执行器”。这意味着你使用装饰器声明的业务逻辑会被翻译为 Temporal 的持久化单元从而获得平台级的可靠性保证。前置条件与 Temporal Server 部署运行本示例前需要准备Python 3.10uv包管理器示例的依赖安装与脚本执行均基于uv一个正在运行的 Temporal Server启动 Temporal Server 最快捷的方式是使用 Temporal CLItemporal server start-dev该命令会在localhost:7233启动一个开发模式服务端——这正是 mcp_agent.config.yaml 中配置的默认地址。同时可访问http://localhost:8233打开 Temporal Web UI实时监控工作流的执行状态、历史事件与调度情况。配置解读execution_engine、Temporal 参数与 MCP Serverexamples/temporal/mcp_agent.config.yaml 是整套示例的配置中枢核心内容如下# Set the execution engine to Temporal execution_engine: temporal # Temporal settings temporal: host: localhost:7233 # Default Temporal server address namespace: default # Default Temporal namespace task_queue: mcp-agent # Task queue for workflows and activities max_concurrent_activities: 10 # Maximum number of concurrent activities rpc_metadata: X-Client-Name: mcp-agent mcp: servers: fetch: command: uvx args: [mcp-server-fetch] description: Fetch content at URLs from the world wide web filesystem: command: npx args: [-y, modelcontextprotocol/server-filesystem] description: Read and write files on the filesystem openai: default_model: gpt-4o-mini各参数含义与作用如下配置项示例值说明execution_enginetemporal切换执行引擎的核心开关可选asyncio或temporaltemporal.hostlocalhost:7233Temporal Server 地址与temporal server start-dev默认端口一致temporal.namespacedefaultTemporal 命名空间用于隔离不同业务的工作流temporal.task_queuemcp-agent任务队列名Worker 与客户端必须使用同一队列才能匹配temporal.max_concurrent_activities10最大并发活动数控制 Worker 同时执行的活动数量上限temporal.rpc_metadataX-Client-Name: mcp-agent附加到 RPC 请求的自定义元数据便于在 Temporal 侧识别客户端logger.transports[console, file]日志输出到控制台与 JSONL 文件path_pattern定义按时间戳命名日志文件mcp.serversfetch、filesystem声明工作流可用的 MCP Serverfetch负责抓取网页filesystem负责读写本地文件值得注意的细节filesystem服务器未在配置中预置目录参数而是由各示例在运行时通过context.config.mcp.servers[filesystem].args.extend([os.getcwd()])动态注入当前工作目录例如 basic.py 与 orchestrator.py 中的写法这样示例在不同目录下运行都能指向正确的文件系统根路径。配置校验依赖仓库根目录下的 schema/mcp-agent.config.schema.json配置顶部的$schema字段即指向该 JSON Schema。四步运行依赖、Server、Worker、示例脚本完整运行任何一个示例都需要四个步骤第一步安装依赖uv pip install -r requirements.txtrequirements.txt 声明了四类依赖核心框架mcp-agent通过file://../../链接到本地仓库根目录、anthropic、openai以及 Temporal 的 Python SDKtemporalio。第二步启动 Temporal Server如前述temporal server start-dev第三步在独立终端启动 Workeruv run run_worker.pyWorker 会注册全部工作流并持续等待任务执行。run_worker.py 的实现极为简洁import workflows # noqa: F401 from main import app from mcp_agent.executor.temporal import create_temporal_worker_for_app async def main(): async with create_temporal_worker_for_app(app) as worker: await worker.run()关键在于import workflows这一行workflows.py 集中导入了五个示例中定义的全部工作流类与编排函数Worker 进程只有先完成这些导入Temporal 才能注册并识别对应的工作流类型。create_temporal_worker_for_app函数正是源码 src/mcp_agent/executor/temporal/init.py 中提供的 worker 构造入口。第四步在另一个终端运行示例脚本uv run basic.py # OR uv run evaluator_optimizer.py # OR uv run orchestrator.py # OR uv run parallel.py # OR uv run router.py五种工作流模式源码解析所有示例共享同一个应用入口 main.pyfrom mcp_agent.app import MCPApp # Create the app with Temporal as the execution engine app MCPApp(nametemporal_workflow_example)MCPApp会根据配置文件自动加载temporal执行引擎示例之间仅在工作流定义上存在差异。1. 基础工作流basic.py掌握装饰器三件套basic.py 演示了 Temporal 工作流的最小骨架也是理解后续所有示例的基石app.workflow class SimpleWorkflow(Workflow[str]): app.workflow_run async def run(self, input: str) - WorkflowResult[str]: finder_agent Agent( namefinder, instructionYou are a helpful assistant., server_names[fetch, filesystem], ) context app.context context.config.mcp.servers[filesystem].args.extend([os.getcwd()]) async with finder_agent: finder_llm await finder_agent.attach_llm(OpenAIAugmentedLLM) result await finder_llm.generate_str(messageinput) return WorkflowResult(valueresult) async def main(): async with app.run() as agent_app: executor: TemporalExecutor agent_app.executor handle await executor.start_workflow( SimpleWorkflow, Print the first 2 paragraphs of https://modelcontextprotocol.io/introduction, ) a await handle.result() print(a)核心要素拆解app.workflow声明该类为 Temporal 工作流泛型参数Workflow[str]表示输入类型app.workflow_run标记run方法为工作流执行体接收输入并返回WorkflowResult[str]包装的结果工作流内部创建 Agentfinder代理绑定fetch与filesystem两个 MCP Server在async with上下文中完成 LLM 的挂载attach_llm(OpenAIAugmentedLLM)与生成调用generate_str执行与等待agent_app.executor暴露TemporalExecutor调用start_workflow(SimpleWorkflow, input)返回句柄await handle.result()阻塞等待工作流在 Temporal 侧完成并取回结果。start_workflow的签名定义见 src/mcp_agent/executor/temporal/init.py。2. Evaluator-Optimizer 工作流让反馈循环驱动内容迭代evaluator_optimizer.py 模拟求职场景根据职位描述、候选人信息与公司资料生成求职信再由评估者持续评判、迭代打磨直到达到质量门槛。optimizer Agent( nameoptimizer, instructionYou are a career coach specializing in cover letter writing. ..., server_names[fetch], ) evaluator Agent( nameevaluator, instructionEvaluate the following response based on the criteria below: 1. Clarity: ... 2. Specificity: ... 3. Relevance: ... 4. Tone and Style: ... 5. Persuasiveness: ... 6. Grammar and Mechanics: ... 7. Feedback Alignment: ... For each criterion: - Provide a rating (EXCELLENT, GOOD, FAIR, or POOR). - Offer specific feedback or suggestions for improvement. Summarize your evaluation as a structured response with: - Overall quality rating. - Specific feedback and areas for improvement., ) evaluator_optimizer EvaluatorOptimizerLLM( optimizeroptimizer, evaluatorevaluator, llm_factoryOpenAIAugmentedLLM, min_ratingQualityRating.EXCELLENT, contextapp.context, ) result await evaluator_optimizer.generate_str( messageinput, request_paramsRequestParams(modelgpt-4o), ) return WorkflowResult(valueresult)该模式的关键设计评估者指令中内置了七大评判维度清晰度、具体性、相关性、语气风格、说服力、语法、反馈对齐度并要求给出EXCELLENT / GOOD / FAIR / POOR分级评价EvaluatorOptimizerLLM将二者组装为反馈闭环min_ratingQualityRating.EXCELLENT设定迭代停止条件——只有达到最高评级才结束循环。通过RequestParams(modelgpt-4o)可为单次生成覆盖默认模型。3. Orchestrator 工作流动态编排多代理协作orchestrator.py 展示了更复杂的多代理编排它不直接使用app.workflowapp.workflow_run而是改用app.async_tool装饰器将整个编排逻辑封装为一个异步工具工作流由框架在后台自动创建app.async_tool(nameOrchestratorWorkflow) async def run_orchestrator(input: str, app_ctx: Optional[AppContext] None) - str: ... orchestrator Orchestrator( llm_factoryOpenAIAugmentedLLM, available_agents[ finder_agent, writer_agent, proofreader, fact_checker, style_enforcer, ], plan_typefull, # 每一步都由编排器动态规划 contextcontext, ) return await orchestrator.generate_str( messageinput, request_paramsRequestParams(modelgpt-4o, max_iterations100), )五个 Agent 分工明确finder负责检索文件与网页并返回最近匹配项的 URI 与内容writer将结果写入磁盘proofreader、fact_checker、style_enforcer分别从语法、事实一致性、风格规范三个角度审阅plan_typefull意味着编排器会在每一步动态决定下一步动作。主流程给它下达的任务是读取 short_story.md 中的学生短篇小说参考 APA 风格指南生成涵盖校对、事实逻辑与风格遵循的分级报告并写入 graded_report.md。4. Parallel 工作流Fan-out/Fan-in 并行处理parallel.py 演示扇出/扇入fan-out/fan-in模式将同一篇短篇小说同时分发给校对、事实核查、风格强化三个专家代理并行处理再由grader汇总为结构化报告。parallel ParallelLLM( fan_in_agentgrader, fan_out_agents[proofreader, fact_checker, style_enforcer], llm_factoryOpenAIAugmentedLLM, contextapp.context, ) result await parallel.generate_str( messagefStudent short story submission: {input}, )该示例还附带了token 用量统计的完整实践工作流内部通过parallel.get_token_node()获取 token 树、用量与成本写入WorkflowResult的metadata字段返回主进程除打印返回结果外还尝试通过handle.query(token_tree)与handle.query(token_summary)向运行中的工作流发起 Temporal 查询实时获取远程工作流内的 token 统计——这是 Temporal 查询Query机制在 mcp-agent 中的典型用法代码注释同时提醒Temporal 模式下客户端进程内的TokenCounter可能为 0应优先依赖工作流侧查询的指标。5. Router 工作流面向 Agent、函数与 Server 的智能路由router.py 展示了 LLM 驱动的路由决策能力且覆盖四种路由场景llm OpenAIAugmentedLLM(nameopenai_router, instructionYou are a router) router LLMRouter( llm_factorylambda _agent: llm, agents[finder_agent, writer_agent, reasoning_agent], functions[print_to_console, print_hello_world], contextapp.context, ) # 路由到 Agent读取 mcp_agent.config.yaml 内容 results await router.route_to_agent(request..., top_k1) # 路由到函数打印输入到控制台 results await anthropic_router.route_to_function(request..., top_k2) function_to_call results[0].result function_to_call(Hello, world!) # 仅凭 Server 名称推断路由目标 results await anthropic_router.route_to_server( requestPrint the first two paragraphs of https://modelcontextprotocol.io/introduction, top_k1, ) # 跨所有类别统一路由servers agents callables results await anthropic_router.route(request..., top_k3)四种路由 API 的区别与适用场景路由方法路由目标范围典型场景route_to_agent仅限 Agent 列表把请求分派给最合适的代理执行route_to_function仅限普通 Python 函数把请求映射到本地可调用对象route_to_server仅限 MCP Server根据 Server 名称/描述推断目标工具来源route所有类别混合跨 Agent、函数、Server 的统一全局路由示例同时给出两个路由器实现LLMRouter基于 OpenAI 系 LLM可通过llm_factory注入任意 LLM与AnthropicLLMRouter预配置 Anthropic LLM 并绑定fetch/filesystem两个 Server。top_k控制返回候选数量且示例注释指出路由会依据请求内容做语义匹配——例如请求“打印到控制台”时即便top_k2也只返回print_to_console而不会误返回print_hello_world。项目结构总览examples/temporal/ ├── main.py # 核心应用配置MCPApp 与执行引擎 ├── run_worker.py # Worker 启动脚本create_temporal_worker_for_app ├── workflows.py # 集中导入全部工作流供 Worker 注册 ├── basic.py # 基础工作流示例 ├── evaluator_optimizer.py # 评估-优化迭代示例 ├── orchestrator.py # 多代理动态编排示例 ├── parallel.py # 并行扇出/扇入示例 ├── router.py # 智能路由示例 ├── interactive.py # 带交互的工作流示例 ├── short_story.md # 示例使用的学生短篇小说样本 ├── graded_report.md # orchestrator/parallel 工作流的输出报告 ├── mcp_agent.config.yaml # 引擎与 Temporal 配置 └── requirements.txt # 依赖清单工作流生命周期总结将上述示例串成一条完整链路可以清晰看到 mcp-agent Temporal 的协作模型定义在任意示例脚本中用app.workflow/app.workflow_run或app.async_tool声明工作流注册Worker 进程通过import workflows加载所有工作流定义并由create_temporal_worker_for_app(app)注册到 Temporal启动客户端进程在app.run()上下文中取得TemporalExecutor调用executor.start_workflow(name, input)向任务队列mcp-agent提交执行执行与恢复Temporal 按事件驱动执行工作流活动失败自动重试进程重启后可从历史事件恢复这正是“持久化执行”在生产环境的核心价值取回结果await handle.result()等待完成必要时可通过handle.query(...)对运行中的工作流发起实时查询如 token 统计。如果想验证更复杂的交互式场景还可运行 interactive.py 了解如何在 Temporal 工作流中插入人工输入环节。生产环境若要深入掌控 executor 的并发、重试与查询行为可直接研读 src/mcp_agent/executor/temporal 下的TemporalExecutor实现与其客户端拦截器interceptor。【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
使用 Temporal 作为 mcp-agent 工作流执行引擎：从部署到五种生产级工作流模式实战
使用 Temporal 作为 mcp-agent 工作流执行引擎从部署到五种生产级工作流模式实战【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent本篇指南围绕 mcp-agent 仓库中 examples/temporal 的完整示例集展开系统讲解如何将 Temporal 配置为 mcp-agent 的持久化执行引擎并依次剖析 basic、evaluator-optimizer、orchestrator、parallel、router 五种工作流模式的源码实现与运行方式。读完本文你将掌握从启动 Temporal Server、注册 Worker到用app.workflow定义可暂停、可恢复、可重试的工作流再到产出graded_report.md分级报告的完整闭环能力。为什么选择 Temporal 作为执行引擎mcp-agent 原生支持两种执行模式asyncio与temporal二者的切换只需修改 mcp_agent.config.yaml 中的execution_engine配置项一行配置即可完成引擎替换。Temporal 是微服务编排平台为工作流提供了**持久化执行durable execution**能力工作流可以长时间运行不受进程重启影响支持暂停、恢复、重试等运维操作由 Temporal 平台统一保障相同的编排能力虽然可以通过进程内in-proc的asyncio模式实现但官方文档明确建议生产环境的 mcp-agent 部署应使用工作流编排后端即 Temporal。从源码角度看TemporalExecutor是这一能力的核心实现位于 src/mcp_agent/executor/temporal/init.py其类注释将其职责定义为“将workflow作为 Temporal 工作流、将workflow_tasks作为 Temporal 活动activities运行的执行器”。这意味着你使用装饰器声明的业务逻辑会被翻译为 Temporal 的持久化单元从而获得平台级的可靠性保证。前置条件与 Temporal Server 部署运行本示例前需要准备Python 3.10uv包管理器示例的依赖安装与脚本执行均基于uv一个正在运行的 Temporal Server启动 Temporal Server 最快捷的方式是使用 Temporal CLItemporal server start-dev该命令会在localhost:7233启动一个开发模式服务端——这正是 mcp_agent.config.yaml 中配置的默认地址。同时可访问http://localhost:8233打开 Temporal Web UI实时监控工作流的执行状态、历史事件与调度情况。配置解读execution_engine、Temporal 参数与 MCP Serverexamples/temporal/mcp_agent.config.yaml 是整套示例的配置中枢核心内容如下# Set the execution engine to Temporal execution_engine: temporal # Temporal settings temporal: host: localhost:7233 # Default Temporal server address namespace: default # Default Temporal namespace task_queue: mcp-agent # Task queue for workflows and activities max_concurrent_activities: 10 # Maximum number of concurrent activities rpc_metadata: X-Client-Name: mcp-agent mcp: servers: fetch: command: uvx args: [mcp-server-fetch] description: Fetch content at URLs from the world wide web filesystem: command: npx args: [-y, modelcontextprotocol/server-filesystem] description: Read and write files on the filesystem openai: default_model: gpt-4o-mini各参数含义与作用如下配置项示例值说明execution_enginetemporal切换执行引擎的核心开关可选asyncio或temporaltemporal.hostlocalhost:7233Temporal Server 地址与temporal server start-dev默认端口一致temporal.namespacedefaultTemporal 命名空间用于隔离不同业务的工作流temporal.task_queuemcp-agent任务队列名Worker 与客户端必须使用同一队列才能匹配temporal.max_concurrent_activities10最大并发活动数控制 Worker 同时执行的活动数量上限temporal.rpc_metadataX-Client-Name: mcp-agent附加到 RPC 请求的自定义元数据便于在 Temporal 侧识别客户端logger.transports[console, file]日志输出到控制台与 JSONL 文件path_pattern定义按时间戳命名日志文件mcp.serversfetch、filesystem声明工作流可用的 MCP Serverfetch负责抓取网页filesystem负责读写本地文件值得注意的细节filesystem服务器未在配置中预置目录参数而是由各示例在运行时通过context.config.mcp.servers[filesystem].args.extend([os.getcwd()])动态注入当前工作目录例如 basic.py 与 orchestrator.py 中的写法这样示例在不同目录下运行都能指向正确的文件系统根路径。配置校验依赖仓库根目录下的 schema/mcp-agent.config.schema.json配置顶部的$schema字段即指向该 JSON Schema。四步运行依赖、Server、Worker、示例脚本完整运行任何一个示例都需要四个步骤第一步安装依赖uv pip install -r requirements.txtrequirements.txt 声明了四类依赖核心框架mcp-agent通过file://../../链接到本地仓库根目录、anthropic、openai以及 Temporal 的 Python SDKtemporalio。第二步启动 Temporal Server如前述temporal server start-dev第三步在独立终端启动 Workeruv run run_worker.pyWorker 会注册全部工作流并持续等待任务执行。run_worker.py 的实现极为简洁import workflows # noqa: F401 from main import app from mcp_agent.executor.temporal import create_temporal_worker_for_app async def main(): async with create_temporal_worker_for_app(app) as worker: await worker.run()关键在于import workflows这一行workflows.py 集中导入了五个示例中定义的全部工作流类与编排函数Worker 进程只有先完成这些导入Temporal 才能注册并识别对应的工作流类型。create_temporal_worker_for_app函数正是源码 src/mcp_agent/executor/temporal/init.py 中提供的 worker 构造入口。第四步在另一个终端运行示例脚本uv run basic.py # OR uv run evaluator_optimizer.py # OR uv run orchestrator.py # OR uv run parallel.py # OR uv run router.py五种工作流模式源码解析所有示例共享同一个应用入口 main.pyfrom mcp_agent.app import MCPApp # Create the app with Temporal as the execution engine app MCPApp(nametemporal_workflow_example)MCPApp会根据配置文件自动加载temporal执行引擎示例之间仅在工作流定义上存在差异。1. 基础工作流basic.py掌握装饰器三件套basic.py 演示了 Temporal 工作流的最小骨架也是理解后续所有示例的基石app.workflow class SimpleWorkflow(Workflow[str]): app.workflow_run async def run(self, input: str) - WorkflowResult[str]: finder_agent Agent( namefinder, instructionYou are a helpful assistant., server_names[fetch, filesystem], ) context app.context context.config.mcp.servers[filesystem].args.extend([os.getcwd()]) async with finder_agent: finder_llm await finder_agent.attach_llm(OpenAIAugmentedLLM) result await finder_llm.generate_str(messageinput) return WorkflowResult(valueresult) async def main(): async with app.run() as agent_app: executor: TemporalExecutor agent_app.executor handle await executor.start_workflow( SimpleWorkflow, Print the first 2 paragraphs of https://modelcontextprotocol.io/introduction, ) a await handle.result() print(a)核心要素拆解app.workflow声明该类为 Temporal 工作流泛型参数Workflow[str]表示输入类型app.workflow_run标记run方法为工作流执行体接收输入并返回WorkflowResult[str]包装的结果工作流内部创建 Agentfinder代理绑定fetch与filesystem两个 MCP Server在async with上下文中完成 LLM 的挂载attach_llm(OpenAIAugmentedLLM)与生成调用generate_str执行与等待agent_app.executor暴露TemporalExecutor调用start_workflow(SimpleWorkflow, input)返回句柄await handle.result()阻塞等待工作流在 Temporal 侧完成并取回结果。start_workflow的签名定义见 src/mcp_agent/executor/temporal/init.py。2. Evaluator-Optimizer 工作流让反馈循环驱动内容迭代evaluator_optimizer.py 模拟求职场景根据职位描述、候选人信息与公司资料生成求职信再由评估者持续评判、迭代打磨直到达到质量门槛。optimizer Agent( nameoptimizer, instructionYou are a career coach specializing in cover letter writing. ..., server_names[fetch], ) evaluator Agent( nameevaluator, instructionEvaluate the following response based on the criteria below: 1. Clarity: ... 2. Specificity: ... 3. Relevance: ... 4. Tone and Style: ... 5. Persuasiveness: ... 6. Grammar and Mechanics: ... 7. Feedback Alignment: ... For each criterion: - Provide a rating (EXCELLENT, GOOD, FAIR, or POOR). - Offer specific feedback or suggestions for improvement. Summarize your evaluation as a structured response with: - Overall quality rating. - Specific feedback and areas for improvement., ) evaluator_optimizer EvaluatorOptimizerLLM( optimizeroptimizer, evaluatorevaluator, llm_factoryOpenAIAugmentedLLM, min_ratingQualityRating.EXCELLENT, contextapp.context, ) result await evaluator_optimizer.generate_str( messageinput, request_paramsRequestParams(modelgpt-4o), ) return WorkflowResult(valueresult)该模式的关键设计评估者指令中内置了七大评判维度清晰度、具体性、相关性、语气风格、说服力、语法、反馈对齐度并要求给出EXCELLENT / GOOD / FAIR / POOR分级评价EvaluatorOptimizerLLM将二者组装为反馈闭环min_ratingQualityRating.EXCELLENT设定迭代停止条件——只有达到最高评级才结束循环。通过RequestParams(modelgpt-4o)可为单次生成覆盖默认模型。3. Orchestrator 工作流动态编排多代理协作orchestrator.py 展示了更复杂的多代理编排它不直接使用app.workflowapp.workflow_run而是改用app.async_tool装饰器将整个编排逻辑封装为一个异步工具工作流由框架在后台自动创建app.async_tool(nameOrchestratorWorkflow) async def run_orchestrator(input: str, app_ctx: Optional[AppContext] None) - str: ... orchestrator Orchestrator( llm_factoryOpenAIAugmentedLLM, available_agents[ finder_agent, writer_agent, proofreader, fact_checker, style_enforcer, ], plan_typefull, # 每一步都由编排器动态规划 contextcontext, ) return await orchestrator.generate_str( messageinput, request_paramsRequestParams(modelgpt-4o, max_iterations100), )五个 Agent 分工明确finder负责检索文件与网页并返回最近匹配项的 URI 与内容writer将结果写入磁盘proofreader、fact_checker、style_enforcer分别从语法、事实一致性、风格规范三个角度审阅plan_typefull意味着编排器会在每一步动态决定下一步动作。主流程给它下达的任务是读取 short_story.md 中的学生短篇小说参考 APA 风格指南生成涵盖校对、事实逻辑与风格遵循的分级报告并写入 graded_report.md。4. Parallel 工作流Fan-out/Fan-in 并行处理parallel.py 演示扇出/扇入fan-out/fan-in模式将同一篇短篇小说同时分发给校对、事实核查、风格强化三个专家代理并行处理再由grader汇总为结构化报告。parallel ParallelLLM( fan_in_agentgrader, fan_out_agents[proofreader, fact_checker, style_enforcer], llm_factoryOpenAIAugmentedLLM, contextapp.context, ) result await parallel.generate_str( messagefStudent short story submission: {input}, )该示例还附带了token 用量统计的完整实践工作流内部通过parallel.get_token_node()获取 token 树、用量与成本写入WorkflowResult的metadata字段返回主进程除打印返回结果外还尝试通过handle.query(token_tree)与handle.query(token_summary)向运行中的工作流发起 Temporal 查询实时获取远程工作流内的 token 统计——这是 Temporal 查询Query机制在 mcp-agent 中的典型用法代码注释同时提醒Temporal 模式下客户端进程内的TokenCounter可能为 0应优先依赖工作流侧查询的指标。5. Router 工作流面向 Agent、函数与 Server 的智能路由router.py 展示了 LLM 驱动的路由决策能力且覆盖四种路由场景llm OpenAIAugmentedLLM(nameopenai_router, instructionYou are a router) router LLMRouter( llm_factorylambda _agent: llm, agents[finder_agent, writer_agent, reasoning_agent], functions[print_to_console, print_hello_world], contextapp.context, ) # 路由到 Agent读取 mcp_agent.config.yaml 内容 results await router.route_to_agent(request..., top_k1) # 路由到函数打印输入到控制台 results await anthropic_router.route_to_function(request..., top_k2) function_to_call results[0].result function_to_call(Hello, world!) # 仅凭 Server 名称推断路由目标 results await anthropic_router.route_to_server( requestPrint the first two paragraphs of https://modelcontextprotocol.io/introduction, top_k1, ) # 跨所有类别统一路由servers agents callables results await anthropic_router.route(request..., top_k3)四种路由 API 的区别与适用场景路由方法路由目标范围典型场景route_to_agent仅限 Agent 列表把请求分派给最合适的代理执行route_to_function仅限普通 Python 函数把请求映射到本地可调用对象route_to_server仅限 MCP Server根据 Server 名称/描述推断目标工具来源route所有类别混合跨 Agent、函数、Server 的统一全局路由示例同时给出两个路由器实现LLMRouter基于 OpenAI 系 LLM可通过llm_factory注入任意 LLM与AnthropicLLMRouter预配置 Anthropic LLM 并绑定fetch/filesystem两个 Server。top_k控制返回候选数量且示例注释指出路由会依据请求内容做语义匹配——例如请求“打印到控制台”时即便top_k2也只返回print_to_console而不会误返回print_hello_world。项目结构总览examples/temporal/ ├── main.py # 核心应用配置MCPApp 与执行引擎 ├── run_worker.py # Worker 启动脚本create_temporal_worker_for_app ├── workflows.py # 集中导入全部工作流供 Worker 注册 ├── basic.py # 基础工作流示例 ├── evaluator_optimizer.py # 评估-优化迭代示例 ├── orchestrator.py # 多代理动态编排示例 ├── parallel.py # 并行扇出/扇入示例 ├── router.py # 智能路由示例 ├── interactive.py # 带交互的工作流示例 ├── short_story.md # 示例使用的学生短篇小说样本 ├── graded_report.md # orchestrator/parallel 工作流的输出报告 ├── mcp_agent.config.yaml # 引擎与 Temporal 配置 └── requirements.txt # 依赖清单工作流生命周期总结将上述示例串成一条完整链路可以清晰看到 mcp-agent Temporal 的协作模型定义在任意示例脚本中用app.workflow/app.workflow_run或app.async_tool声明工作流注册Worker 进程通过import workflows加载所有工作流定义并由create_temporal_worker_for_app(app)注册到 Temporal启动客户端进程在app.run()上下文中取得TemporalExecutor调用executor.start_workflow(name, input)向任务队列mcp-agent提交执行执行与恢复Temporal 按事件驱动执行工作流活动失败自动重试进程重启后可从历史事件恢复这正是“持久化执行”在生产环境的核心价值取回结果await handle.result()等待完成必要时可通过handle.query(...)对运行中的工作流发起实时查询如 token 统计。如果想验证更复杂的交互式场景还可运行 interactive.py 了解如何在 Temporal 工作流中插入人工输入环节。生产环境若要深入掌控 executor 的并发、重试与查询行为可直接研读 src/mcp_agent/executor/temporal 下的TemporalExecutor实现与其客户端拦截器interceptor。【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
