1. 项目概述：用 MCP-Agent 构建你的第一个智能体应用

如果你最近在关注 AI 应用开发，尤其是智能体（Agent）领域，那你大概率听说过 Model Context Protocol（MCP）。简单来说，MCP 是一个标准协议，它让大语言模型（LLM）能够安全、标准化地访问外部工具和数据源，比如文件系统、数据库、API 等等。这解决了智能体开发中的一个核心痛点：如何让 LLM 可靠地“动手操作”外部世界。

但 MCP 本身是一个底层协议，直接用它来构建复杂的、生产级的智能体应用，就像用 TCP 套接字直接写 Web 服务一样，虽然灵活，但需要处理大量繁琐的细节：服务器生命周期管理、连接池、错误重试、工具编排、状态持久化……这还没算上如何实现那些被验证有效的智能体模式，比如 Map-Reduce、Orchestrator-Worker、Evaluator-Optimizer 等。

这就是 mcp-agent 出现的原因。它不是一个全新的框架，而是一个基于 MCP 的“上层建筑”。你可以把它理解为一个 “MCP 原生”的智能体开发框架 。它的核心愿景很明确： MCP 是你构建智能体所需的一切，而简单、可组合的模式比复杂的架构更能产出高质量的、可交付的智能体 。

mcp-agent 为你抽象了所有 MCP 连接的脏活累活，并内置实现了 Anthropic 《构建高效智能体》指南中的所有经典模式。更重要的是，它从设计之初就考虑了生产环境，通过集成 Temporal 提供了开箱即用的“持久化执行”（Durable Execution）能力，这意味着你的智能体可以暂停、恢复、重试，而无需修改业务代码。无论你是想快速验证一个想法，还是构建一个需要 7x24 小时运行的企业级自动化流程， mcp-agent 都试图提供一条从原型到生产的平滑路径。

1.1 核心价值：为什么选择 mcp-agent？

市面上 AI 框架层出不穷， mcp-agent 的独特定位使其在几个关键维度上脱颖而出：

• 真正的 MCP 原生 ：它不是简单支持 MCP，而是完全围绕 MCP 构建。任何符合 MCP 标准的服务器（官方提供的 filesystem、fetch，或社区开发的 Slack、Jira 服务器，甚至你用 FastMCP 快速自建的服务器）都能即插即用，无需编写任何适配器代码。这带来了无与伦比的生态兼容性和未来可扩展性。
• 模式即代码，开箱即用 ：Anthropic 总结的智能体模式（如并行处理、路由、意图分类、规划器-执行器、评估器-优化器）不再是需要你从零实现的论文概念。在 mcp-agent 中，它们被封装成可复用的 AugmentedLLM 类或工厂函数，你可以像搭积木一样组合它们，快速构建复杂的工作流。
• 生产就绪的基石 ：它不仅仅关注“能跑通”。结构化日志、分布式追踪（OpenTelemetry）、细粒度的 Token 计数、安全的密钥管理、以及最重要的——通过 Temporal 实现的持久化执行，这些生产级应用所需的特性都被集成到核心设计中。你可以专注于智能体的业务逻辑，而将可靠性、可观测性交给框架。
• 极简的 Pythonic API ：框架通过装饰器（如 @app.workflow , @app.tool ）和上下文管理器（ async with app.run(): , async with agent: ）将复杂性隐藏起来。你的代码看起来就像普通的异步 Python 代码，而不是在操作一个复杂的“图”或“节点”系统。对于熟悉 asyncio 的开发者来说，学习曲线非常平缓。

接下来，我将带你从零开始，深入 mcp-agent 的每一个核心组件，并通过一个完整的实战项目，展示如何用它构建一个能自动调研、总结并生成报告的智能体。我们会涵盖从环境搭建、配置管理、核心概念到高级模式（如持久化工作流和智能体即服务）的方方面面。

2. 环境准备与项目初始化

工欲善其事，必先利其器。 mcp-agent 推荐使用现代化的 Python 包管理工具 uv ，它能极大地简化依赖管理和虚拟环境创建。当然，传统的 pip 也同样支持。

2.1 安装与项目脚手架

首先，确保你已安装 Python（建议 3.10+）。然后安装 uv （如果尚未安装）：

# 使用 pip 安装 uv
pip install uv

# 或者使用系统的包管理器，例如在 macOS 上
# brew install uv

接下来，最快上手的方式是使用 mcp-agent 自带的 CLI 工具来初始化一个新项目。这个工具会为你创建标准的项目结构、配置文件示例和入口代码。

# 安装 mcp-agent CLI 工具
uvx mcp-agent

# 创建一个新目录并进入
mkdir my-first-mcp-agent && cd my-first-mcp-agent

# 使用 CLI 初始化项目，这里我们选择 ‘basic’ 模板
uvx mcp-agent init --template basic

执行 init 命令后，你会看到类似如下的目录结构被创建：

my-first-mcp-agent/
├── mcp_agent.config.yaml      # 主配置文件
├── mcp_agent.secrets.yaml.example # 密钥配置文件示例
├── main.py                    # 应用入口文件
├── pyproject.toml             # 项目依赖和配置 (由 uv 生成)
└── .gitignore                 # Git 忽略文件

这个 basic 模板已经包含了一个可运行的“查找器”智能体示例，它能够使用 filesystem 和 fetch 这两个 MCP 服务器。让我们先看看它的核心配置。

2.2 配置文件深度解析

mcp-agent 使用 YAML 文件进行配置，清晰地将公开设置与敏感密钥分离。

