前言：为什么 Agent 需要一套"通信协议"？

2026年，AI Agent 已经不是什么新鲜词了。但真正动手做过 Agent 系统的人都有一个共同感受：单个 Agent 能力有限，多个 Agent 协作才是生产级应用的正确姿势。

问题来了——多个 Agent 之间怎么通信？怎么共享上下文？怎么避免每个 Agent 都在"自说自话"？

这就是 MCP（Model Context Protocol）要解决的问题。Anthropic 在 2024 年底开源了这个协议，到 2026 年已经成了 Agent 生态的事实标准。今天我们就从零开始，搭建一个基于 MCP 的多 Agent 协作系统。

一、MCP 到底是什么？

一句话概括：MCP 是 AI Agent 世界的 USB 接口。

以前你接个鼠标、接个键盘、接个U盘，每个设备都有自己的驱动程序。有了 USB 之后，插上就能用。MCP 对 Agent 做的事情完全一样——它定义了一套标准协议，让 Agent 可以即插即用地连接各种工具、数据源和服务。

从技术架构上看，MCP 采用的是 Client-Server 模型：

• MCP Host：发起请求的一方（比如你的 AI 应用）
• MCP Client：协议客户端，负责与 Server 通信
• MCP Server：暴露工具和资源的服务端

核心交互流程很简单：

1. Client 向 Server 发送 initialize 请求，协商协议版本和能力
2. Client 发送 tools/list 获取 Server 支持的工具列表
3. Client 发送 tools/call 调用具体工具
4. Server 返回执行结果

二、动手写第一个 MCP Server

废话不多说，直接上代码。我们用 Python 搭建一个简单的 MCP Server，它能查询数据库中的用户信息。

2.1 环境准备

pip install mcp sqlite3

2.2 Server 代码

import sqlite3
from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("user-query-server")

# 初始化数据库
def get_db():
    conn = sqlite3.connect("users.db")
    conn.row_factory = sqlite3.Row
    return conn

@app.list_tools()
async def list_tools():
    """声明这个 Server 提供哪些工具"""
    return [
        Tool(
            name="query_user",
            description="根据用户名查询用户信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "username": {
                        "type": "string",
                        "description": "要查询的用户名"
                    }
                },
                "required": ["username"]
            }
        ),
        Tool(
            name="list_users",
            description="列出所有用户",
            inputSchema={
                "type": "object",
                "properties": {
                    "limit": {
                        "type": "integer",
                        "description": "返回数量限制",
                        "default": 10
                    }
                }
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    """处理工具调用"""
    db = get_db()

    if name == "query_user":
        username = arguments.get("username", "")
        cursor = db.execute(
            "SELECT * FROM users WHERE name LIKE ?",
            (f"%{username}%",)
        )
        rows = cursor.fetchall()
        if rows:
            results = [dict(row) for row in rows]
            return [TextContent(type="text", text=str(results))]
        return [TextContent(type="text", text=f"未找到用户: {username}")]

    elif name == "list_users":
        limit = arguments.get("limit", 10)
        cursor = db.execute("SELECT * FROM users LIMIT ?", (limit,))
        rows = cursor.fetchall()
        results = [dict(row) for row in rows]
        return [TextContent(type="text", text=str(results))]

    raise ValueError(f"未知工具: {name}")

# 启动 Server
if __name__ == "__main__":
    import asyncio
    from mcp.server.stdio import stdio_server

    async def main():
        async with stdio_server() as (read_stream, write_stream):
            await app.run(read_stream, write_stream, app.create_initialization_options())

    asyncio.run(main())

就这么简单。30 行核心代码，你就拥有了一个标准的 MCP Server。任何支持 MCP 协议的 Client 都能直接调用它。

三、多 Agent 协作架构设计

单个 Agent 能做的事情有限。真正的生产级系统，通常需要多个 Agent 各司其职：

Agent 角色	职责	典型工具
Planner	任务拆解和规划	思维链推理
Coder	代码生成和修改	文件读写、代码执行
Researcher	信息检索和分析	网页搜索、文档查询
Reviewer	代码审查和质量检查	静态分析、测试执行
Executor	任务执行和部署	Shell 命令、API 调用

3.1 通信模式

多 Agent 之间的通信有两种主流模式：

模式一：中心化调度（Orchestrator）

┌─────────────┐
│  Orchestrator │
└──┬──┬──┬──┬──┘
   │  │  │  │
   ▼  ▼  ▼  ▼
  A1 A2 A3 A4

Orchestrator 负责任务分配和结果汇总。优点是逻辑清晰，缺点是单点瓶颈。

模式二：去中心化协作（Peer-to-Peer）

  A1 ←→ A2
  ↑ ↘ ↗ ↑
  ↓   ✕  ↓
  A3 ←→ A4

Agent 之间直接通信，通过消息总线协调。优点是扩展性好，缺点是调试困难。

实际项目中推荐用模式一，因为大部分业务场景都是"一个大脑指挥多个手脚"的模式。

3.2 基于 MCP 的 Agent 协作实现

核心思路：把每个 Agent 本身也封装成一个 MCP Server。

from mcp.server import Server
from mcp.types import Tool, TextContent
import json

class AgentServer:
    """Agent 基类，自动注册为 MCP Server"""

    def __init__(self, name: str, description: str):
        self.server = Server(name)
        self.name = name
        self.description = description
        self._register_tools()

    def _register_tools(self):
        @self.server.list_tools()
        async def list_tools():
            return self.get_tools()

        @self.server.call_tool()
        async def call_tool(name, arguments):
            return await self.execute(name, arguments)

    def get_tools(self) -> list:
        """子类实现：声明工具列表"""
        raise NotImplementedError

    async def execute(self, name: str, arguments: dict):
        """子类实现：工具执行逻辑"""
        raise NotImplementedError


class CoderAgent(AgentServer):
    """代码生成 Agent"""

    def __init__(self):
        super().__init__("coder-agent", "负责代码生成和修改")

    def get_tools(self):
        return [
            Tool(
                name="generate_code",
                description="根据需求生成代码",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "requirement": {"type": "string"},
                        "language": {"type": "string", "default": "python"}
                    },
                    "required": ["requirement"]
                }
            )
        ]

    async def execute(self, name, arguments):
        if name == "generate_code":
            # 这里可以接入 LLM 进行代码生成
            req = arguments["requirement"]
            lang = arguments.get("language", "python")
            code = await self._call_llm(req, lang)
            return [TextContent(type="text", text=code)]

    async def _call_llm(self, requirement, language):
        # 调用大模型生成代码
        prompt = f"用{language}实现以下需求：{requirement}"
        # ... 调用 API
        return f"# Generated code for: {requirement}\nprint('Hello, Agent!')"

