多智能体系统架构设计：MCP与A2A协议的分层协作实战

1. 从单体智能到协作网络：多智能体系统到底在解决什么问题

如果你最近在折腾 AI Agent，大概率会有一种感觉：单个 Agent 能做的事情，很快就摸到天花板了。你给它接上工具、挂上知识库、写好提示词，它能帮你查资料、写代码、跑命令，但一旦任务变复杂——比如"帮我分析这份需求文档，拆出模块，生成接口定义，再写一版可运行的骨架代码，最后跑一遍测试"——单体 Agent 就开始顾此失彼。它要么在长上下文里迷失，要么在工具调用链里绕圈，要么干脆把中间步骤忘干净。

这不是模型不够聪明的问题，而是架构问题。一个 Agent 同时扮演需求分析师、架构师、程序员、测试工程师，本质上是在让一个大脑同时处理四种截然不同的认知模式。人类团队不会这么干，AI 系统也不该这么干。

多智能体系统（Multi-Agent System）的核心思路，就是把一个复杂任务拆成若干个职责单一、边界清晰的智能体，让它们通过标准化的协议互相通信、分工协作。而 MCP（Model Context Protocol）和 A2A（Agent-to-Agent）这两个协议，恰好分别解决了这个架构里最关键的两个问题：Agent 怎么连接外部能力，以及Agent 之间怎么互相说话。

这一篇是系列第七篇，前面几篇我们聊过 MCP 的基础接入、工具注册、上下文管理，也聊过 A2A 的消息格式和任务编排。这一篇我想把视角拉高一点，不再纠结单个 API 怎么调，而是讲清楚：当你真的要设计一个多智能体 AI 系统时，MCP 和 A2A 各自应该放在架构的哪个位置，它们怎么配合，以及我在实际搭建过程中踩过的那些坑。

适合读这篇的人：已经了解 LLM 基础调用、写过至少一个能跑通的 Agent Demo、现在想往多智能体方向进阶的开发者。如果你还没写过 Agent，建议先回去补一下工具调用和 ReAct 循环的基础，不然这篇里的很多设计取舍你会觉得"为什么要这么麻烦"。

2. MCP 与 A2A 的职责边界：别把两个协议用混了

2.1 MCP 解决的是"Agent 与能力之间"的接口问题

MCP 的本质，是给 LLM 定义了一套标准化的能力接入方式。在没有 MCP 之前，你每接一个工具，就要写一套 function calling 的 schema，每换一个模型，schema 格式可能还要改。MCP 把这个过程抽象成了 Server 和 Client 两层：能力提供方实现一个 MCP Server，把工具、资源、提示词模板暴露出来；Agent 侧作为 MCP Client，通过统一协议去发现和调用这些能力。

我习惯用一个类比来解释：MCP 就像是 USB-C 接口。以前每个设备有自己的充电口，现在统一了，你不需要关心对面是硬盘还是显示器，插上就能用。MCP Server 可以是本地进程，也可以是远程服务，Agent 不需要知道工具背后是 Python 脚本还是 HTTP API，它只需要知道"这里有一个叫search_docs的工具，输入是 query，输出是文档列表"。

在实际项目里，我通常会把 MCP Server 按能力域来划分，而不是按工具数量。比如：

• 一个filesystemServer，负责所有文件读写、目录遍历
• 一个databaseServer，封装所有 SQL 查询和 schema 探查
• 一个browserServer，处理网页抓取和 DOM 操作
• 一个code_runnerServer，负责沙箱内执行代码

这样划分的好处是，每个 Server 的职责边界清晰，权限也好控制。你不可能让一个负责查数据库的 Agent 顺手把文件系统删了，因为它的 MCP Client 根本没连那个 Server。

2.2 A2A 解决的是"Agent 与 Agent 之间"的协作问题

A2A 要处理的是完全不同层面的问题。当你有多个 Agent，每个 Agent 有自己的角色、自己的上下文、自己的工具集，它们之间怎么传递任务、怎么同步状态、怎么处理依赖关系？