mcp_agent.config.yaml ：应用核心配置 这个文件定义了你的应用如何运行、连接哪些 MCP 服务器以及日志等行为。

# mcp_agent.config.yaml
execution_engine: asyncio # 执行引擎，可选 ‘asyncio’ 或 ‘temporal’
logger:
  transports: [console]   # 日志输出方式，可选 console（控制台）和 file（文件）
  level: info             # 日志级别：debug, info, warning, error
  # 如果启用 file transport，需指定路径
  # path: “logs/mcp-agent.jsonl”

mcp:
  servers:
    fetch:
      command: “uvx”
      args: [“mcp-server-fetch”] # 使用 uvx 运行 fetch 服务器
    filesystem:
      command: “npx”
      args:
        - “-y”
        - “@modelcontextprotocol/server-filesystem”
        - “.” # 允许访问当前目录，出于安全考虑，请谨慎设置路径
        # 你可以添加多个目录，例如: [“.”, “/path/to/docs”]

openai:
  default_model: gpt-4o # 默认使用的 OpenAI 模型

关键配置点解析：

1. execution_engine : 这是框架的核心设计之一。 asyncio 模式用于本地快速开发和测试，所有代码在单进程中运行。当你需要生产级的持久化、容错和可恢复性时，可以无缝切换到 temporal 模式，而 无需修改任何业务逻辑代码 。
2. mcp.servers : 这里定义了你的智能体可以访问的“能力”。每个服务器对应一个 command 和 args ，框架会负责启动和管理这些子进程的生命周期。 fetch 服务器让智能体能获取网页内容， filesystem 服务器则允许读写本地文件。你可以在这里添加任何其他 MCP 服务器。
3. 路径安全警告 ： filesystem 服务器的路径参数（上面的 . ）决定了智能体可以访问的文件范围。 在生产环境中，务必将其限制在绝对必要的最小目录内 ，避免智能体误操作或恶意指令导致数据损失。

mcp_agent.secrets.yaml ：敏感信息管理 这个文件用于存储所有 API 密钥和敏感配置， 务必将其加入 .gitignore ，不要提交到版本控制系统。

# mcp_agent.secrets.yaml (从 .example 文件复制并重命名)
openai:
  api_key: “sk-...” # 你的 OpenAI API Key
# 其他可能的密钥，例如 Anthropic, Google 等
# anthropic:
#   api_key: “sk-ant-...”

框架也支持从环境变量读取密钥，这在容器化部署中非常有用。你可以在 secrets.yaml 中使用 ${ENV_VAR_NAME} 语法来引用环境变量。

注意 ：安全是智能体应用的重中之重。除了管理好密钥文件，还需仔细审查你赋予智能体的工具权限（尤其是文件系统和网络访问），并在可能的情况下为工具调用添加人工审核环节（ mcp-agent 通过 request_human_input 支持此功能）。

2.3 安装依赖并运行

初始化项目后，使用 uv 安装依赖：

# 使用 uv 同步依赖（会读取 pyproject.toml）
uv sync

# 或者，你也可以手动添加 mcp-agent 及其 OpenAI 支持
uv add “mcp-agent[openai]”

现在，确保你的 mcp_agent.secrets.yaml 文件已正确配置 OpenAI API Key，然后运行示例应用：

uv run main.py

如果一切顺利，你将看到控制台输出智能体启动日志，并最终打印出对 README.md 文件的总结。恭喜，你的第一个 mcp-agent 应用已经跑起来了！这个简单的例子背后，是框架在默默处理 MCP 服务器连接、工具发现、LLM 调用等一系列复杂操作。

3. 核心组件与架构深度剖析

理解了如何运行一个示例后，我们需要深入 mcp-agent 的核心抽象。只有掌握了这些构建块，你才能自由地设计和实现复杂的智能体。整个框架围绕几个关键概念展开： MCPApp 、 Agent 、 AugmentedLLM 和 Workflow 。

3.1 MCPApp：应用的基石与运行时容器

MCPApp 是你的智能体应用的根容器和运行时环境。它负责初始化配置、管理日志和追踪系统、维护 MCP 服务器注册表、并提供执行引擎（asyncio 或 Temporal）的上下文。

from mcp_agent.app import MCPApp

# 创建一个应用实例，给它起个名字
app = MCPApp(name=“research_assistant”)

async def main():
    # ‘app.run()’ 是一个异步上下文管理器。
    # 进入时，它会加载配置、启动日志、初始化 MCP 服务器连接。
    # 离开时，它会优雅地关闭所有连接和资源。
    async with app.run() as running_app:
        # ‘running_app’ 提供了运行时的上下文，例如 logger 和 server_registry
        logger = running_app.logger
        logger.info(“应用启动成功”, data={“name”: app.name})
        # 可以在这里查询当前注册了哪些 MCP 服务器
        servers = list(running_app.context.server_registry.registry.keys())
        logger.info(“可用的 MCP 服务器”, data={“servers”: servers})

为什么需要这个“应用”容器？ 想象一下，如果没有 MCPApp ，你需要手动完成以下工作：解析 YAML 配置、根据配置启动和管理多个 MCP 服务器子进程、处理服务器崩溃和重启、为不同的执行环境（开发/测试/生产）加载不同的密钥、设置全局的日志和监控。 MCPApp 将这些横切关注点（cross-cutting concerns）统一管理起来，让你的业务代码保持干净和专注。

3.2 Agent 与 AgentSpec：智能体的定义与蓝图

Agent 是 mcp-agent 中的核心执行单元。一个 Agent 由两部分核心要素定义：

