构建生产级 AI 代理（使用 MCP 与 A2A）：来自实战的指南

Source: Dev.to

TL;DR 在本文中，我将分享我从脆弱、定制的 AI 代理架构转向使用 模型上下文协议（Model Context Protocol，MCP） 的稳健、标准化方法的历程。我会带你了解：

• 为什么 代理对代理（Agent‑to‑Agent，A2A） 通信是生产系统中缺失的环节。
• 我如何使用标准化合约设计 每日会议纪要助理。
• 我构建的完整代码基础设施（以及你如何也能做到）。
• 为什么我认为标准化 上下文 比标准化 提示词 更重要。

如果你已经厌倦了调试为何你的代理会产生幻觉式的函数调用，这篇文章适合你。

从混沌到契约：我如何驯服代理人的狂野西部

---

TL;DR

在本文中，我分享了从脆弱、定制的 AI 代理架构到使用 Model Context Protocol (MCP) 的稳健、标准化方法的旅程。我将为你讲解：

• 为什么 Agent‑to‑Agent (A2A) 通信是生产系统中缺失的环节。
• 我如何使用标准化合同设计 每日会议纪要助理。
• 我构建的完整代码基础设施（以及你如何也能做到）。
• 为什么我认为标准化 上下文 比标准化 提示词 更重要。

如果你厌倦了调试代理为何产生了幻觉式的函数调用，这篇文章适合你。

介绍

我仍然记得那些在调试我的第一个复杂多代理系统时熬的深夜。我有一个 Research Agent（研究代理），它应该与 Writer Agent（写作代理）对话。它在我的 Jupyter notebook 中运行得非常顺畅，但一旦部署…就一团糟。

• Research Agent 输出 JSON；Writer Agent 期待 Markdown。
• “Memory” 模块是一个全局字典，始终被覆盖。

这像是一座纸牌屋。

根据我的经验，这正是当今大多数 AI 工程停滞的地方。我们构建了令人印象深刻的演示，但生产可靠性却难以实现，因为我们缺乏一个根本的 protocol（协议）来进行通信。

随后我发现了像 MCP（Model Context Protocol） 这样的通用协议。我意识到问题不在于我的提示工程——而在于我的架构。我不需要更聪明的模型；我需要更好的契约。依我之见，采用严格的协议就是玩具与工具之间的区别。

这篇文章的内容

这不是一篇关于“AI 未来”的高层次空洞文章。它是一篇 脚踏实地、代码密集的实战指南，展示我如何构建一个生产级的 Agent‑to‑Agent（A2A）系统。

我将向你展示我是如何打造 每日会议纪要助手——一个能够：

1. 连接我的日历。
2. 拉取会议转录文本。
3. 对其进行摘要。
4. 将行动项通过邮件发送给我的系统。

“魔法”并不在于摘要本身，而在于 底层实现。

---

技术栈

我选择这套技术栈是因为我更看重可靠性而非炒作：

---

为什么要阅读？

如果你曾经感受到以下痛苦：

• 为每个新工具编写自定义 API 包装器。
• 代理因不知道何时停止而陷入循环。
• 尝试将本地专用代理连接到基于云的 LLM。

…那么我构建的这个实验性概念验证（PoC）将直击你的内心。我写它是因为我希望六个月前有人向我展示过这种模式。

---

让我们设计

在我写下任何代码之前，我先退后一步设计交互。我想：“如果这些代理是员工，他们会如何传递文件？”

在我看来，一个代理需要三件事：

1. Tools – 它可以做的事情（搜索网络，发送电子邮件）。
2. Resources – 它可以读取的东西（日历，日志）。
3. Prompts – 用于请求事物的标准化方式。

我绘制了以下（简化的）流程：

Client  →  FastMCP Server (exposes tools/resources)  →  LLM

我这样设计是因为我希望 SummaryServer 完全不依赖 ResearchServer。将它们解耦后，我可以在以后更换研究引擎，而不会破坏日历集成。

---

让我们开始烹饪

下面是实际实现。我将项目结构化为 Server（提供功能）和 Client（消费功能）两部分。

第 1 步 – FastMCP 服务器

import os
from mcp.server.fastmcp import FastMCP

# 初始化 FastMCP 服务器。
# 在我看来，给服务器起一个清晰的名字对多代理调试至关重要。
mcp = FastMCP("DailyAssistant")

@mcp.tool()
async def search_web(query: str, limit: int = 5) -> str:
    """
    为给定查询在网络上搜索。

    参数:
        query: 搜索关键词。
        limit: 返回的最大结果数。
    """
    # 在真实部署中，这里会调用 Tavily、Serper 等。
    # 为了演示概念，我们模拟返回，以专注于协议本身。
    return (
        f"Mock search results for '{query}':\n"
        "1. Result A\n"
        "2. Result B"
    )

@mcp.resource("config://app_settings")
def get_app_settings() -> str:
    """获取应用配置设置。"""
    return "Theme: Dark\nNotifications: Enabled"

这段代码的作用

• 定义了一个提供 search_web 工具 和 config://app_settings 资源 的服务器。

我这样组织的原因

