mcp-agent MCP Agent Server 实战把 Agent 工作流封装成标准 MCP 服务端【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agentmcp-agent 提供了Agent as Server的完整落地路径用MCPApp注册工作流与工具通过create_mcp_server_for_app一键把 Agent 工作流暴露成标准 MCP Server任何 MCP 客户端Claude Desktop、Cursor、MCP Inspector、自定义程序都能直接调用。本文基于仓库 examples/mcp_agent_server/README.md 及 asyncio、temporal 两套实现完整讲解架构动机、装饰器式工具定义、运行部署、日志回传与多 Agent 互操作读者读完后即可在自己项目中构建、启动并接入 Agent 驱动的 MCP 服务端。引言从客户端消费工具到Agent 即服务端传统 MCP 架构中Claude、Cursor、VS Code 等客户端充当 Agent去消费 MCP Server 暴露的工具。而 mcp-agent 的MCP Agent Server模式把范式反转过来Agents as Servers把 Agent 工作流整体打包为 MCP ServerAgent Interoperability通过标准协议实现多 Agent 互相调用Decoupled ArchitectureAgent 逻辑与客户端界面解耦。为什么要以这种方式暴露 Agent核心收益包括Agent 组合构建多 Agent 系统、平台无关任意 MCP 兼容客户端可用、可扩展在专用基础设施上运行工作流而非限于客户端环境、可复用一次构建、多处消费以及封装性复杂逻辑收敛为定义良好的自包含接口。从源码看这一模式的关键实现集中在 create_mcp_server_for_app它接收一个MCPApp实例扫描其上注册的所有工作流与装饰器工具统一挂载为 FastMCP 工具端点再交给任意 MCP 传输层对外服务。两种执行模式asyncio 内存执行与 Temporal 持久执行仓库在examples/mcp_agent_server/下提供了两种执行引擎的实现通过mcp_agent.config.yaml中的execution_engine字段切换。asyncio 模式examples/mcp_agent_server/asyncio内存执行、零外部依赖部署简单启动与运行速度快适合开发、测试以及复杂度较低的工作流配置见 asyncio/mcp_agent.config.yaml第一行即execution_engine: asyncio并声明fetch、filesystem两个外部 MCP Server分别用uvx mcp-server-fetch与npx -y modelcontextprotocol/server-filesystem启动。Temporal 模式examples/mcp_agent_server/temporal以 Temporal 作为编排引擎实现durable execution工作流可长时间运行、暂停、恢复、自动重试与失败恢复通过 Temporal Web UIhttp://localhost:8233获得工作流可观测性适合生产环境与复杂工作流。对应的 temporal/mcp_agent.config.yaml 将execution_engine设为temporal并提供temporal配置段默认host: localhost:7233、namespace: default、task_queue: mcp-agent、max_concurrent_activities: 10注释中还给出可选的workflow_task_modules预加载声明workflow_task活动的模块与workflow_task_retry_policies按活动覆盖重试策略示例。两种实现的能力对照能力说明协议标准化Agent 通过标准 MCP 协议通信保证互操作性工作流封装复杂 Agent 工作流对外暴露为简单的 MCP 工具执行灵活性可选内存执行asyncio或持久执行Temporal客户端无关性可接入 Claude、VSCode、Cursor、MCP Inspector 或自定义应用多 Agent 生态构建多个 Agent 互相交互协作的系统推荐实践用 app.tool 与 app.async_tool 声明工具原文档明确推荐使用装饰器方式声明工具这是最简开发体验。同步工具直接返回最终结果无需轮询状态异步工具返回workflow_id与run_id配合通用的workflows-get_status端点查询状态与结果。from mcp_agent.app import MCPApp app MCPApp(namemy_agent_server) app.tool async def do_something(arg: str) - str: Do something synchronously and return the final result. return done app.async_tool(namedo_something_async) async def do_something_async(arg: str) - str: Start work asynchronously. Returns workflow_id and run_id. Use workflows-get_status with the returned IDs to retrieve status and results. return started在实际示例 asyncio/main.py 中这一写法被完整落地app.tool装饰的grade_story用ParallelLLM并行驱动 proofreader、fact_checker、style_enforcer 三个子 Agent 并汇入 grader 输出结构化报告app.async_tool装饰的grade_story_async启动同样的并行流程但返回执行 ID 供调用方轮询。app.tool还支持丰富元数据参数见同文件sampling_demoname、title、description、annotations{idempotentHint: False}、icons[Icon(srcemoji:crystal_ball)]与meta{category: demo, feature: sampling}。装饰器工具在调用create_mcp_server_for_app(app)时自动注册无需手动编写工具适配层。Temporal 示例 temporal/main.py 还展示了在装饰器工具中嵌入 base64 图标数据mag.png与结构化输出开关structured_outputFalse的用法。经典工作流路径定义、注册、服务端、客户端工作流定义工作流通过继承Workflow基类并实现run方法定义配合app.workflow与app.workflow_run注册app.workflow class BasicAgentWorkflow(Workflow[str]): app.workflow_run async def run(self, input: str) - WorkflowResult[str]: # Workflow implementation... return WorkflowResult(valueresult)Workflow与WorkflowResult定义于 src/mcp_agent/executor/workflow.py是执行引擎统一调度的基础单元。服务端创建服务端由create_mcp_server_for_app创建可挂载不同传输层mcp_server create_mcp_server_for_app(agent_app) await mcp_server.run_stdio_async()同样可切换为 SSE、Websocket 或 Streamable HTTP 传输。asyncio 示例的main()默认调用run_sse_async()Temporal 示例同样使用 SSE通过--custom-fastmcp-settings参数可注入自定义 FastMCP 配置如{host: localhost, port: 8001, debug: True, log_level: DEBUG}。客户端连接客户端使用gen_client源码位于 src/mcp_agent/mcp/gen_client.py按服务端名称连接from mcp_agent.mcp.gen_client import gen_client async with gen_client(basic_agent_server, context.server_registry) as server: # Call agent workflow tools result await server.call_tool( workflows-BasicAgentWorkflow-run, arguments{run_parameters: {input: Your input here}} )通用工作流端点服务端默认暴露三类通用端点workflows-list列出可用工作流及其参数 schemaworkflows-get_status按run_id可选workflow_id查询运行中工作流的状态workflows-cancel取消正在运行的工作流。Temporal 示例额外提供workflows-resume用于向等待中的工作流发送恢复信号详见后文。当你显式定义工作流类时workflows-Workflow-run、workflows-Workflow-get_status等具名端点仍可用。快速开始asyncio 示例的运行前置条件Python 3.10UV 包管理器Anthropic 与 OpenAI 的 API Key配置密钥cp mcp_agent.secrets.yaml.example mcp_agent.secrets.yaml编辑mcp_agent.secrets.yamlanthropic: api_key: your-anthropic-api-key openai: api_key: your-openai-api-key一键运行客户端脚本在examples/mcp_agent_server/asyncio目录下执行uv run client.py脚本会自动以子进程启动服务端main.py→ 连接服务端 → 运行 BasicAgentWorkflow → 监控并打印工作流状态。分离运行服务端与客户端终端一启动服务端uv run main.py # 可选使用自定义 FastMCP 设置 uv run main.py --custom-fastmcp-settings终端二运行客户端uv run client.py # 可选使用自定义 FastMCP 设置 uv run client.py --custom-fastmcp-settings按功能切片测试feature flagsclient.py支持--features参数选择测试子集可选值workflows、tools、sampling、elicitation、notifications、all# 默认全部功能 uv run client.py # 仅工作流 uv run client.py --features workflows # 仅工具 uv run client.py --features tools # sampling elicitation 演示 uv run client.py --features sampling elicitation # 仅通知服务端日志等 uv run client.py --features notifications # 提高服务端日志详细度 uv run client.py --server-log-level debug控制台输出约定服务端日志以[SERVER LOG] ...前缀出现其他服务端发起的通知如notifications/progress、notifications/resources/list_changed以[SERVER NOTIFY] method: ...出现。客户端轮询逻辑位于 asyncio/client.py对workflows-BasicAgentWorkflow-run返回的WorkflowExecution提取run_id后循环调用workflows-BasicAgentWorkflow-get_status每 5 秒轮询直至状态为completed/error/cancelled随后调用get_token_usage打印该工作流运行的 token 用量与成本明细该工具由mcp.tool定义基于context.token_counter递归聚合各模型输入/输出 token 并估算成本。Temporal 示例持久化执行与暂停/恢复搭建本地 Temporal Server# 安装 Temporal CLI 后启动本地开发服务 temporal server start-dev服务端默认监听localhost:7233与mcp_agent.config.yaml中temporal.host一致Temporal Web UI 位于http://localhost:8233。运行步骤四个终端# 1. 安装依赖 uv pip install -r requirements.txt # 2. 启动 Temporal 服务端如上 # 3. 启动 Temporal Worker uv run basic_agent_server_worker.py # 4. 启动 MCP 服务端 uv run main.py # 5. 运行客户端 uv run client.py其中 worker 脚本 temporal/basic_agent_server_worker.py 使用create_temporal_worker_for_app(app)注册全部工作流并等待任务async def main(): async with create_temporal_worker_for_app(app) as worker: await worker.run()工作流信号暂停与恢复Temporal 的核心演示是PauseResumeWorkflow。其run方法通过执行器的wait_for_signal挂起等待resume信号超时60 秒则抛出不可重试的ApplicationError使整个工作流失败app.workflow class PauseResumeWorkflow(Workflow[str]): app.workflow_run async def run(self, message: str) - WorkflowResult[str]: print(fStarting PauseResumeWorkflow with message: {message}) print(fWorkflow is pausing, workflow_id: {self.id}, run_id: {self.run_id}) # Wait for the resume signal - this will pause the workflow await app.context.executor.wait_for_signal( signal_nameresume, workflow_idself.id, run_idself.run_id, ) print(Signal received, workflow is resuming...) result fWorkflow successfully resumed! Original message: {message} return WorkflowResult(valueresult)完整流程见 temporal/client.py 与 README调用workflows-PauseResumeWorkflow-run启动工作流工作流暂停等待信号用workflows-resume工具携带workflow_id、run_id或 Temporal UI 手动发送resume信号收到信号后工作流继续执行并返回结果。pause_result await server.call_tool( workflows-PauseResumeWorkflow-run, arguments{run_parameters: {message: Custom message for the workflow}} ) execution WorkflowExecution(**json.loads(pause_result.content[0].text)) run_id execution.run_id workflow_id execution.workflow_id await server.call_tool( workflows-resume, arguments{workflow_id: workflow_id, run_id: run_id} )用 Temporal UI 监控工作流打开http://localhost:8233→ 进入 Workflows 区可查看所有工作流执行的状态与详情、历史事件并直接发送信号。这正体现了 Temporal 模式相比内存执行的 out-of-the-box 可观测性优势。客户端侧注意事项Temporal 示例的客户端通过Settings程序化构造execution_engineasyncio、transportsse、urlhttp://127.0.0.1:8000/sse连接服务端。代码注释明确指出客户端承担上游 MCP 客户端角色服务端发起 sampling 时由客户端本地响应审批提示 LLM 调用这些本地流程不在 Temporal 工作流内因此客户端必须使用 asyncio 执行器否则会触发TemporalExecutor.execute must be called from within a workflow错误。同时客户端将human_input_callbackNone关闭采样审批提示以保持非交互而 elicitation 仍由控制台回调处理。接收服务端日志与通知MCP Agent Server 会通告logging能力logging/setLevel并通过notifications/message向上游转发结构化日志。客户端构造会话时传入logging_callback并设置日志级别即可接收from datetime import timedelta from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream from mcp import ClientSession from mcp.types import LoggingMessageNotificationParams from mcp_agent.mcp.mcp_agent_client_session import MCPAgentClientSession async def on_server_log(params: LoggingMessageNotificationParams) - None: print(f[SERVER LOG] [{params.level.upper()}] [{params.logger}] {params.data}) def make_session(read_stream: MemoryObjectReceiveStream, write_stream: MemoryObjectSendStream, read_timeout_seconds: timedelta | None) - ClientSession: return MCPAgentClientSession( read_streamread_stream, write_streamwrite_stream, read_timeout_secondsread_timeout_seconds, logging_callbackon_server_log, ) # 连接后请求服务端的最低日志级别 # await server.set_logging_level(info)两个示例的client.py均端到端演示了这一机制注册logging_callback并调用set_logging_level(info)或通过--server-log-level debug提高详细度服务端日志实时出现在客户端控制台。此外服务端还会推送非日志类通知示例通过继承MCPAgentClientSession重写_received_notification以[SERVER NOTIFY] method: ...格式打印。接入任意 MCP 客户端由于 mcp-agent 应用以 MCP Server 形式暴露可像普通 MCP Server 一样被任何 MCP 客户端消费。MCP Inspectornpx modelcontextprotocol/inspector \ uv \ --directory /path/to/mcp-agent/examples/mcp_agent_server/asyncio \ run \ main.py启动后可在 UI 中查看全部工具、测试工作流执行、观察请求/响应详情。Temporal 版本将目录替换为examples/mcp_agent_server/temporal即可。Claude Desktop定位 Claude Desktop 配置文件通常在~/.claude-desktop/config.json添加服务端配置basic-agent-server: { command: /path/to/uv, args: [ --directory, /path/to/mcp-agent/examples/mcp_agent_server/asyncio, run, main.py ] }重启 Claude Desktop服务端即出现在工具抽屉中Claude Desktop workaround用which uvx、which npx找到可执行文件完整路径更新mcp_agent.config.yamlmcp: servers: fetch: command: /full/path/to/uvx # Replace with your path args: [mcp-server-fetch] filesystem: command: /full/path/to/npx # Replace with your path args: [-y, modelcontextprotocol/server-filesystem]自定义客户端使用gen_client构建自定义客户端见上文客户端连接通过client_session_factory注入自定义会话类如带日志回调的MCPAgentClientSession子类即可把工作流工具接入自己的应用。多 Agent 互操作模式MCP Agent Server 最具价值的能力之一是多 Agent 协作。仓库文档给出的概念示意┌────────────────┐ ┌────────────────┐ │ │ │ │ │ Research │ MCP │ Writing │ │ Agent Server │◄────────┤ Agent Server │ │ │ │ │ └────────────────┘ └────────────────┘ ▲ ▲ │ │ │ ┌────────────┐ │ │ │ │ │ │ │ Claude ├───────┤ │ │ Desktop │ │ │ │ │ │ │ └────────────┘ │典型协作方式Claude Desktop 同时使用两个 Agent ServerWriting Agent 可以把 Research Agent 当作工具调用所有通信均通过 MCP 协议完成。进阶部署到 mcp-agent CloudBetaasyncio 示例支持把应用一键部署为云托管 MCP App# 1. 登录并创建云账号Google/GitHub uv run mcp-agent login # 2. 部署应用 uv run mcp-agent deploy mcp_agent_server -c /absolute/path/to/your/project部署过程中会提示提供 OpenAI/Anthropic 密钥成功后终端会输出应用 ID 与形如https://...deployments.mcp-agent.com的 App URL。README 中的部署日志显示CLI 会处理 secrets 转换复用已有openai.api_keyhandle、打包 bundle、上报部署状态OFFLINE→ 可上线。需要留意的是若main.py存在__main__入口部署时会收到忽略提示因为云端由托管运行时驱动。示例代码结构速览asynciomain.py定义工作流并创建 MCP 服务端、client.py连接服务端并运行工作流、mcp_agent.config.yamlMCP Server 与执行引擎配置、mcp_agent.secrets.yamlAPI Key不入库、short_story.mdParallelWorkflow 测试素材temporalmain.py、basic_agent_server_worker.pyTemporal Worker、client.py、mcp_agent.config.yamlTemporal 执行引擎配置。仓库中还存在examples/mcp_agent_server/context_isolation/目录含 README.md、server.py、clients.py从目录名可以推断它用于演示 MCP Agent Server 在上下文隔离场景下的多客户端并发访问可作为进阶阅读材料。小结MCP Agent Server 模式把构建 Agent与分发 Agent彻底解耦开发阶段用app.tool/app.async_tool或Workflow基类快速封装 Agent 逻辑运行阶段用create_mcp_server_for_app一键暴露为标准 MCP Server再按需选择 asyncio 内存执行开发测试或 Temporal 持久执行生产部署支持暂停/恢复、自动重试与 Web UI 观测。无论消费方是 Claude Desktop、MCP Inspector 还是自定义gen_client客户端接入路径完全一致——这正是 Agent 生态走向互操作与组合的标准姿势。【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考