1. 指令（Instruction） ：一段自然语言描述，告诉 LLM 这个智能体的角色、目标和行为准则。
2. 能力（Capabilities） ：通过 server_names 指定该智能体可以调用的 MCP 服务器（即工具集）。

from mcp_agent.agents.agent import Agent

# 定义一个“研究助理”智能体
research_agent = Agent(
    name=“researcher”, # 智能体名称，用于日志和追踪
    instruction=“””
        你是一个专业的研究助理。你的任务是利用网络搜索和本地文件系统，
        高效、准确地查找、汇总信息，并以结构化的格式呈现。
        在引用任何外部来源时，务必注明出处。
        如果信息不完整或存在矛盾，请明确指出。
        “””,
    server_names=[“fetch”, “filesystem”], # 该智能体可以使用 fetch 和 filesystem 服务器
    # 你还可以通过 ‘functions’ 参数添加自定义的 Python 函数作为工具
)

Agent 的生命周期管理 Agent 本身也是一个异步上下文管理器。在 async with agent: 块内，框架会确保该智能体所需的 MCP 服务器连接已就绪，并为其准备好工具列表。

async with research_agent:
    # 此时，research_agent 已经准备好了它可用的工具
    available_tools = await research_agent.list_tools()
    print(f“智能体 ‘{research_agent.name}’ 可用的工具: {available_tools}”)

AgentSpec ：智能体的“蓝图” 在更复杂的应用中，你可能需要从配置文件（如 YAML）中动态加载或定义大量智能体。 AgentSpec 就是 Agent 的静态定义（或称为“蓝图”），它包含了创建一个 Agent 所需的所有信息，但本身不持有运行时状态。你可以使用工厂函数从文件加载它们。

# agents.yaml
agents:
  - name: researcher
    instruction: “专业的研究助理，擅长信息检索与汇总。”
    server_names: [“fetch”, “filesystem”]
  - name: writer
    instruction: “专业的文案写手，擅长将技术内容转化为易懂的博客文章。”
    server_names: [“filesystem”]
  - name: critic
    instruction: “严格的评审员，负责评估内容的准确性、清晰度和风格。”
    server_names: [] # 这个智能体不需要外部工具，仅依赖 LLM 本身

from pathlib import Path
from mcp_agent.workflows.factory import load_agent_specs_from_file

async with app.run() as running_app:
    # 从 YAML 文件加载智能体蓝图
    specs = load_agent_specs_from_file(
        “agents.yaml”,
        context=running_app.context, # 需要传入运行时上下文以解析配置
    )
    # specs 是一个字典，key 是智能体名，value 是 AgentSpec 对象
    researcher_spec = specs[“researcher”]
    # 你可以根据 spec 在运行时创建 Agent
    researcher_agent = Agent.from_spec(researcher_spec, context=running_app.context)

这种“蓝图”模式非常强大，它允许你将智能体的定义与代码逻辑分离，便于进行 A/B 测试、动态路由（根据输入选择不同的智能体）或构建智能体工作流。

3.3 Augmented LLM：赋能的语言模型

Agent 定义了“谁”和“能做什么”，但真正执行推理和决策的是大语言模型。 AugmentedLLM 是 mcp-agent 对原生 LLM SDK（如 OpenAI, Anthropic）的增强包装。它将一个 Agent 的工具集、记忆（如果配置了）和结构化输出能力“附加”到 LLM 上。

from mcp_agent.workflows.llm.augmented_llm_openai import OpenAIAugmentedLLM
from mcp_agent.workflows.llm.augmented_llm import RequestParams
from pydantic import BaseModel

# 假设我们已经有一个 research_agent
async with research_agent:
    # 将 OpenAI 的 LLM “附加” 到智能体上，并自动注入工具
    llm = await research_agent.attach_llm(OpenAIAugmentedLLM)

    # 方法 1: 生成普通文本
    simple_answer = await llm.generate_str(
        message=“请总结 https://docs.mcp-agent.com 首页的主要内容”,
        request_params=RequestParams(maxTokens=500, temperature=0.7) # 控制生成参数
    )

    # 方法 2: 生成结构化输出 (需要 Pydantic 模型)
    class ResearchSummary(BaseModel):
        topic: str
        key_findings: list[str]
        confidence: float
        source_urls: list[str]

    structured_answer = await llm.generate_structured(
        message=“分析给定的网页，并提取关键发现。”,
        response_model=ResearchSummary # 指定输出的 Pydantic 模型
    )
    # structured_answer 是一个 ResearchSummary 实例，可以直接访问属性
    print(f“主题: {structured_answer.topic}”)

AugmentedLLM 的核心价值：

1. 工具调用自动化 ：你不需要手动编写代码去解析 LLM 的“函数调用”响应，然后执行工具，再将结果送回 LLM。 generate_str 或 generate_structured 内部会自动处理多轮的“LLM 思考 -> 选择工具 -> 执行工具 -> 返回结果 -> 继续思考”的循环，直到 LLM 给出最终答案或达到限制。
2. 结构化输出 ： generate_structured 利用 LLM 的 JSON 模式或函数调用能力，强制输出符合你定义的 Pydantic 模型的数据。这极大地简化了后续的数据处理流程。
3. 统一的接口 ：无论底层是 OpenAI、Anthropic 还是其他提供商， AugmentedLLM 提供了统一的调用接口。切换 LLM 提供商通常只需要更改导入的类和配置中的 API 密钥。

3.4 Workflow 与装饰器：构建复杂、持久化的业务流程