四、实战：搭建一个自动化代码审查系统

理论讲够了，来点实际的。我们搭建一个 自动化 Code Review 系统，由三个 Agent 协作完成：

1. Reader Agent：读取代码文件
2. Analyzer Agent：分析代码质量
3. Reporter Agent：生成审查报告

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

async def code_review_pipeline(file_path: str):
    """代码审查流水线"""

    # 1. Reader Agent - 读取代码
    reader_params = StdioServerParameters(
        command="python",
        args=["agents/reader_server.py"]
    )

    # 2. Analyzer Agent - 分析代码
    analyzer_params = StdioServerParameters(
        command="python",
        args=["agents/analyzer_server.py"]
    )

    # 3. Reporter Agent - 生成报告
    reporter_params = StdioServerParameters(
        command="python",
        args=["agents/reporter_server.py"]
    )

    # 串行执行流水线
    async with stdio_client(reader_params) as (reader_read, reader_write):
        async with ClientSession(reader_read, reader_write) as reader:
            await reader.initialize()
            code_result = await reader.call_tool("read_file", {"path": file_path})
            code_content = code_result[0].text

    async with stdio_client(analyzer_params) as (analyzer_read, analyzer_write):
        async with ClientSession(analyzer_read, analyzer_write) as analyzer:
            await analyzer.initialize()
            analysis = await analyzer.call_tool("analyze_code", {"code": code_content})

    async with stdio_client(reporter_params) as (reporter_read, reporter_write):
        async with ClientSession(reporter_read, reporter_write) as reporter:
            await reporter.initialize()
            report = await reporter.call_tool("generate_report", {
                "code": code_content,
                "analysis": analysis[0].text
            })

    return report[0].text

# 运行
if __name__ == "__main__":
    result = asyncio.run(code_review_pipeline("src/main.py"))
    print(result)

这套系统跑起来之后，你只需要指定一个代码文件，三个 Agent 就会自动协作完成审查，输出一份结构化的报告。

五、踩坑记录：你大概率会遇到的问题

坑 1：工具描述写不好，Agent 就"傻了"

MCP 的工具描述（description）不是给人看的，是给 LLM 看的。描述写得模糊，LLM 就不知道什么时候该调用这个工具。

# ❌ 错误示范
Tool(name="query", description="查询数据")

# ✅ 正确示范
Tool(
    name="query_user_by_name",
    description="根据用户名精确或模糊查询用户信息。支持模糊匹配，输入部分姓名即可。返回用户ID、姓名、邮箱、注册时间。",
    inputSchema={...}
)

坑 2：超时处理

Agent 调用外部工具时，经常会遇到超时。一定要在 Client 端设置合理的超时时间：

import asyncio

try:
    result = await asyncio.wait_for(
        session.call_tool("slow_tool", args),
        timeout=30.0
    )
except asyncio.TimeoutError:
    return [TextContent(type="text", text="工具调用超时，请稍后重试")]

坑 3：上下文窗口溢出

多 Agent 协作时，每个 Agent 都会消耗上下文窗口。如果不对消息做裁剪，几个回合下来就会爆。

解决方案：每个 Agent 维护自己的 滑动窗口，只保留最近 N 轮对话和关键的长文本摘要。

六、性能优化建议

1. 工具缓存：对频繁调用的工具结果做缓存，避免重复计算
2. 并行调用：没有依赖关系的工具调用，用 asyncio.gather 并行执行
3. 流式返回：大结果集用流式传输，避免一次性加载到内存
4. 连接池：MCP Server 连接用连接池管理，避免频繁建立/销毁

# 并行调用示例
results = await asyncio.gather(
    session.call_tool("query_db", {"sql": "SELECT * FROM orders"}),
    session.call_tool("fetch_api", {"url": "https://api.example.com/data"}),
    session.call_tool("read_file", {"path": "config.json"})
)

总结

MCP 协议给了 Agent 生态一个统一的"语言"。在这个基础上，多 Agent 协作不再是纸上谈兵，而是可以工程化落地的方案。

几个关键 takeaways：

• MCP 的价值不在于技术复杂度，而在于标准化。就像 HTTP 统一了 Web，MCP 正在统一 Agent 的工具调用。
• 多 Agent 系统的核心挑战不是通信，而是任务拆解。如何把一个复杂任务合理地拆分给不同 Agent，这才是架构师要思考的问题。
• 工具描述就是 Agent 的 API 文档，写得好不好直接决定了 Agent 的智商。

Agent 时代才刚刚开始。现在入场，还来得及。

---

如果觉得有帮助，点个赞再走 👇 有问题欢迎评论区交流。