Agent通信协议设计：MCP、ACP与自定义协议的权衡

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层，取各家之长。

💬 更多交流内容，请点击本站论坛交流区：silicon-agi.com

💬 交流评论

评论加载中…

Agent通信协议的核心问题#

MCP：工具集成的事实标准#

设计理念#

协议结构#

MCP Server实现#

ACP：Agent间通信的学术路线#

FIPA ACL基础#

消息结构#

交互协议示例：FIPA-Request#

自定义协议：实用主义路线#

设计思路#

消息路由器#

三种协议对比#

选型决策#

总结#

📌 相关推荐