简单的单次问答智能体很有用，但真实世界的业务流程往往是多步骤、有状态、可能长时间运行甚至需要人工干预的。 mcp-agent 通过 Workflow 抽象和一组装饰器来优雅地处理这些场景。

一个 Workflow 是一个有明确输入输出的、可能长时间运行的任务单元。它由多个 Task （任务）组成。框架通过装饰器将普通的异步函数转化为可被持久化执行的工作流任务。

from datetime import timedelta
from mcp_agent.executor.workflow import Workflow, WorkflowResult
from mcp_agent.agents.agent import Agent
from mcp_agent.workflows.llm.augmented_llm_openai import OpenAIAugmentedLLM

# 1. 定义一个工作流类
@app.workflow # 关键装饰器：将这个类注册为工作流
class ResearchAndDraftWorkflow(Workflow[WorkflowResult[str]]):
    # WorkflowResult 是一个泛型包装，这里指定最终输出是 str 类型

    # 2. 定义工作流中的任务
    @app.workflow_task(
        schedule_to_close_timeout=timedelta(minutes=10) # 任务超时设置
    )
    async def conduct_research(self, topic: str) -> str:
        “”“使用研究智能体进行调研，返回原始笔记。”“”
        async with Agent(
            name=“researcher”,
            instruction=“进行深入的网络调研，收集关于给定主题的可靠信息。”,
            server_names=[“fetch”],
        ) as agent:
            llm = await agent.attach_llm(OpenAIAugmentedLLM)
            notes = await llm.generate_str(
                message=f“请调研关于 ‘{topic}’ 的最新发展和关键观点。提供详细的笔记。”
            )
            return notes

    @app.workflow_task
    async def draft_article(self, research_notes: str) -> str:
        “”“使用写作智能体根据调研笔记起草文章。”“”
        async with Agent(
            name=“writer”,
            instruction=“将技术性调研笔记转化为一篇通俗易懂、结构清晰的博客文章草稿。”,
            server_names=[“filesystem”],
        ) as agent:
            llm = await agent.attach_llm(OpenAIAugmentedLLM)
            draft = await llm.generate_str(
                message=f“根据以下调研笔记，撰写一篇博客文章草稿：\n\n{research_notes}”
            )
            return draft

    # 3. 定义工作流的执行入口
    @app.workflow_run
    async def run(self, topic: str) -> WorkflowResult[str]:
        “”“工作流的执行逻辑。这是工作流启动时调用的方法。”“”
        self.logger.info(f“开始处理主题: {topic}”)
        # 顺序执行任务
        notes = await self.conduct_research(topic)
        article_draft = await self.draft_article(notes)

        # 返回结果
        return WorkflowResult(value=article_draft)

装饰器魔法与持久化执行：

• @app.workflow : 将一个类声明为工作流，使其可被框架发现和管理。
• @app.workflow_task : 将类中的一个方法标记为工作流内的一个独立“任务”。任务是可以被持久化记录的最小单元。
• @app.workflow_run : 标记工作流的入口方法。当工作流被触发时，会执行这个方法。

最关键的一点是 ：这些装饰器的代码在 asyncio 和 temporal 两种执行引擎下是 完全一样 的。在 asyncio 模式下，它们就是普通的异步函数调用。一旦你将 execution_engine 配置为 temporal ，框架和 Temporal 后端会自动将这些任务调用转化为持久化的、可恢复的、分布式的活动。这意味着：

• 容错性 ：如果执行任务的进程崩溃，Temporal 会在其他健康的 Worker 上重新调度该任务。
• 可恢复性 ：你可以暂停一个运行中的工作流，稍后再恢复它。
• 长期运行 ：工作流可以运行数小时甚至数天，状态由 Temporal 可靠地保存。
• 信号与人工输入 ：工作流可以在运行时等待外部信号（如人工审批），这通过 request_human_input 实现。

这种“一次编写，随处运行”（从本地开发到分布式生产）的能力，是 mcp-agent 面向生产设计的核心体现。

4. 实战：构建一个自动研究汇报智能体

理论讲得再多，不如动手实践。让我们构建一个相对复杂的智能体应用： 自动研究汇报生成器 。这个智能体会完成以下流程：

1. 接收一个研究主题 。
2. 并行执行 ：同时启动两个子智能体，一个进行网络搜索（使用 fetch ），另一个扫描本地知识库（使用 filesystem 读取指定目录下的 Markdown 文件）。
3. 汇总分析 ：将并行收集到的信息汇总，交给一个“分析员”智能体进行去重、交叉验证和提炼。
4. 生成报告 ：根据分析结果，生成一份结构化的 Markdown 格式报告。
5. 保存与通知 ：将报告保存到本地文件，并（模拟）发送一个完成通知。

我们将使用 mcp-agent 的 Parallel (Map-Reduce) 模式来实现第 2 步。

4.1 项目结构与配置

首先，创建项目目录和文件：

research_agent_project/
├── mcp_agent.config.yaml
├── mcp_agent.secrets.yaml
├── research_workflow.py
├── knowledge_base/          # 本地知识库目录
│   └── some_notes.md
└── outputs/                 # 报告输出目录

mcp_agent.config.yaml :

execution_engine: asyncio
logger:
  transports: [console, file]
  level: info
  path: “logs/research_agent.jsonl”

mcp:
  servers:
    fetch:
      command: “uvx”
      args: [“mcp-server-fetch”]
    filesystem:
      command: “npx”
      args:
        - “-y”
        - “@modelcontextprotocol/server-filesystem”
        - “./knowledge_base” # 只允许访问知识库目录，更安全
        - “./outputs”        # 允许写入输出目录

