基于MCP协议构建生产级智能体:mcp-agent框架详解与实践 1. 项目概述为什么我们需要一个“MCP原生”的智能体框架如果你在过去一年里深度参与过AI应用开发尤其是智能体Agent的构建那你大概率经历过这样的场景为了给大语言模型LLM接入一个外部工具比如读取本地文件、查询数据库或者调用某个SaaS API你需要写一堆胶水代码——处理认证、定义函数签名、编写调用逻辑、处理错误最后还得把这些工具“教”给LLM。这个过程不仅重复而且当你需要组合多个工具或者想让多个智能体协作时代码复杂度会呈指数级增长。更头疼的是生态碎片化。OpenAI有Function CallingAnthropic有Tool Use各家云厂商的SDK也各不相同。你今天为GPT-4o写的工具适配器明天换到Claude 3.5可能就得重写一半。这种“重复造轮子”和“供应商锁定”的感觉正是AI应用工程化路上最大的绊脚石之一。Model Context ProtocolMCP的出现就是为了解决这个根本问题。你可以把它想象成AI世界的“USB协议”或“HTTP协议”。它定义了一套标准让任何工具我们称之为“MCP服务器”都能以一种统一的方式向任何兼容MCP的客户端比如Claude Desktop、Cursor或者我们今天要聊的mcp-agent暴露自己的能力。一个文件系统MCP服务器无论背后是Python、Node.js还是Go实现的它提供的“读取文件”、“列出目录”工具其调用方式对客户端来说是完全一样的。那么有了MCP这个“协议层”我们还需要什么我们需要一个基于此协议、专门为构建生产级智能体应用而设计的框架层。这就是lastmile-ai/mcp-agent项目的核心价值。它不是另一个从零开始的智能体框架而是一个站在MCP巨人肩膀上的“集成开发环境”。它把Anthropic那篇著名的工程指南《Building Effective Agents》中提出的各种高效模式如Map-Reduce、Orchestrator-Worker、Evaluator-Optimizer等用可组合、可复用的Python组件实现出来并且天然地与MCP生态打通。简单来说mcp-agent让你告别了“先选框架再痛苦地适配各种工具”的老路走上了一条“基于标准协议专注智能体行为逻辑”的新路。你的智能体从诞生第一天起就能调用成百上千个现成的MCP工具而你只需要关心你要用这些工具解决什么问题你的智能体应该如何思考和工作2. 核心设计哲学简单、可组合与生产就绪在深入代码之前理解mcp-agent的三个核心设计哲学至关重要。这能帮你判断它是否适合你的项目以及如何更好地利用它。2.1 简单至上告别复杂的图与节点许多智能体框架喜欢用“有向无环图”DAG或“节点”Node来可视化工作流。这对于描述静态的数据流水线很有效但对于动态、多轮、需要条件判断的智能体交互来说有时反而增加了认知负担。mcp-agent选择了一条更“Pythonic”的路用代码定义行为用装饰器管理生命周期。这意味着你构建智能体的思维模式和你写普通Python异步程序的模式几乎一样。你需要循环就用for或while。你需要条件分支就用if-else。框架负责在背后帮你打理好MCP服务器的连接池、工具的动态发现与绑定、LLM的会话管理以及可选的工作流的持久化。你作为开发者关注点可以始终停留在业务逻辑上。2.2 可组合模式像搭乐高一样构建复杂智能体“可组合性”是mcp-agent的杀手锏。它没有提供一个庞杂的、试图解决所有问题的“超级智能体”类而是提供了一系列小巧、专注的“模式”Pattern每个模式解决一类特定问题路由器Router根据用户意图将请求分发给最合适的子智能体或工具。并行处理Map-Reduce将一个大任务拆分成多个子任务分发给多个专家智能体并行处理最后汇总结果。编排器-执行器Orchestrator-Worker一个“经理”智能体负责制定计划、分解任务并协调多个“员工”智能体去执行。评估器-优化器Evaluator-Optimizer一个智能体负责生成内容另一个智能体负责评估并给出反馈循环迭代直至结果达标。这些模式本身是独立的但通过框架提供的工厂方法你可以像搭乐高积木一样将它们组合起来构建出适应复杂场景的复合型智能体。例如你可以先用一个路由器判断用户是想“写代码”还是“分析数据”然后将“写代码”的请求路由给一个由编排器和多个执行器负责前端、后端、测试组成的智能体小组。2.3 生产就绪从原型到部署的无缝衔接很多AI项目死在从Jupyter Notebook到生产服务器的路上。mcp-agent在架构之初就考虑了生产环境的需求持久化执行Durable Execution通过与 Temporal 集成你的智能体工作流可以获得“超能力”。即使进程崩溃、服务器重启工作流也能从断点恢复状态不会丢失。这对于需要长时间运行、涉及人工审批或外部API调用的任务至关重要。配置与密钥管理采用清晰的YAML文件分离环境配置和敏感信息API密钥、OAuth凭证支持环境变量注入符合十二要素应用规范。可观测性内置结构化日志、OpenTelemetry链路追踪和Token用量统计。你可以清晰地知道智能体每一步做了什么、调用了哪些工具、消耗了多少Token。云原生部署提供CLI工具和云平台mcp-agent Cloud目前为Beta版可以一键将你的智能体应用部署为可伸缩的云服务并自动提供HTTPS终端的MCP服务器。这种“开箱即用”的生产力让你能快速验证想法并同样快速、有信心地将验证后的想法推向实际用户。3. 快速上手指南从零构建你的第一个智能体理论说得再多不如动手一试。我们从一个最简单的例子开始感受一下mcp-agent的开发体验。这个智能体将能读取本地文件并抓取网页内容。3.1 环境准备与项目初始化首先确保你使用现代Python环境管理工具。官方强烈推荐使用uv它比传统的pip和venv更快、更可靠。# 1. 安装uv如果尚未安装 curl -LsSf https://astral.sh/uv/install.sh | sh # 2. 创建项目目录并初始化 mkdir my-first-agent cd my-first-agent uv init接下来安装mcp-agent及其对OpenAI的支持我们以GPT-4o为例。uv add mcp-agent[openai]注意方括号[openai]是uv/pip的“额外依赖”语法它会同时安装mcp-agent核心包和openai这个可选依赖。如果你计划使用Anthropic、Google Gemini或Azure OpenAI可以替换或添加为[anthropic, google, azure]。3.2 编写核心应用代码在项目根目录创建main.py写入以下内容import asyncio import os from pathlib import Path from mcp_agent.app import MCPApp from mcp_agent.agents.agent import Agent from mcp_agent.workflows.llm.augmented_llm_openai import OpenAIAugmentedLLM # 1. 创建MCPApp实例这是所有运行时组件的根容器 app MCPApp(namemy_first_agent_app) async def main(): # 2. 启动应用运行时上下文 async with app.run() as running_app: # 获取一个结构化日志记录器 logger running_app.logger logger.info(MCP应用启动成功) # 3. 定义一个智能体Agent # 给它起个名字写一段指令并指定它能使用的MCP服务器 finder_agent Agent( name网络与文件侦察兵, instruction你是一个助手可以读取本地文件系统的内容也可以通过网络获取网页信息。 请根据用户的问题灵活运用这些能力来寻找答案。如果信息不足请如实告知。, server_names[filesystem, fetch], # 关键这里声明了要连接的两个MCP服务器 ) # 4. 在智能体的上下文管理器中工作 async with finder_agent: # 此时框架已自动连接并初始化了filesystem和fetch服务器 # 我们可以列出所有可用的工具 available_tools await finder_agent.list_tools() logger.info(智能体可用工具列表, data{tools: [t.name for t in available_tools]}) # 5. 为智能体附加一个“增强型LLM” # 这会将上一步获取的工具动态注入到LLM的上下文中 llm await finder_agent.attach_llm(OpenAIAugmentedLLM) # 6. 开始与LLM对话它会自动选择并调用合适的工具 # 示例1读取本地README文件 file_result await llm.generate_str( message请帮我查看当前目录下README.md文件的内容并原样显示出来。 ) logger.info(文件读取结果, data{content: file_result}) # 示例2抓取网页并总结 web_result await llm.generate_str( message请访问 https://www.anthropic.com/research/building-effective-agents 并总结其前两段的核心观点。 ) logger.info(网页总结结果, data{summary: web_result}) # 示例3多轮对话 - 基于网页内容生成推文 tweet await llm.generate_str(将刚才总结的核心观点压缩成一条不超过280字符的推文。) logger.info(生成的推文, data{tweet: tweet}) if __name__ __main__: # 运行异步主函数 asyncio.run(main())这段代码清晰地展示了mcp-agent的核心工作流创建应用MCPApp管理全局配置、日志和服务器连接池。定义智能体Agent封装了身份指令和能力边界可用的服务器。附加增强型LLMAugmentedLLM将智能体的能力工具与LLM的推理能力结合。交互像调用普通LLM API一样提问LLM在背后会自动判断何时、如何调用工具。3.3 配置MCP服务器与密钥代码中我们引用了filesystem和fetch服务器但还没告诉框架去哪里找它们。我们需要一个配置文件。在项目根目录创建mcp_agent.config.yaml# mcp_agent.config.yaml execution_engine: asyncio # 执行引擎先用简单的异步模式。生产环境可换为temporal logger: transports: [console] # 日志输出到控制台。也可用[file, console]同时输出到文件 level: info # 日志级别: debug, info, warning, error mcp: servers: # 定义名为“fetch”的MCP服务器 fetch: command: uvx # 使用uvx全局运行器来启动这个服务器 args: [mcp-server-fetch] # 服务器包名uvx会自动安装并运行它 # 定义名为“filesystem”的MCP服务器 filesystem: command: npx # 使用Node.js的npx来启动 args: - -y - modelcontextprotocol/server-filesystem - . # 这里指定文件服务器可以访问的根目录.代表当前项目目录 # 注意在生产环境中务必限制目录范围例如 /safe/path openai: default_model: gpt-4o # 设置默认使用的OpenAI模型接下来是敏感信息。创建mcp_agent.secrets.yaml文件务必将其加入.gitignore# mcp_agent.secrets.yaml openai: api_key: ${OPENAI_API_KEY} # 从环境变量OPENAI_API_KEY读取然后在终端设置你的OpenAI API密钥export OPENAI_API_KEY你的-api-key-here或者你也可以直接将密钥写在secrets.yaml里不推荐尤其是提交到版本库时。3.4 运行与观察现在在终端运行你的智能体uv run main.py你会看到类似以下的输出INFO:mcp_agent: MCP应用启动成功 INFO:mcp_agent: 智能体可用工具列表 data{tools: [read_file, list_directory, fetch_url]} INFO:mcp_agent: 文件读取结果 data{content: # My First Agent Project\n\nWelcome to...} INFO:mcp_agent: 网页总结结果 data{summary: 文章前两段强调了构建高效智能体的关键在于...} INFO:mcp_agent: 生成的推文 data{tweet: 刚读完Anthropic的智能体构建指南核心观点...}恭喜你的第一个能“动手动脚”操作文件和网络的智能体已经运行起来了。整个过程你没有写一行工具调用的具体代码所有read_file、fetch_url的逻辑都由MCP服务器提供并由mcp-agent框架自动协调完成。4. 核心组件深度解析理解了基本流程后我们来拆解mcp-agent的几个核心组件看看它们是如何协作的以及你该如何定制它们。4.1 MCPApp应用的基石与指挥中心MCPApp是你的智能体应用的单例入口和依赖注入容器。它负责加载配置合并mcp_agent.config.yaml、mcp_agent.secrets.yaml和环境变量。管理生命周期通过async with app.run()上下文管理器确保所有资源服务器连接、数据库池等被正确初始化和清理。提供服务发现内部维护一个ServerRegistry让你可以通过名字获取到已连接的MCP服务器客户端。注册工作流与工具你可以用app.workflow、app.tool等装饰器将你的函数暴露为可被智能体调用的能力。实操心得配置的优先级与覆盖配置的加载顺序决定了最终生效的值。mcp-agent的配置优先级从低到高是框架默认值mcp_agent.config.yaml文件mcp_agent.secrets.yaml文件环境变量支持${VAR_NAME}语法在YAML中引用在代码中通过Settings对象传入MCPApp的构造参数。这意味着你可以在开发环境使用YAML文件配置而在CI/CD或容器化部署时完全通过环境变量来覆盖敏感信息和环境特定配置这非常符合云原生应用的最佳实践。4.2 Agent与AgentSpec智能体的蓝图与实例Agent类代表一个具体的、可运行的智能体实例。它包含两大要素指令Instruction一段自然语言描述定义了智能体的角色、目标和行为准则。这是引导LLM行为的关键。能力边界通过server_names参数指定它可以访问哪些MCP服务器从而间接决定了它可以使用哪些工具。AgentSpec则是Agent的“蓝图”或“配置定义”。它通常以YAML或JSON格式存储允许你将智能体的定义与代码分离。这在以下场景非常有用动态智能体加载根据运行时情况从数据库或配置中心加载不同的智能体蓝图。工厂模式使用load_agent_specs_from_file等工厂方法批量创建智能体。作为路由器的输入路由器模式需要一组AgentSpec作为候选路由目标。# agents.yaml - 一个AgentSpec的例子 - name: research_assistant instruction: 你是一个研究助理擅长从网络和学术数据库中查找、总结信息。 你的回答应当严谨、有引用来源。 server_names: [fetch, scholar_mcp] # 假设有一个学术数据库的MCP服务器 icon: data:image/svgxml;base64,... # 可选用于UI展示的图标注意事项指令工程Instruction Engineering的质量Agent的instruction字段看似简单实则极大影响智能体表现。写得模糊智能体可能行为不稳定写得太死板又限制了其灵活性。一个好的指令应明确角色你是谁例如“资深软件架构师”目标你的核心任务是什么例如“分析代码仓库并提出重构建议”约束与规范你必须遵守什么规则例如“优先使用提供的工具不要假设未给出的信息”、“所有结论必须有代码或文档依据”输出格式你应如何呈现结果例如“以Markdown列表形式给出每个问题包含文件名、问题描述、建议修改”在实践中你需要像产品经理一样反复打磨这段指令并通过测试用例来验证其效果。4.3 Augmented LLM连接大脑与手脚的桥梁AugmentedLLM增强型LLM是框架中最精妙的部分之一。它不是一个新模型而是一个装饰器模式的封装类。当你调用agent.attach_llm(OpenAIAugmentedLLM)时发生了几件事工具发现与格式化框架向agent指定的所有MCP服务器查询可用的工具列表并将这些工具的描述转换成LLM能理解的格式如OpenAI的function calling schema。会话管理创建一个维护对话历史的会话上下文。工具调用循环当你使用llm.generate_str()发起请求时它会进入一个循环将用户消息、历史、工具描述发送给LLM。LLM返回一个包含可能工具调用的响应。AugmentedLLM拦截这个响应如果包含工具调用则自动执行对应的MCP工具。将工具执行的结果作为新的上下文再次发送给LLM。重复此过程直到LLM返回一个不包含工具调用的纯文本回答循环结束。这个过程对开发者是完全透明的。你得到的就是最终答案。这极大地简化了代码因为你不必手动解析LLM的响应、调用函数、再把结果拼接回对话。高级用法结构化输出Structured Output除了生成文本AugmentedLLM还支持生成结构化数据这对于需要将LLM输出集成到后续自动化流程中非常有用。from pydantic import BaseModel from mcp_agent.workflows.llm.augmented_llm import RequestParams # 1. 定义你期望的数据结构 class CodeReview(BaseModel): file_path: str issue_type: str # e.g., bug, performance, style description: str suggestion: str severity: int # 1-5, 5为最严重 async with agent: llm await agent.attach_llm(OpenAIAugmentedLLM) # 2. 使用generate_structured方法并指定response_model review_results await llm.generate_structured( message请分析当前目录下main.py文件找出潜在问题并提供改进建议。, response_modellist[CodeReview], # 可以返回一个列表 request_paramsRequestParams(maxTokens1000, temperature0.1) # 控制生成参数 ) # 3. review_results 现在是一个类型安全的CodeReview对象列表 for issue in review_results: if issue.severity 4: print(f[严重] {issue.file_path}: {issue.description})4.4 工作流与装饰器从脚本到可持久化服务简单的智能体交互用上面的模式就够了。但对于复杂的、多步骤的、可能长时间运行的业务流程你需要“工作流”Workflow。mcp-agent通过一组装饰器让你能用编写普通异步函数的方式定义出可以被持久化执行引擎管理的工作流。from datetime import timedelta from typing import List from mcp_agent.executor.workflow import Workflow, WorkflowResult from pydantic import BaseModel # 定义工作流的输入输出模型 class ResearchInput(BaseModel): topic: str depth: str overview # overview | deep class ResearchOutput(BaseModel): report: str sources: List[str] # 使用app.workflow装饰器声明一个工作流类 app.workflow class ResearchWorkflow(Workflow[WorkflowResult[ResearchOutput]]): # 工作流任务可以被单独调度、重试、超时控制 app.workflow_task(schedule_to_close_timeouttimedelta(minutes10)) async def search_web(self, topic: str) - List[str]: 使用fetch服务器搜索网络信息 async with Agent(namesearcher, instructionSearch the web., server_names[fetch]) as agent: llm await agent.attach_llm(OpenAIAugmentedLLM) # 这里可以设计更复杂的搜索和筛选逻辑 result await llm.generate_str(fSearch for recent and reliable articles about {topic}.) # 模拟返回找到的URL列表 return [fhttps://example.com/article1-about-{topic}, fhttps://example.com/article2-about-{topic}] app.workflow_task(schedule_to_close_timeouttimedelta(minutes5)) async def summarize_articles(self, urls: List[str]) - str: 总结找到的文章 async with Agent(namesummarizer, instructionSummarize articles concisely., server_names[fetch]) as agent: llm await agent.attach_llm(OpenAIAugmentedLLM) combined_summary for url in urls: summary await llm.generate_str(fSummarize the key points from {url} in three bullet points.) combined_summary f## Summary of {url}\n{summary}\n\n return combined_summary # 工作流的入口点用app.workflow_run装饰 app.workflow_run async def run(self, input_data: ResearchInput) - WorkflowResult[ResearchOutput]: # 1. 搜索 sources await self.search_web(input_data.topic) # 2. 总结 report await self.summarize_articles(sources) # 3. 返回结果 return WorkflowResult(valueResearchOutput(reportreport, sourcessources))这段代码的魔力在于当你把execution_engine从asyncio切换到temporal时整个工作流的执行状态会被Temporal持久化。即使执行summarize_articles任务的进程突然崩溃Temporal也会在另一个健康的Worker上重新调度这个任务并从search_web完成后的状态继续执行确保业务流程的可靠性。5. 高级模式实战构建一个复合型智能体系统现在我们利用框架提供的可组合模式构建一个稍微复杂点的系统一个能处理用户多种请求的“智能客服中心”。它包含一个路由器根据问题类型将请求分发给代码专家、文档研究员或通用助手。5.1 定义多个专业智能体蓝图首先我们在agents.yaml中定义三个不同专长的智能体蓝图# agents.yaml - name: code_expert instruction: 你是一个经验丰富的软件工程师精通多种编程语言和框架。 你的任务是分析代码、调试问题、提供重构建议和编写代码片段。 回答必须专业、准确并包含可运行的代码示例。 server_names: [filesystem, github_mcp] # 假设有GitHub的MCP服务器 icon: - name: research_specialist instruction: 你是一个严谨的研究员和信息整理专家。 擅长从网络、文档和知识库中查找、验证和总结信息。 你的回答应当结构清晰、引用来源并区分事实与观点。 server_names: [fetch, confluence_mcp] # 假设有Confluence的MCP服务器 icon: - name: general_assistant instruction: 你是一个友好、乐于助人的通用助手。 可以回答一般性问题进行创意写作或者进行轻松的对话。 如果你无法处理用户的问题应礼貌地引导他们联系更专业的客服。 server_names: [] # 这个助手不直接访问外部工具仅依赖LLM自身知识 icon: 5.2 构建路由智能体接下来我们创建一个主程序它加载这些蓝图并使用create_router_llm工厂方法构建一个路由智能体。# router_agent.py import asyncio from pathlib import Path from mcp_agent.app import MCPApp from mcp_agent.workflows.factory import load_agent_specs_from_file, create_router_llm app MCPApp(namecustomer_support_center) async def main(): async with app.run() as running_app: logger running_app.logger # 1. 从YAML文件加载智能体蓝图 specs load_agent_specs_from_file( str(Path(./agents.yaml)), contextrunning_app.context, ) logger.info(f成功加载 {len(specs)} 个智能体蓝图) # 2. 创建路由增强型LLM # 路由器本身也是一个AugmentedLLM它内部会判断用户意图并选择最合适的子智能体来处理 router_llm await create_router_llm( agentsspecs, # 传入候选智能体列表 provideropenai, # 指定路由器自身使用的LLM提供商 contextrunning_app.context, router_instruction你是一个智能路由分发器。请根据用户的问题将其分配给最合适的专家处理。 可选专家有 - code_expert (): 处理任何与编程、代码、软件工程、调试、技术架构相关的问题。 - research_specialist (): 处理需要查找资料、总结文档、研究分析、事实核查类的问题。 - general_assistant (): 处理日常聊天、创意写作、简单问答及其他不属于上述两类的问题。 请只输出你选择的专家名字不要输出其他任何解释。, ) # 3. 测试路由功能 test_queries [ 帮我看看这段Python代码为什么报错print(Hello, World, 总结一下最近三个月关于‘AI Agent框架’的主要技术文章观点。, 今天天气怎么样, 如何用React实现一个可拖拽的组件, ] for query in test_queries: logger.info(f用户提问: {query}) # 路由器不直接回答问题而是返回应该由哪个专家来处理 routed_agent_name await router_llm.generate_str(messagequery) logger.info(f路由器决策: 分配给 [{routed_agent_name.strip()}] 处理) print(- * 40) # 4. 模拟完整处理流程这里简化实际需根据路由结果调用对应智能体 # 在实际应用中你会根据routed_agent_name去实例化对应的Agent并执行任务。 if __name__ __main__: asyncio.run(main())运行这个程序你会看到路由器根据问题内容将其分发给不同的专家。例如代码问题会路由给code_expert研究类问题给research_specialist而“今天天气怎么样”这种问题可能会交给general_assistant。5.3 组合更复杂的模式并行处理Map-Reduce假设我们的research_specialist接到一个大型研究任务比如“分析AI在医疗、金融、教育三个领域的应用现状”。我们可以让它在内部使用并行处理Map-Reduce模式同时研究三个子领域最后汇总报告。# map_reduce_example.py import asyncio from mcp_agent.app import MCPApp from mcp_agent.agents.agent import Agent from mcp_agent.workflows.factory import create_parallel_llm from mcp_agent.workflows.llm.augmented_llm_openai import OpenAIAugmentedLLM app MCPApp(nameparallel_research_demo) async def parallel_research_task(topic: str, subtopics: list[str]) - str: 并行研究一个主题下的多个子主题 async with app.run() as running_app: # 1. 为每个子主题创建一个专门的研究员智能体 specialist_agents [] for subtopic in subtopics: agent Agent( namefresearcher_{subtopic[:10]}, instructionf你是一名专注于{topics}领域的研究员特别是{subtopic}方向。请深入查找并总结相关信息。, server_names[fetch], # 每个研究员都可以上网 ) specialist_agents.append(agent) # 2. 创建并行LLM它会将任务分发给所有智能体然后汇总结果 parallel_llm await create_parallel_llm( agentsspecialist_agents, provideropenai, contextrunning_app.context, reducer_instruction你是一名主编。以下是多位研究员关于同一主题不同侧面的报告。 请将他们的发现整合成一份连贯、全面、结构清晰的最终报告。 报告应包含引言、各子领域分析、综合结论和未来展望。, ) # 3. 执行并行研究 # 每个子智能体会同时收到“请研究{topic}领域下的{subtopic}”的指令 final_report await parallel_llm.generate_str( messagef请深入研究{topic}领域并提交你的分析报告。 # 注意在实际的create_parallel_llm实现中消息传递和结果汇总的机制可能更复杂。 # 这里是一个概念性示例。具体API请参考最新文档。 ) return final_report async def main(): topic 人工智能应用 subtopics [医疗健康, 金融服务, 智能教育] print(f开始并行研究: {topic}) report await parallel_research_task(topic, subtopics) print(f\n 最终整合报告 \n{report}) if __name__ __main__: asyncio.run(main())这个例子展示了如何将多个智能体组织起来协同完成一个任务。create_parallel_llm工厂方法封装了“分发任务 - 并行执行 - 聚合结果”的完整逻辑你只需要提供一组智能体和聚合指令即可。6. 部署与生产化考量让智能体在本地跑起来只是第一步。要让它成为可靠的服务还需要考虑部署、监控和维护。6.1 配置持久化执行Temporal要将上面ResearchWorkflow例子变成持久化的首先需要修改配置# mcp_agent.config.yaml (生产配置片段) execution_engine: temporal # 关键切换执行引擎 temporal: namespace: your-namespace task_queue: mcp-agent-tasks # Temporal服务地址可以是本地开发集群或云服务 address: localhost:7233 # 工作流和活动的重试策略 retry_policy: initial_interval: 1s backoff_coefficient: 2.0 maximum_interval: 60s然后你需要启动一个Temporal Worker来执行你的工作流# worker.py import asyncio from mcp_agent.app import MCPApp from mcp_agent.executor.temporal import create_temporal_worker_for_app # 确保导入你定义的工作流类以便Worker注册它们 from your_workflow_module import ResearchWorkflow app MCPApp(nameproduction_agent_app) async def main(): async with create_temporal_worker_for_app(app) as worker: # Worker会持续运行监听task_queue并执行分配到该队列的工作流和活动任务 print(Temporal Worker 已启动并开始监听任务...) await worker.run() # 这是一个阻塞调用直到进程被终止 if __name__ __main__: asyncio.run(main())现在你可以通过Temporal客户端或mcp-agent的CLI来触发工作流即使你的Worker进程重启工作流状态也不会丢失。6.2 可观测性配置生产环境必须要有日志和监控。mcp-agent内置了强大的可观测性支持。# mcp_agent.config.yaml logger: transports: [console, file] # 同时输出到控制台和文件 level: info path: logs/app-{unique_id}.jsonl # 结构化JSON日志每行一个JSON对象 path_settings: unique_id: timestamp timestamp_format: %Y%m%d_%H%M%S otel: # OpenTelemetry配置 enabled: true exporters: - console # 开发时输出到控制台 # - otlp # 生产环境可配置为发送到Jaeger、Tempo等后端 service_name: mcp-agent-production在代码中你可以通过running_app.logger记录结构化日志方便后续用ELK、Loki等工具进行分析。6.3 将智能体暴露为MCP服务器一个强大的特性是你可以将你的MCPApp本身变成一个MCP服务器。这意味着你编写的工具和工作流可以被任何兼容MCP的客户端如Claude Desktop、Cursor、VS Code插件直接调用。# mcp_server.py from mcp_agent.app import MCPApp from mcp_agent.server import create_mcp_server_for_app from pydantic import BaseModel app MCPApp(namemy_tool_server) # 定义一个工具它将通过MCP协议暴露给客户端 app.tool async def analyze_sentiment(text: str) - dict: 分析一段文本的情感倾向。 # 这里可以集成你的情感分析模型或调用其他API # 简单示例 positive_words [好, 优秀, 高兴, 满意] negative_words [差, 糟糕, 伤心, 不满] score sum(1 for w in positive_words if w in text) - sum(1 for w in negative_words if w in text) sentiment 积极 if score 0 else 消极 if score 0 else 中性 return {sentiment: sentiment, score: score, original_text: text} app.tool class FormatCodeRequest(BaseModel): code: str language: str python app.tool async def format_code(request: FormatCodeRequest) - str: 格式化代码。 # 这里可以调用black、prettier等格式化工具 # 示例简单处理 if request.language python: # 模拟格式化去除首尾空格确保末尾有换行 return request.code.strip() \n return request.code if __name__ __main__: # 创建MCP服务器并运行在标准输入输出上适用于Claude Desktop等 server create_mcp_server_for_app(app) server.run_stdio() # 也可以运行在HTTP或SSE上 # server.run_sse(port8080)运行这个脚本然后在Claude Desktop的设置中添加一个本地MCP服务器指向这个脚本或编译后的可执行文件。之后你在Claude Desktop里就可以直接调用analyze_sentiment和format_code这两个工具了。6.4 使用CLI进行云部署Betamcp-agent提供了命令行工具简化项目管理与部署。# 1. 登录到mcp-agent CloudBeta uvx mcp-agent login # 2. 初始化一个新项目如果还没做 uvx mcp-agent init --template basic --dir my-cloud-agent # 3. 部署应用到云端 uvx mcp-agent deploy my-cloud-agent # 部署后你会获得一个HTTPS端点可以作为MCP服务器使用。 # 4. 查看已部署的应用 uvx mcp-agent cloud apps list # 5. 查看应用日志 uvx mcp-agent cloud logs my-cloud-agent云服务会帮你管理Temporal集群、提供HTTPS终结点、处理密钥的安全存储让你能更专注于业务逻辑。7. 常见问题与排查技巧在实际开发中你可能会遇到一些典型问题。以下是一些排查思路和解决方案。7.1 MCP服务器连接失败问题启动应用时日志报错Failed to initialize server filesystem或类似连接错误。检查命令路径确认mcp_agent.config.yaml中command和args是否正确。对于npx或uvx启动的服务器确保这些工具已全局安装npm install -g modelcontextprotocol/server-filesystem或pipx install mcp-server-fetch。手动测试尝试在终端直接运行配置中的命令例如npx -y modelcontextprotocol/server-filesystem .看能否独立启动服务器。权限问题某些服务器如filesystem可能需要访问特定目录确保应用有该目录的读写权限。网络策略如果服务器需要通过网络访问如数据库MCP服务器确保网络连通。7.2 LLM不调用工具问题向智能体提问后LLM直接回答了但没有按预期调用工具。检查工具列表在async with agent:块内使用tools await agent.list_tools()打印一下确认工具是否被正确发现和加载。审查指令Instruction智能体的instruction是否清晰指示了它应该使用工具例如“你必须使用提供的工具来获取信息”比“你可以使用工具”更强制。优化提示词有时LLM需要更明确的指引。尝试在用户消息中更直接地要求它使用特定工具例如“请使用read_file工具查看config.yaml文件的内容”。LLM参数尝试调整RequestParams例如降低temperature如0.1可以使LLM的行为更确定、更倾向于遵循指令。7.3 处理速率限制和长耗时操作问题调用外部API或执行复杂操作时遇到速率限制或超时。配置重试与超时在mcp_agent.config.yaml中为MCP服务器配置重试策略。mcp: servers: my_slow_api: command: python args: [my_server.py] # 服务器进程级别的超时设置 timeout: 300 # 秒 # 客户端调用工具时的重试策略 client_retry_policy: max_attempts: 3 initial_delay: 1.0 max_delay: 30.0异步与并发控制对于需要调用多个独立工具的智能体考虑使用asyncio.gather来并发执行但要注意目标API的并发限制。使用工作流与持久化对于非常耗时的任务如处理大量数据务必将其设计为Workflow并使用temporal引擎。这样即使进程中断任务也能恢复。7.4 调试与日志分析问题智能体行为不符合预期需要深入调试。启用Debug日志将配置中的logger.level设为debug。这会输出MCP协议通信的详细内容、工具调用请求和响应、Token消耗等是排查问题的第一手资料。结构化日志查询如果你将日志输出到文件JSON Lines格式可以使用jq等工具进行过滤分析。例如jq select(.msg tool_call) logs/app-*.jsonl可以查看所有工具调用记录。使用Token计数器框架内置了Token计数功能。你可以在代码中订阅Token使用事件实时监控成本。async with app.run() as running_app: token_counter running_app.context.token_counter async def monitor(usage): print(fToken使用更新: 本次{usage.completion_tokens}补全, {usage.prompt_tokens}提示, 总计{usage.total_tokens}) watch_id await token_counter.watch(callbackmonitor, node_typellm) # ... 执行你的智能体逻辑 ... await token_counter.unwatch(watch_id)7.5 安全与权限管理问题如何控制智能体对资源的访问权限最小权限原则在mcp_agent.config.yaml中为每个MCP服务器配置尽可能严格的访问范围。例如文件系统服务器只授予对特定项目目录的读取权限而非整个硬盘。环境隔离为开发、测试、生产环境使用不同的配置文件和密钥。永远不要在代码或配置文件中硬编码生产密钥。审计日志确保所有工具调用都被记录到不可篡改的审计日志中包括调用参数敏感信息可脱敏和结果。人工审核环节对于高风险操作如删除文件、发送邮件、执行数据库写操作在设计工作流时使用request_human_input功能强制插入人工审批节点。8. 总结与最佳实践经过以上几个章节的拆解相信你已经对mcp-agent有了比较全面的认识。它不是一个颠覆性的新概念而是一个将现有最佳实践MCP协议、有效智能体模式、持久化执行进行了优雅集成和工程化封装的框架。最后分享几点我个人在实践中的体会和最佳实践1. 始于简单渐进复杂不要一开始就试图设计一个包含路由、并行、评估的庞大智能体系统。从一个能解决具体问题的最小可行智能体如文件查找器开始验证MCP工具链和基础流程。然后逐步引入更复杂的模式。2. 指令Instruction是灵魂花在打磨智能体instruction上的时间远比调整模型参数更有价值。好的指令应该清晰、具体、有约束力并且经过大量真实场景的测试迭代。可以考虑为不同的智能体角色建立指令模板库。3. 配置即代码环境分离充分利用YAML配置和环境变量的灵活性。将服务器连接信息、模型选择、超时设置等放在配置文件中。将API密钥、数据库密码等敏感信息通过mcp_agent.secrets.yaml或环境变量管理。这能让你的应用在不同环境间轻松迁移。4. 拥抱持久化工作流对于任何超出简单问答、涉及多步骤、可能失败或需要人工干预的任务尽早将其重构为Workflow并使用temporal引擎。这看似增加了前期复杂度但为未来的可靠性、可观测性和可维护性打下了坚实基础。5. 善用社区生态MCP的核心优势在于其生态。定期关注 MCP服务器列表 你会发现有大量现成的工具可供集成从数据库Postgres, MySQL到云服务AWS, GitHub, Slack再到开发工具Docker, Sentry。这能极大扩展你智能体的能力边界而无需自己从头编写集成。6. 监控与成本控制从第一天起就启用结构化日志和Token计数。清晰了解每个请求的路径、耗时和成本这对于优化提示词、调整工作流、控制预算至关重要。可以考虑设置Token消耗的警报阈值。mcp-agent降低的是“构建基于MCP的可靠智能体应用”的工程复杂度而不是AI智能体设计本身的复杂度。它给了你一套强大的工具箱和稳固的脚手架但如何设计出真正有用、可靠的智能体依然需要你对业务逻辑的深刻理解和对LLM能力的持续探索。希望这篇详尽的指南能帮助你顺利起步在这个充满可能性的新范式下构建出令人惊艳的AI应用。