最原始的做法是让 Agent 之间直接互相调用函数，A 把结果传给 B，B 处理完传给 C。但这样做的问题是耦合太紧。A 必须知道 B 的接口长什么样，B 改了参数 A 就得跟着改。而且一旦任务链变长，错误处理、超时重试、状态回滚都会变成噩梦。

A2A 的思路是引入一个消息层。Agent 之间不直接调用，而是通过标准化的消息格式通信。每条消息包含：发送方、接收方、任务描述、输入数据、期望的输出格式、超时时间、优先级。接收方处理完后，再发一条响应消息回去。这个模式听起来很像微服务里的消息队列，实际上设计理念也确实相通。

我在实际项目里用 A2A 时，最看重的是它带来的可观测性。因为所有 Agent 间的通信都走消息层，你可以很轻松地记录每一条消息的流转、耗时、成功失败状态。调试多智能体系统最痛苦的就是"不知道哪一步出了问题"，有了消息日志，你可以像查快递轨迹一样，看到任务卡在哪个 Agent 手里。

2.3 两个协议在架构中的位置对比

这张表是我自己在设计系统时反复对照的。一个常见的误区是：有人试图用 MCP 来做 Agent 间的通信，把另一个 Agent 包装成一个 MCP 工具。短期看能跑通，但长期会出问题——因为 Agent 是有状态的、会主动发起任务的，而 MCP 工具的设计假设是被动、无状态的。用错了抽象层，后面扩展会非常痛苦。

3. 多智能体系统的分层设计：从任务入口到结果汇总

3.1 编排层：谁来决定任务怎么分

多智能体系统的第一层是编排层（Orchestration Layer）。这一层的职责是接收用户输入，理解意图，把大任务拆成子任务，然后决定每个子任务交给哪个 Agent。

编排层可以是一个专门的 Orchestrator Agent，也可以是一段确定性的代码逻辑。我的经验是：能用代码编排的就别用 Agent 编排。原因很简单，代码编排是确定性的、可测试的、可调试的；Agent 编排虽然灵活，但引入了不确定性，一旦编排逻辑出错，你很难定位是模型理解错了还是提示词写歪了。

具体做法上，我会把任务分成两类：

• 结构化任务：流程固定，比如"先解析文档，再抽取字段，再校验，再入库"。这种直接用代码写死流程，每个步骤调用对应的 Agent。
• 非结构化任务：流程需要根据中间结果动态决定，比如"根据用户问题判断需要查哪些数据源"。这种才交给 Orchestrator Agent 去决策。

Orchestrator Agent 的提示词里，我会明确列出所有可用的下游 Agent 及其能力描述，让它输出一个结构化的任务计划。这个计划通常是一个 JSON 数组，每个元素包含agent_name、task_description、input_data、depends_on。有了depends_on字段，编排层就能构建出任务依赖图，并行执行没有依赖的任务。

3.2 执行层：每个 Agent 只做一件事

执行层的每个 Agent 都应该遵循单一职责原则。我见过太多项目，一个 Agent 既负责写代码又负责跑测试还负责修 bug，结果就是提示词越写越长，行为越来越不可预测。

一个设计良好的执行层 Agent 应该包含这几个要素：

• 明确的角色定义：一句话说清楚它是干什么的，比如"你是一个专门负责从需求文档中抽取功能点的分析 Agent"。
• 受限的工具集：只挂载它真正需要的 MCP Server。写代码的 Agent 不需要数据库权限，查数据的 Agent 不需要文件写入权限。
• 固定的输出格式：每个 Agent 的输出都应该是结构化的，方便下游消费。我通常要求所有 Agent 输出 JSON，并在提示词里给出 schema。
• 清晰的失败信号：Agent 遇到无法处理的情况时，应该返回一个明确的错误码和原因，而不是硬编一个看起来像答案的东西。

这里有个实操细节：Agent 的上下文要隔离。不要让所有 Agent 共享一个巨大的对话历史，那样既浪费 token 又容易互相干扰。每个 Agent 只应该拿到它完成任务所需的最小上下文。A2A 消息里传递的input_data就是这个最小上下文的载体。