openai:
  default_model: gpt-4o-mini # 使用成本更低的模型进行演示

mcp_agent.secrets.yaml : (填入你的 OpenAI API Key)

4.2 实现并行研究（Map-Reduce）模式

mcp-agent 提供了 create_parallel_llm 工厂函数，可以轻松创建并行处理工作流。它遵循“Map-Reduce”模式：将输入分发（Map）给多个并行的专家智能体处理，然后将它们的结果聚合（Reduce）起来。

# research_workflow.py
import asyncio
from pathlib import Path
from datetime import datetime
from mcp_agent.app import MCPApp
from mcp_agent.agents.agent import Agent
from mcp_agent.workflows.llm.augmented_llm_openai import OpenAIAugmentedLLM
from mcp_agent.workflows.factory import create_parallel_llm
from mcp_agent.executor.workflow import Workflow, WorkflowResult

app = MCPApp(name=“auto_research_agent”)

# 定义两个专家智能体的蓝图（在真实场景中，可以从 YAML 加载）
WEB_RESEARCHER_SPEC = {
    “name”: “web_researcher”,
    “instruction”: “””
        你是一个专业的网络信息研究员。你的任务是使用 fetch 工具访问给定的 URL 或进行网络搜索，
        提取关键、准确、最新的信息。优先使用权威来源（如官方文档、知名技术博客、学术网站）。
        在回复中，请清晰标注信息来源的 URL。
        “””,
    “server_names”: [“fetch”],
}

LOCAL_RESEARCHER_SPEC = {
    “name”: “local_researcher”,
    “instruction”: “””
        你是一个本地知识库分析员。你的任务是仔细阅读本地文件系统（Markdown、文本文件）中的内容，
        提取与给定主题相关的信息、观点和结论。注意文件之间的关联性。
        “””,
    “server_names”: [“filesystem”],
}

@app.workflow
class ParallelResearchWorkflow(Workflow[WorkflowResult[str]]):
    “”“并行研究工作流”“”

    @app.workflow_task
    async def parallel_research(self, topic: str) -> str:
        “”“并行执行网络研究和本地研究，然后汇总结果。”“”
        async with app.run() as running_app:
            # 1. 根据蓝图创建两个 Agent 实例
            web_agent = Agent.from_spec(WEB_RESEARCHER_SPEC, context=running_app.context)
            local_agent = Agent.from_spec(LOCAL_RESEARCHER_SPEC, context=running_app.context)

            # 2. 使用工厂函数创建并行 LLM
            # 它接受一个 Agent 列表和一个 Reduce 智能体（负责汇总）
            parallel_llm = await create_parallel_llm(
                agents=[web_agent, local_agent], # Map 阶段的专家们
                reducer_instruction=“””
                    你是一个高级研究分析师。现在你有两份关于同一主题的调研材料：
                    一份来自网络搜索，另一份来自本地知识库。
                    你的任务是将这两份材料合并，去重，消除矛盾，提炼出最核心、最可靠的发现，
                    并组织成一份连贯的综合性研究笔记。如果发现信息冲突，以网络权威来源为准，并备注说明。
                    “””,
                provider=“openai”,
                context=running_app.context,
            )

            # 3. 执行并行研究。parallel_llm 会同时驱动两个专家智能体工作，
            #    然后将它们的结果交给 Reduce 智能体进行汇总。
            research_notes = await parallel_llm.generate_str(
                message=f“请围绕 ‘{topic}’ 这个主题进行深入研究。”
            )
            return research_notes

    @app.workflow_task
    async def generate_report(self, consolidated_notes: str) -> str:
        “”“根据研究笔记生成最终报告。”“”
        async with Agent(
            name=“report_writer”,
            instruction=“””
                你是一名技术文档工程师。根据提供的研究笔记，撰写一份结构清晰、专业、易于理解的 Markdown 格式报告。
                报告应包含：摘要、核心发现、详细分析、结论与建议、参考文献（如果提供了来源）。
                语言风格保持客观、严谨。
                “””,
            server_names=[“filesystem”], # 需要写入文件
        ) as agent:
            llm = await agent.attach_llm(OpenAIAugmentedLLM)
            report = await llm.generate_str(
                message=f“请基于以下研究笔记，生成一份完整的报告：\n\n{consolidated_notes}”
            )
            return report

    @app.workflow_task
    async def save_report(self, report_content: str, topic: str) -> str:
        “”“将报告保存到本地文件。”“”
        # 使用 filesystem 服务器的工具来写文件
        # 注意：这里我们直接通过 app 的上下文获取 MCP 客户端来调用工具，更底层一些。
        # 另一种方式是在 Agent 内让 LLM 去调用 ‘write_file’ 工具。
        async with app.run() as running_app:
            from mcp_agent.mcp.gen_client import gen_client
            async with gen_client(“filesystem”, running_app.context.server_registry, context=running_app.context) as client:
                # 生成文件名
                timestamp = datetime.now().strftime(“%Y%m%d_%H%M%S”)
                safe_topic = “”.join(c for c in topic if c.isalnum() or c in (‘ ‘, ‘-’, ‘_’)).rstrip()
                filename = f“outputs/research_report_{safe_topic}_{timestamp}.md”
                # 调用 filesystem 服务器的 write_file 工具
                # 注意：工具调用参数需要符合 MCP 工具定义，通常是 JSON 对象。
                # 这里假设 filesystem 服务器提供了 write_file 工具，其参数为 {“path”: “…”, “content”: “…”}
                # 实际调用前，最好先用 client.list_tools() 确认工具名和参数格式。
                try:
                    # 这是一个示例，实际工具名和参数可能不同
                    result = await client.call_tool({
                        “name”: “write_file”,
                        “arguments”: {
                            “path”: filename,
                            “content”: report_content
                        }
                    })
                    self.logger.info(f“报告已保存至: {filename}”)
                    return filename
                except Exception as e:
                    self.logger.error(f“保存报告失败: {e}”)
                    raise

    @app.workflow_run
    async def run(self, topic: str) -> WorkflowResult[str]:
        “”“工作流主逻辑”“”
        self.logger.info(f“🚀 开始自动研究流程，主题: {topic}”)

        # 步骤 1: 并行研究
        self.logger.info(“🔍 阶段1: 并行研究 (网络 + 本地)”)
        research_notes = await self.parallel_research(topic)
        self.logger.info(“✅ 并行研究完成，开始生成报告。”)

        # 步骤 2: 生成报告
        self.logger.info(“📝 阶段2: 生成报告”)
        final_report = await self.generate_report(research_notes)

        # 步骤 3: 保存报告
        self.logger.info(“💾 阶段3: 保存报告”)
        saved_path = await self.save_report(final_report, topic)

        # 步骤 4: (模拟) 发送通知
        self.logger.info(f“📬 研究完成！报告已保存至: {saved_path}”)
        # 在实际应用中，这里可以集成邮件、Slack、Webhook 等通知方式
        # 例如，可以调用一个自定义的 ‘notify’ 工具。

        return WorkflowResult(value=saved_path)