• 装饰器（@mcp.tool()）让定义紧贴实现，易于维护。
• 将工具写在单独的 JSON 文件中常常会导致实现改动而模式未同步的偏差。

我的收获

• 类型提示（query: str）并非装饰用。MCP 会利用它们生成 LLM 最终看到的 JSON 模式。如果这里的类型写得马虎，后续代理会真的感到困惑。

---

第 2 步 – 代理客户端

import asyncio
import sys
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run_client():
    # 我决定使用 Stdio 进行本地通信。
    # 对于 side‑car 模式，它默认更快且更安全。
    server_params = StdioServerParameters(
        command=sys.executable,
        args=["src/server/agent_server.py"],
        env=None,
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 示例：让服务器执行一次网络搜索。
            result = await session.call_tool(
                "search_web",
                {"query": "MCP protocol overview", "limit": 3},
            )
            print("Search result:", result)

if __name__ == "__main__":
    asyncio.run(run_client())

这段代码的作用

• 启动一个本地 stdio 子进程来运行 agent_server.py。
• 使用 ClientSession 调用 search_web 工具，而无需关心其内部实现。

为何重要

• 客户端只需要 契约（工具名称 + JSON 模式）。
• 更换服务器实现（例如迁移到远程 FastAPI 端点）只需修改 StdioServerParameters 即可。

---

第 3 步 – 将所有部件连起来（FastAPI 包装器）

如果你更倾向于使用基于 HTTP 的服务器而不是 stdio，完全相同的装饰器也可配合 FastAPI 使用：

from fastapi import FastAPI
from mcp.server.fastapi import FastAPIMCP

app = FastAPI()
mcp = FastAPIMCP(app, "DailyAssistant")

@mcp.tool()
async def send_email(to: str, subject: str, body: str) -> str:
    """
    模拟邮件发送器。
    """
    # 在生产环境中，你会接入 SMTP 服务或 SendGrid。
    return f"Email sent to {to} with subject '{subject}'."

运行 uvicorn this_module:app --reload 即可通过 MCP 协议以 HTTP 方式公开 send_email 工具。

---

经验教训

---

TL;DR (Revisited)

• 使用 MCP 通过 Python 装饰器来定义 tools 和 resources。
• 保持 agents 简洁：它们只需要了解 what 可以调用，而不必关心 how 实现。
• 解耦服务 → 在不破坏契约的前提下替换实现。

最后思考

通过像 MCP 这样的协议对 context 进行标准化，使我那不稳定的演示转变为可投入生产的系统。为干净的契约所付出的努力收获颇丰：错误更少，调试更简便，并且能够独立地扩展各个代理。

如果你正在构建超出单一代理概念验证的任何项目，建议尝试 MCP。你的未来自我（以及运维团队）一定会感激不已。

Session Example

async with MCPClientSession(read, write) as session:
    await session.initialize()

    # Dynamic Discovery
    tools = await session.list_tools()
    print(f"Connected! Found tools: {[t.name for t in tools.tools]}")

    # Execution
    result = await session.call_tool(
        "search_web",
        arguments={"query": "MCP adoption"}
    )
    print(f"Tool Output: {result.content[0].text}")

What This Does

它会将服务器作为子进程启动，并通过标准输入/输出进行连接。随后会动态询问 “你能做什么？”（list_tools），再让它执行具体操作。

My Experience Here

我最初尝试用 HTTP 来完成所有操作，但对于本地代理——比如在笔记本上运行的编码助手——stdio 的表现要好得多。它没有网络开销，并且简化了认证流程（只要能运行进程，就拥有访问权限）。

---

让我们开始

如果你想自己运行此 PoC，我已经把它简化到极致。

前置条件

• Python 3.10+
• 虚拟环境（始终使用 venv！）

克隆仓库

（下面提供了公共仓库的链接。）

安装依赖

pip install mcp httpx

验证安装

python -c "import mcp; print(mcp.__version__)"

---

让我们运行

运行系统非常简单。

python src/client/agent_client.py

您应该会看到输出，表明握手成功，随后显示模拟搜索结果。

需要注意的事项

如果看到通用的 ConnectionRefused 或管道错误，通常是因为服务器脚本在启动时崩溃（例如缺少导入），导致握手无法完成。请务必先单独验证服务器能够正常运行！

结束语

构建这个实验性的 “每日分钟助理” 让我认识到，AI 的未来不仅仅是更大的上下文窗口——更在于 结构化上下文。

在我看来，我们正从 “提示工程” 迈向 上下文工程。MCP 方法让我们能够把工具和资源视为一等公民，弥合华丽演示与可靠生产软件之间的鸿沟。

希望本指南能为你省去我曾经遇到的一些头疼事。代码已开源，尽情 fork、改造，并告诉我你构建了什么。

Tags: ai, python, mcp, agents

---

免责声明

此处表达的观点和意见仅代表我个人，并不代表我的雇主或我所隶属的任何组织的立场、观点或意见。内容基于我的个人经验和实验，可能不完整或不准确。任何错误或误解均非本意，如有陈述被误解或曲解，我在此提前致歉。