3.3 汇总层：结果怎么合并、冲突怎么处理

汇总层是最容易被忽视的一层。多个 Agent 的输出汇总到一起时，经常会出现冲突：分析 Agent 说这个字段是必填的，校验 Agent 说这个字段可以为空；代码 Agent 生成的接口和文档 Agent 描述的接口对不上。

处理冲突的策略我一般分三档：

1. 优先级裁决：给每个 Agent 的输出设定可信度权重，冲突时以高权重为准。比如文档 Agent 的可信度高于代码 Agent，因为文档是源头。
2. 投票机制：对于分类、判断类任务，让多个 Agent 独立给出结论，取多数。这个在需要高可靠性的场景很有用，但成本也高。
3. 人工介入：对于无法自动裁决的冲突，挂起任务并通知人工处理。别想着所有冲突都能自动解决，有些就是需要人来拍板。

汇总层的输出应该是最终交付物，而不是又一份中间结果。这意味着汇总层要负责格式转换、字段补全、一致性校验。我通常会在汇总层加一个 Final Validator，用确定性的代码检查输出是否符合预期 schema，不符合就直接打回重跑。

4. 用 MCP 搭建 Agent 的能力底座：实操中的取舍

4.1 MCP Server 的粒度怎么定

这是我在实际项目里被问得最多的问题：一个 MCP Server 应该暴露多少个工具？

我的答案是：按能力域划分，单个 Server 的工具数量控制在 5 到 15 个之间。太少了，Server 数量爆炸，管理成本高；太多了，Agent 在选择工具时容易混淆，而且单个 Server 的权限范围太大，不符合最小权限原则。

举个例子，我做代码生成系统时，把 MCP Server 分成了这几个：

• repo_server：代码仓库操作，包括读文件、写文件、列目录、搜索代码、查看 git 历史
• test_server：测试相关，包括运行测试、查看测试报告、获取覆盖率
• lint_server：代码检查，包括跑 linter、格式化、类型检查
• doc_server：文档相关，包括读需求文档、写接口文档、查 API 手册

每个 Server 的工具数量都在 10 个左右，Agent 挂载时按需选择。写代码的 Agent 挂repo_server和lint_server，跑测试的 Agent 挂test_server和repo_server（只读权限）。

4.2 工具描述怎么写才能让 Agent 选对

MCP 工具的 description 字段，直接决定了 Agent 能不能在正确的时候调用正确的工具。我见过太多项目，工具描述写得像 API 文档，什么"执行查询操作，返回结果集"，Agent 根本不知道什么时候该用它。

好的工具描述应该包含三部分：

• 什么时候用：明确使用场景。比如"当需要根据关键词查找代码文件时使用此工具"。
• 输入是什么：参数的含义和格式。别只写参数名，要写清楚期望的值长什么样。
• 输出是什么：返回结果的结构。Agent 需要知道拿到结果后怎么解析。

我通常会这样写：

工具名：search_code 描述：在代码仓库中根据关键词搜索匹配的代码片段。当你需要定位某个函数、类或变量的定义位置时使用此工具。输入为搜索关键词字符串，支持正则表达式。返回匹配的文件路径、行号和代码片段列表，最多返回 20 条结果。

这段描述里，"当你需要定位某个函数、类或变量的定义位置时使用"就是使用场景，"输入为搜索关键词字符串，支持正则表达式"是输入说明，"返回匹配的文件路径、行号和代码片段列表"是输出说明。Agent 读完这段，基本不会用错。

4.3 MCP 连接的生命周期管理

MCP 连接不是建立一次就一劳永逸的。在实际运行中，Server 可能崩溃、网络可能抖动、token 可能过期。我踩过最大的坑就是：Agent 在任务执行到一半时 MCP 连接断了，它没有报错，而是继续用空结果往下走，最后产出一个看起来完整但完全错误的交付物。

后来我加了几层保护：