async def main():
    “”“启动工作流”“”
    # 在实际应用中，topic 可能来自用户输入、API 调用或队列
    topic = “大语言模型在代码生成中的应用与挑战”
    workflow_instance = ParallelResearchWorkflow()
    result = await workflow_instance.run(topic)
    print(f“工作流执行完毕。报告路径: {result.value}”)

if __name__ == “__main__”:
    asyncio.run(main())

代码解读与避坑指南：

1. create_parallel_llm 的使用 ：这是实现并行模式的关键。你提供一组 Agent （执行 Map）和一个 reducer_instruction （定义 Reduce 阶段智能体的角色）。框架会并行运行各个 Map 智能体，然后将它们的所有输出和原始用户查询一起交给 Reduce 智能体进行最终处理。
2. 直接 MCP 工具调用 ：在 save_report 任务中，我们绕过了 Agent 和 AugmentedLLM ，直接通过 gen_client 获取 MCP 客户端并调用工具。这适用于那些不需要 LLM 推理的、确定性的操作。 务必提前检查工具的确切名称和参数格式 ，可以通过 await client.list_tools() 查看。
3. 错误处理 ：生产代码中，每个 await 调用都应该有相应的 try...except 块，并考虑重试逻辑。 mcp-agent 与 Temporal 集成后，任务级别的重试和超时可以通过装饰器参数配置。
4. 资源清理 ：示例中为了简洁，在多个任务内部都使用了 async with app.run() 。在实际工作流中，更好的做法是在工作流 run 方法开始时启动一次 app ，然后将其上下文传递给各个任务方法使用，避免重复初始化。

运行这个工作流，你将看到日志输出清晰地展示了各个阶段的执行过程，最终在 outputs/ 目录下生成一份 Markdown 报告。这个例子展示了如何将简单的智能体组合成强大的并行处理流水线。

5. 高级特性与生产化考量

当你掌握了基础组件和模式后，就可以探索 mcp-agent 的高级功能，将它们应用到生产环境中。

5.1 可观测性：日志、监控与 Token 计数

对于生产系统，知道“发生了什么”至关重要。 mcp-agent 内置了结构化日志和 OpenTelemetry 支持。

结构化日志 ：配置中的 logger.transports: [console, file] 会将 JSON 格式的日志同时输出到控制台和文件。每一条日志都包含时间戳、级别、消息、以及丰富的上下文数据（如 agent_name , workflow_id , tool_call_id ），非常适合用 ELK、Loki 等日志系统进行聚合和分析。

Token 计数与成本监控 ：LLM API 调用是主要成本来源。 mcp-agent 在应用上下文中提供了一个 token_counter 。

async with app.run() as running_app:
    token_counter = running_app.context.token_counter

    # 定义一个监控回调函数
    async def on_token_update(node, usage):
        # node 可以是 ‘llm’, ‘agent’, ‘workflow’ 等
        # usage 包含 prompt_tokens, completion_tokens, total_tokens
        if usage.total_tokens > 10000:
            print(f“⚠️ 警告: 节点 ‘{node.name}’ 已消耗 {usage.total_tokens} tokens，可能超预算！”)

    # 注册一个监控器，关注所有 LLM 节点，阈值设为 1
    watch_id = await token_counter.watch(
        callback=on_token_update,
        node_type=“llm”,
        threshold=1, # 每次更新都触发回调
        include_subtree=True, # 包含子节点的消耗
    )

    # … 运行你的智能体 …

    # 获取总消耗
    total_usage = token_counter.get_usage()
    print(f“本次会话总 Token 消耗: {total_usage.total_tokens}”)
    print(f“  其中 Prompt: {total_usage.prompt_tokens}, Completion: {total_usage.completion_tokens}”)

    # 取消监控
    await token_counter.unwatch(watch_id)

这个功能对于实现预算控制、成本告警和用量分析非常有用。

