Agent通信协议的核心问题
当多个Agent需要协作时，“说什么"和"怎么说"决定了系统的天花板。一个好的通信协议需要回答：
- 消息格式：Agent间交换什么结构的数据？
- 寻址机制：消息发给谁？谁能收到？
- 语义对齐：不同Agent对同一概念的理解是否一致？
- 会话管理：多轮对话如何维持上下文？
- 错误处理：消息无法理解或处理失败怎么办？
目前主流的三种路线——MCP（Model Context Protocol）、ACP（Agent Communication Protocol）和自定义协议——各有不同的权衡。
MCP：工具集成的事实标准
设计理念
MCP由Anthropic提出，核心定位是标准化LLM与外部工具/数据源的连接。它不是Agent间通信协议，而是Agent与"世界"的接口协议。
[LLM Agent] ←MCP→ [File System MCP Server]
←MCP→ [Database MCP Server]
←MCP→ [API MCP Server]
协议结构
// MCP协议核心消息类型
// 1. 工具发现
{
"jsonrpc": "2.0",
"method": "tools/list",
"id": 1
}
// 响应
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{
"name": "search_docs",
"description": "搜索文档库",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"limit": {"type": "integer", "default": 10}
},
"required": ["query"]
}
}
]
}
}
// 2. 工具调用
{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "search_docs",
"arguments": {"query": "架构设计", "limit": 5}
},
"id": 2
}
MCP Server实现
from mcp.server import Server
from mcp.types import Tool, TextContent
import json
class DocSearchMCPServer:
def __init__(self):
self.server = Server("doc-search")
self._register_handlers()
def _register_handlers(self):
@self.server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="search_docs",
description="搜索内部文档库",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"limit": {
"type": "integer",
"default": 10,
"minimum": 1,
"maximum": 50
}
},
"required": ["query"]
}
),
Tool(
name="get_doc",
description="根据ID获取文档全文",
inputSchema={
"type": "object",
"properties": {
"doc_id": {"type": "string"}
},
"required": ["doc_id"]
}
)
]
@self.server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "search_docs":
results = await self._search(
arguments["query"],
arguments.get("limit", 10)
)
return [TextContent(
type="text",
text=json.dumps(results, ensure_ascii=False)
)]
elif name == "get_doc":
doc = await self._get_doc(arguments["doc_id"])
return [TextContent(type="text", text=doc)]
async def _search(self, query: str, limit: int) -> list[dict]:
# 实际搜索逻辑
pass
async def _get_doc(self, doc_id: str) -> str:
# 实际获取逻辑
pass
ACP：Agent间通信的学术路线
FIPA ACL基础
ACP源自FIPA（Foundation for Intelligent Physical Agents）标准，定义了Agent间的言语行为类型（Communicative Acts）：
| 行为类型 | 含义 | 示例 |
|---|---|---|
| Inform | 声明事实 | “文件已保存” |
| Request | 请求执行 | “请分析这个文件” |
| Query | 查询信息 | “数据库有多少条记录？” |
| Propose | 提出方案 | “我建议用方案A” |
| Agree | 同意 | “同意执行” |
| Refuse | 拒绝 | “无法在限定时间完成” |
| Cancel | 取消 | “取消之前的请求” |
消息结构
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
class Performative(Enum):
INFORM = "inform"
REQUEST = "request"
QUERY = "query"
PROPOSE = "propose"
AGREE = "agree"
REFUSE = "refuse"
CANCEL = "cancel"
@dataclass
class ACLMessage:
performative: Performative
sender: str
receivers: list[str]
content: str
language: str = "JSON" # 内容编码语言
ontology: str = "default" # 本体/概念体系
conversation_id: str = field(default_factory=lambda: str(uuid.uuid4()))
reply_with: str | None = None
in_reply_to: str | None = None
reply_by: str | None = None # 超时时间
protocol: str = "fipa-request" # 交互协议
def to_dict(self) -> dict:
return {
"performative": self.performative.value,
"sender": self.sender,
"receivers": self.receivers,
"content": self.content,
"conversation_id": self.conversation_id,
"in_reply_to": self.in_reply_to,
"protocol": self.protocol
}
交互协议示例：FIPA-Request
Agent A Agent B
│── request(分析报告) ──→│
│ │── agree(同意) ──→│
│ │── inform(报告内容) ──→│
│ │
│←── (完成) ─────────────│
class FIPARequestProtocol:
"""FIPA Request交互协议实现"""
async def initiate(self, agent_a, agent_b: str, content: str):
msg = ACLMessage(
performative=Performative.REQUEST,
sender=agent_a.id,
receivers=[agent_b],
content=content,
conversation_id=str(uuid.uuid4()),
reply_with=f"req-{uuid.uuid4().hex[:8]}"
)
await agent_a.send(msg)
# 等待响应
response = await agent_a.wait_for_reply(msg, timeout=30)
if response.performative == Performative.AGREE:
# 等待结果
result = await agent_a.wait_for_reply(response, timeout=300)
if result.performative == Performative.INFORM:
return result.content
elif response.performative == Performative.REFUSE:
raise Exception(f"Agent B refused: {response.content}")
自定义协议：实用主义路线
设计思路
当MCP的"工具调用"模型和ACP的"言语行为"模型都不够用时，自定义协议可以根据业务需求精确设计：
from pydantic import BaseModel
from typing import Literal, Any
class AgentMessage(BaseModel):
"""自定义Agent通信消息"""
msg_type: Literal[
"task_assign", # 任务分配
"task_result", # 任务结果
"clarify", # 澄清请求
"handoff", # 移交
"broadcast", # 广播通知
"status_sync" # 状态同步
]
sender: str
target: str | None = None # None=广播
payload: dict[str, Any]
context_id: str # 会话上下文ID
priority: int = 0 # 0=普通, 1=紧急, 2=关键
requires_ack: bool = True # 是否需要确认
class TaskAssignPayload(BaseModel):
task_id: str
task_type: str
description: str
deadline: str
dependencies: list[str] = []
class ClarifyPayload(BaseModel):
question: str
options: list[str] | None = None
context: str
消息路由器
class MessageRouter:
def __init__(self):
self.handlers: dict[str, dict[str, callable]] = {}
self.middleware: list[callable] = []
def register(self, agent_id: str, msg_type: str, handler: callable):
self.handlers.setdefault(agent_id, {})[msg_type] = handler
def add_middleware(self, mw: callable):
"""日志、鉴权、限流等中间件"""
self.middleware.append(mw)
async def route(self, msg: AgentMessage) -> Any:
# 中间件链
for mw in self.middleware:
msg = await mw(msg)
if msg is None:
return # 被中间件拦截
# 路由到目标Agent
if msg.target is None:
return await self._broadcast(msg)
handler = self.handlers.get(msg.target, {}).get(msg.msg_type)
if not handler:
raise ValueError(f"No handler for {msg.target}.{msg.msg_type}")
return await handler(msg)
三种协议对比
| 维度 | MCP | ACP | 自定义协议 |
|---|---|---|---|
| 定位 | Agent↔工具 | Agent↔Agent | 灵活 |
| 标准化 | 高 | 中 | 低 |
| 学习成本 | 低 | 高 | 中 |
| 生态 | 快速增长 | 学术为主 | 无 |
| 灵活性 | 中 | 中 | 高 |
| 语义丰富度 | 低 | 高 | 可调 |
| 序列化 | JSON-RPC | JSON/XML | 任意 |
| 适用规模 | 单Agent+多工具 | 多Agent协作 | 任意 |
选型决策
你的场景是什么?
├─ Agent连接外部工具/数据源
│ └→ MCP（生态最好，工具复用）
├─ 多个自主Agent协作
│ ├─ 需要丰富语义(协商/拒绝)?
│ │ └→ ACP（言语行为完备）
│ └─ 只需要简单的任务传递?
│ └→ 自定义协议（简单高效）
└─ 混合场景
└→ MCP(工具层) + 自定义协议(Agent层)
总结
协议选择没有对错，只有匹配度。MCP是当前Agent-工具连接的事实标准，生态优势明显。ACP在学术研究中语义更完备，但工业采用率低。自定义协议最灵活但缺乏生态。实践建议：MCP做工具层 + 轻量自定义协议做Agent层，取各家之长。