• 连接健康检查：每次调用工具前，先 ping 一下 Server，确认连接可用。
• 调用超时设置：每个工具调用设置合理的超时时间，超时后返回明确的错误，而不是无限等待。
• 重试与降级：对于幂等的读操作，失败后自动重试 2 到 3 次；对于写操作，失败后不自动重试，而是上报给编排层决定。
• 结果校验：工具返回后，检查返回结构是否符合预期 schema，不符合就当作失败处理。

这些保护措施看起来繁琐，但在生产环境里，它们能把很多"静默失败"变成"显式失败"，大大降低调试难度。

5. 用 A2A 串起 Agent 协作：消息设计与状态管理

5.1 A2A 消息的字段设计

A2A 消息是整个协作层的血液，字段设计得好不好，直接决定了系统的可扩展性。我经过几轮迭代后，固定下来一套消息结构，包含这些字段：

• message_id：消息唯一标识，用于追踪和去重
• correlation_id：关联 ID，同一个任务链上的所有消息共享这个 ID，方便串联
• sender：发送方 Agent 标识
• receiver：接收方 Agent 标识
• task_type：任务类型，接收方根据这个字段决定用哪个处理逻辑
• payload：任务输入数据，结构化 JSON
• expected_output_schema：期望的输出格式，让接收方知道该返回什么
• timeout_ms：超时时间
• priority：优先级，用于消息队列调度
• retry_count：当前重试次数

这套字段里，我觉得最有用的是correlation_id和expected_output_schema。前者让全链路追踪成为可能，后者让 Agent 之间的契约变得显式。接收方拿到消息后，先看expected_output_schema，就知道自己该产出什么结构的数据，不用猜。

5.2 同步调用还是异步消息

A2A 通信有两种模式：同步和异步。

同步模式下，发送方发出消息后阻塞等待，直到收到响应或超时。这种模式实现简单，适合短任务、强依赖的场景。比如编排层让分析 Agent 抽取字段，必须等它返回才能进行下一步。

异步模式下，发送方发出消息后立即返回，接收方处理完后通过回调或消息队列通知。这种模式适合长任务、弱依赖的场景。比如代码生成和文档生成可以并行，谁先完成都行，最后汇总时再等齐。

我的经验是：默认用异步，只在确实需要立即结果时才用同步。因为同步调用会阻塞 Agent，降低整体吞吐。而且同步调用链一长，超时时间很难设置——设短了容易误杀，设长了整体延迟高。

异步模式下，状态管理就变得很重要。每个任务需要有一个状态机，记录它当前处于哪个阶段：pending、running、completed、failed、timeout。编排层通过查询状态机来决定下一步动作。这个状态机我通常用一个轻量的存储来实现，比如 Redis 或者 SQLite，关键是保证并发安全。

5.3 Agent 注册与发现机制

当 Agent 数量多起来之后，你需要一个注册中心来管理"现在有哪些 Agent 可用，它们各自能处理什么任务"。

最简单的做法是配置文件，启动时加载。但这样每次加 Agent 都要改配置重启，不够灵活。稍微好一点的做法是用一个注册服务，Agent 启动时向注册中心报到，声明自己的agent_id、capabilities、endpoint。编排层在分配任务时，先查注册中心，找到能处理该任务的 Agent 列表，再根据负载情况选一个。

我在项目里用的是能力标签匹配的方式。每个 Agent 注册时声明自己能处理的task_type列表，编排层根据任务的task_type去匹配。如果多个 Agent 都能处理，就按当前队列长度做负载均衡。这套机制不复杂，但能让系统在 Agent 增减时保持弹性。

6. 实战踩坑：那些文档里不会写的教训

6.1 上下文膨胀导致 Agent 变傻

这是我在多智能体系统里踩过最贵的坑。一开始为了"让 Agent 有足够信息"，我把所有上游 Agent 的输出都塞进下游 Agent 的上下文。结果任务链跑到第四五个 Agent 时，上下文已经几万 token，Agent 开始出现"遗忘"——明明前面给了字段定义，它生成代码时还是用错。