5.2 人工介入（Human-in-the-loop）

并非所有决策都应该完全自动化。 mcp-agent 通过 request_human_input 支持工作流在特定节点暂停，等待外部输入（如人工审批、额外数据）。

from mcp_agent.human_input.types import HumanInputRequest

@app.workflow
class ApprovalWorkflow(Workflow[WorkflowResult[str]]):
    @app.workflow_task
    async def draft_proposal(self, topic: str) -> str:
        # … 生成提案草案 …
        return draft

    @app.workflow_task
    async def wait_for_approval(self, draft: str) -> bool:
        “”“暂停工作流，等待人工审批”“”
        request = HumanInputRequest(
            prompt=f“请审批以下提案草案：\n\n{draft}\n\n是否批准？(输入 ‘approve’ 或 ‘reject’)”,
            required=True,
            metadata={
                “workflow_id”: self.context.workflow_id,
                “draft_topic”: “某项目提案”
            }
        )
        # 这会使工作流在此处暂停，直到收到外部信号
        response = await self.context.request_human_input(request)
        # response.content 是人工输入的内容，例如 “approve”
        return response.content.strip().lower() == “approve”

    @app.workflow_run
    async def run(self, topic: str) -> WorkflowResult[str]:
        draft = await self.draft_proposal(topic)
        approved = await self.wait_for_approval(draft)
        if approved:
            return WorkflowResult(value=“提案已批准并进入下一阶段。”)
        else:
            return WorkflowResult(value=“提案被驳回。”)

在 Temporal 模式下，工作流会持久化地等待。你可以通过 mcp-agent CLI 或 Cloud API 向等待中的工作流发送信号来恢复它：

uvx mcp-agent cloud workflows resume <workflow_id> --payload ‘{“content”: “approve”}’

5.3 将智能体暴露为 MCP 服务器

mcp-agent 最强大的特性之一是双向性：你不仅可以消费 MCP 服务器，还可以将你自己的 MCPApp （包括其所有的工具和工作流） 暴露为一个标准的 MCP 服务器 。这意味着你的智能体应用可以被 Claude Desktop、Cursor、或其他任何支持 MCP 的客户端直接调用。

# agent_as_server.py
from mcp_agent.app import MCPApp
from mcp_agent.server import create_mcp_server_for_app
from pydantic import BaseModel

app = MCPApp(name=“my_tool_provider”)

# 定义一些工具，它们会自动通过 MCP 暴露
@app.tool
async def calculate_bmi(weight_kg: float, height_m: float) -> float:
    “”“计算身体质量指数 (BMI).”“”
    if height_m <= 0:
        raise ValueError(“身高必须为正数”)
    return round(weight_kg / (height_m ** 2), 2)

@app.tool
async def generate_haiku(topic: str) -> str:
    “”“根据给定主题生成一首俳句。”“”
    # 这里可以集成一个 LLM 调用
    return f“关于 {topic} 的俳句：\n古池蛙跃入，\n水之音。”

# 你甚至可以暴露整个工作流作为一个“提示”（Prompt）或复杂的工具
@app.prompt
def research_prompt(topic: str) -> str:
    return f“””
        你是一个研究助手。请深入研究以下主题，并提供一份简洁的报告。
        主题：{topic}
        报告需包含：概述、关键点、潜在应用。
        “””

if __name__ == “__main__”:
    # 创建 MCP 服务器并运行在标准输入/输出上（这是 MCP 服务器的标准运行方式）
    server = create_mcp_server_for_app(app)
    server.run_stdio()
    # 现在，任何 MCP 客户端（如 Claude Desktop）都可以连接到这个进程，
    # 并看到 calculate_bmi 和 generate_haiku 这两个工具。

运行这个脚本，它就变成了一个 MCP 服务器。在 Claude Desktop 的设置中，你可以添加一个 “Local MCP Server”，指定这个 Python 脚本的路径，Claude 就能直接调用 calculate_bmi 和 generate_haiku 工具了。这为构建可交互的、嵌入到现有工作流中的 AI 工具打开了大门。

5.4 部署到云端

对于生产部署， mcp-agent 提供了云平台（目前为 Beta 版）和基于 Temporal 的自托管方案。

使用 mcp-agent Cloud (Beta) ： 这是最快的方式，提供托管的 Temporal 集群、密钥管理、HTTPS 终结点和监控仪表板。

# 1. 登录
uvx mcp-agent login
# 2. 初始化并部署你的应用
uvx mcp-agent init --template basic --dir my-cloud-agent
cd my-cloud-agent
# 编辑配置和密钥后…
uvx mcp-agent deploy my-cloud-agent
# 3. 管理你的应用
uvx mcp-agent cloud apps list
uvx mcp-agent cloud workflows list <app_name>

自托管 Temporal ： 对于需要完全控制权的团队，可以自行部署 Temporal 集群。 mcp-agent 提供了与 Temporal 无缝集成的 Worker 实现。

# 将配置改为 temporal
execution_engine: temporal
temporal:
  namespace: “your-namespace”
  task_queue: “mcp-agent-queue”
  # … 其他 Temporal 连接配置

然后，你需要运行一个 Temporal Worker 来执行工作流：

# worker.py
import asyncio
from mcp_agent.app import MCPApp
from mcp_agent.executor.temporal import create_temporal_worker_for_app

app = MCPApp(name=“production_research_agent”)

async def main():
    async with create_temporal_worker_for_app(app) as worker:
        # Worker 会持续运行，监听 task_queue 并执行工作流
        await worker.run()