后来我改成按需传递：每个 Agent 只拿到它完成任务必需的最小信息。具体做法是在 A2A 消息的payload里只放当前任务相关的数据，而不是整个任务链的累积结果。如果某个 Agent 确实需要历史信息，让它通过 MCP 工具去查，而不是一股脑塞进上下文。

这个改动之后，不仅 Agent 的输出质量提升了，token 成本也降了一大半。

6.2 Agent 之间的"踢皮球"

多智能体系统里有一种很隐蔽的失败模式：A Agent 认为这个任务该 B 处理，B 认为该 A 处理，结果任务在两者之间来回传递，永远不落地。

这个问题的根源是职责边界模糊。解决办法是在设计阶段就把每个 Agent 的职责写清楚，并且在 A2A 消息里加一个handled_by字段，记录这个任务已经被哪些 Agent 处理过。如果一个任务被同一个 Agent 处理超过两次，或者流转超过预设的最大跳数，就强制上报给编排层，由编排层裁决或转人工。

我在编排层加了一个"任务跳数计数器"，任何任务流转超过 10 跳就自动挂起并告警。这个简单的保护机制，帮我拦下了好几次潜在的无限循环。

6.3 工具调用的副作用没有隔离

MCP 工具里如果有写操作——写文件、改数据库、发请求——一定要做副作用隔离。我遇到过 Agent 在重试逻辑下，把同一个文件写了三遍，因为前两次它以为失败了，实际上只是响应超时。

解决办法有两个：一是写操作幂等化，比如写文件时先检查内容是否已存在，存在就跳过；二是写操作不自动重试，失败后上报，由编排层决定是否重试。我倾向于后者，因为幂等化需要每个工具自己实现，容易漏；而统一在编排层控制重试策略，更可靠。

另外，所有写操作我都要求 Agent 先输出一个"计划"，说明它打算改什么，编排层审核通过后才真正执行。这个"计划-执行"两步走，虽然增加了一次往返，但能有效防止 Agent 误操作。

6.4 超时设置的经验值

超时时间设多少，这个问题没有标准答案，但有一些经验值可以参考：

这些值不是拍脑袋定的，是我在实际运行中根据 P99 延迟不断调整出来的。原则是：超时时间应该略大于正常情况下的 P99 延迟，这样既能拦住真正的异常，又不会误杀正常但稍慢的请求。

7. 一个可复现的最小多智能体系统骨架

7.1 目录结构与依赖

说了这么多设计原则，最后给一个可以直接跑起来的最小骨架。这个骨架实现了"需求文档 -> 功能点抽取 -> 接口定义 -> 代码骨架"的流程，用到了 MCP 做文件操作，用 A2A 做 Agent 间通信。

目录结构如下：

multi_agent_system/ ├── orchestrator/ │ ├── main.py # 编排层入口 │ └── task_planner.py # 任务拆解逻辑 ├── agents/ │ ├── analyzer.py # 需求分析 Agent │ ├── designer.py # 接口设计 Agent │ └── coder.py # 代码生成 Agent ├── mcp_servers/ │ ├── filesystem.py # 文件操作 MCP Server │ └── doc_parser.py # 文档解析 MCP Server ├── a2a/ │ ├── message.py # 消息结构定义 │ └── bus.py # 消息总线 └── config/ └── agents.yaml # Agent 注册配置

依赖主要是mcp官方 SDK、pydantic做数据校验、asyncio做异步调度。不需要额外的消息队列中间件，初期用内存队列就够，等 Agent 数量上来了再换 Redis 或 RabbitMQ。

7.2 核心代码片段

消息结构定义：

from pydantic import BaseModel from typing import Any, Dict, Optional import uuid class A2AMessage(BaseModel): message_id: str = str(uuid.uuid4()) correlation_id: str sender: str receiver: str task_type: str payload: Dict[str, Any] expected_output_schema: Optional[Dict] = None timeout_ms: int = 60000 priority: int = 5 retry_count: int = 0

编排层的任务拆解：

async def plan_and_execute(user_input: str): # 第一步：分析需求 analyze_msg = A2AMessage( correlation_id=str(uuid.uuid4()), sender="orchestrator", receiver="analyzer", task_type="extract_features", payload={"document": user_input}, expected_output_schema={"features": "list[str]"} ) features = await bus.send_and_wait(analyze_msg) # 第二步：设计接口 design_msg = A2AMessage( correlation_id=analyze_msg.correlation_id, sender="orchestrator", receiver="designer", task_type="design_api", payload={"features": features["features"]}, expected_output_schema={"apis": "list[dict]"} ) apis = await bus.send_and_wait(design_msg) # 第三步：生成代码 code_msg = A2AMessage( correlation_id=analyze_msg.correlation_id, sender="orchestrator", receiver="coder", task_type="generate_code", payload={"apis": apis["apis"]}, expected_output_schema={"files": "list[dict]"} ) code = await bus.send_and_wait(code_msg) return code

MCP Server 的注册（以文件操作为例）：

from mcp.server import Server from mcp.types import Tool, TextContent server = Server("filesystem") @server.list_tools() async def list_tools(): return [ Tool( name="read_file", description="读取指定路径的文件内容。当需要查看文件内容时使用。输入为文件路径字符串，返回文件文本内容。", inputSchema={ "type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"] } ), Tool( name="write_file", description="将内容写入指定路径的文件。当需要创建或修改文件时使用。输入为文件路径和内容字符串，返回操作结果。", inputSchema={ "type": "object", "properties": { "path": {"type": "string"}, "content": {"type": "string"} }, "required": ["path", "content"] } ) ]

7.3 运行与验证

跑起来之后，你可以用一个简单的需求文档测试：

输入：实现一个用户注册功能，需要邮箱、密码、昵称三个字段，注册成功后发送欢迎邮件。

预期输出应该包含：分析 Agent 抽出的功能点列表、设计 Agent 给出的接口定义（注册接口的路径、方法、请求体、响应体）、代码 Agent 生成的骨架代码。

验证的时候重点看三件事：一是每个 Agent 的输出是否符合expected_output_schema；二是correlation_id是否贯穿全链路；三是总耗时是否在可接受范围内。如果某个环节卡住，去消息日志里查对应的message_id，就能定位到问题。

8. 扩展方向：从骨架到生产系统还差什么

这个骨架能跑通流程，但离生产系统还有距离。根据我的经验，接下来需要补的主要是这几块：

可观测性。加一套完整的日志和指标采集，记录每个 Agent 的调用次数、成功率、P99 延迟、token 消耗。这些数据是后续优化的基础。我通常用 OpenTelemetry 做链路追踪，把 A2A 消息的correlation_id作为 trace_id，这样在追踪系统里能看到完整的调用链。

容错与降级。给每个 Agent 配置降级策略：主 Agent 失败时切备用 Agent，或者降级到更简单的处理逻辑。比如代码生成 Agent 失败时，可以降级为只输出接口定义，让用户手动补代码。

权限与审计。生产环境里，每个 Agent 能访问哪些 MCP Server、能执行哪些操作，都要有明确的权限控制。所有写操作都要留审计日志，记录谁在什么时候改了什么。

成本控制。多智能体系统的 token 消耗是单体 Agent 的好几倍，必须做成本监控。我会给每个任务设置 token 预算，超预算就告警或中断。另外，能缓存的中间结果尽量缓存，避免重复计算。

人工介入通道。再智能的系统也会遇到处理不了的情况，必须留一个人工介入的入口。我的做法是在编排层加一个human_review状态，任务进入这个状态后暂停，等待人工确认或修改后再继续。

这些东西每一个展开都能写一整篇，这里就不细说了。核心思路是：先把流程跑通，再逐步加固。别一上来就追求完美架构，那样很容易陷入过度设计，最后连 Demo 都跑不起来。

我在实际搭建多智能体系统时最大的体会是：架构的复杂度应该由任务的实际复杂度决定，而不是由技术的可能性决定。能用两个 Agent 解决的问题，别硬拆成五个。MCP 和 A2A 是工具，不是目标。它们的价值在于让系统在需要扩展时能平滑扩展，而不是让简单问题变复杂。