if __name__ == “__main__”:
    asyncio.run(main())

运行这个 Worker，然后通过 Temporal 客户端或 mcp-agent 的 Cloud API 来触发工作流。Temporal 会负责工作流的状态持久化、任务分发、重试和超时。

6. 常见问题与排查技巧

在实际使用 mcp-agent 的过程中，你可能会遇到一些典型问题。以下是一些排查思路和解决方案。

6.1 MCP 服务器连接失败

症状 ：应用启动时报错，提示无法连接或启动某个 MCP 服务器（如 fetch , filesystem ）。

排查步骤 ：

1. 检查命令是否存在 ： mcp-agent 通过 command 和 args 启动服务器。确保 uvx 、 npx 等命令在系统的 PATH 中。对于 npx ，可能需要安装 Node.js。
2. 检查服务器包 ：确认所需的 MCP 服务器包已全局安装或可用。例如，对于 fetch ，可能需要运行 uvx mcp-server-fetch 看是否能独立启动。对于 filesystem ，运行 npx -y @modelcontextprotocol/server-filesystem . 测试。
3. 查看详细日志 ：将 logger.level 设置为 debug ，重新运行应用。调试日志会显示框架尝试启动服务器的完整命令和任何错误输出。
4. 权限问题 ：确保应用有权限执行指定的命令和访问指定的文件路径（对于 filesystem 服务器）。

6.2 LLM 调用超时或无响应

症状 ：智能体卡住，长时间没有输出，最终可能超时。

排查步骤 ：

1. 检查 API 密钥和网络 ：确认 mcp_agent.secrets.yaml 中的 API 密钥正确，且网络可以访问对应的 LLM 服务商（如 OpenAI）。
2. 检查模型名称 ：确认配置中 default_model 是有效的模型名（如 gpt-4o , gpt-4o-mini ）。
3. 降低复杂度或添加超时 ：如果任务非常复杂，LLM 可能需要更长时间思考。可以在 RequestParams 中设置 timeout ，或者在 @app.workflow_task 装饰器中设置更长的 schedule_to_close_timeout 。
4. 启用工具调用追踪 ：在调试阶段，可以增加日志级别，观察 LLM 与工具之间的交互过程，看是否卡在某个特定的工具调用上。

6.3 工作流在 Temporal 模式下不执行

症状 ：切换到 temporal 引擎后，触发工作流但没有反应。

排查步骤 ：

1. 检查 Temporal 集群连接 ：确认 mcp_agent.config.yaml 中的 Temporal 配置（地址、命名空间）正确，且网络可达。
2. 检查 Worker 是否在运行 ：Temporal 工作流需要有一个或多个 Worker 进程在监听对应的 task_queue 。确保你的 Worker 脚本（ create_temporal_worker_for_app ）正在运行，并且没有报错。
3. 检查工作流和任务队列名称 ：触发工作流的客户端（或 CLI）使用的 task_queue 必须与 Worker 监听的 task_queue 一致。
4. 查看 Temporal Web UI ：如果部署了 Temporal Web UI，可以在那里查看工作流执行历史、错误信息和任务队列状态，这是最强大的调试工具。

6.4 智能体行为不符合预期

症状 ：智能体没有调用预期的工具，或者给出了奇怪的回答。

排查步骤 ：

1. 审查指令（Instruction） ：智能体的行为很大程度上由 instruction 决定。确保指令清晰、明确地说明了角色、目标和约束。例如，如果不想让智能体随意写文件，可以在指令中强调“未经明确指示，不得创建或修改任何文件”。
2. 检查可用工具列表 ：在 async with agent: 块内，使用 await agent.list_tools() 打印出智能体实际可用的工具列表，确保 server_names 配置正确，且 MCP 服务器提供了预期的工具。
3. 分析交互过程 ：将日志级别调到 debug ，观察 LLM 收到的提示（prompt）、它产生的工具调用请求以及工具的返回结果。这能帮你判断是 LLM 理解有误，还是工具返回的数据有问题。
4. 迭代优化 ：智能体开发是一个迭代过程。根据观察到的失败案例，不断调整 instruction 、工具配置甚至任务拆分的粒度。

6.5 性能优化建议

1. 连接复用 ：在可能的情况下，复用 MCPApp 和 Agent 实例。避免在循环或频繁调用的函数中反复创建和销毁它们，因为建立 MCP 服务器连接有开销。
2. 异步并发 ：利用 asyncio.gather 来并发执行多个独立的 LLM 调用或工具调用，只要它们之间没有依赖关系。 create_parallel_llm 内部就是这样做的。
3. 模型选择 ：对于不需要极高创造力的任务（如信息提取、总结），使用更小、更快的模型（如 gpt-4o-mini ）可以显著降低成本并提高速度。
4. 缓存 ：对于频繁查询且不常变化的数据（如某些文件内容、API 响应），可以考虑在工具层或应用层添加缓存机制，减少对 LLM 和外部服务的重复调用。
5. 监控与告警 ：务必配置好 Token 使用监控和错误告警。利用 token_counter.watch 设置预算阈值，并结合外部监控系统（如 Prometheus + Alertmanager）或日志告警，在出现异常时及时通知。

mcp-agent 作为一个正在快速发展的框架，其生态和最佳实践也在不断丰富。最有效的学习方式是多参考官方文档中的 Examples 目录，那里有从基础到高级的大量可运行示例，涵盖了本文提到的绝大多数模式和应用场景。遇到问题时，也可以在项目的 Discord 社区寻求帮助。