Agent/docs/research/a2a-mamp-protocol.md

MAMP：Minimal Agent Message Protocol

更新日期： 2026-03-04

设计目标

实现与其他 Agent 系统（非本系统）的通用交互接口，保持最简化原则。

与现有方案的关系：

• A2A 跨设备通信：内部 Agent 间通信（基于 Trace API）
• MAMP 协议（本文档）：与外部 Agent 系统的通用交互

---

核心设计原则

1. 最小化协议：只定义消息信封，不管内容格式
2. 适配器模式：通过适配器层与内部系统集成
3. 松耦合：各家 Agent 保持独立实现
4. 渐进式：先实现基础功能，需要时再扩展

---

消息格式

基础消息结构

{
  "protocol": "mamp/1.0",
  "message_id": "msg-uuid-123",
  "conversation_id": "conv-uuid-456",
  "from": "agent://your-domain.com/agent-123",
  "to": "agent://other-domain.com/agent-456",
  "content": [...],
  "metadata": {
    "timestamp": "2026-03-04T10:00:00Z"
  }
}

字段说明

字段	类型	必需	说明
protocol	string	是	协议版本标识
message_id	string	是	消息唯一标识
conversation_id	string	否	对话标识（不提供则新建对话）
from	string	是	发送方 Agent URI
to	string	是	接收方 Agent URI
content	string/array	是	消息内容（支持多模态）
metadata	object	是	元数据（时间戳等）

新建 vs 续跑

规则：通过 conversation_id 字段判断

• 无 conversation_id（null 或不存在）→ 新建对话，接收方生成并返回 conversation_id
• 有 conversation_id → 续跑对话，接收方查找对应的内部 trace_id

conversation_id 与 trace_id 的关系：

• conversation_id：跨 Agent 的对话标识符，双方共享
• trace_id：每个 Agent 内部的执行记录，各自独立
• 每个 Agent 维护 conversation_id → trace_id 映射

---

多模态内容格式

Content 结构

参考 Anthropic SDK 和现有多模态实现（agent/docs/multimodal.md）：

{
  "content": [
    {
      "type": "text",
      "text": "这是文本内容"
    },
    {
      "type": "image",
      "source": {
        "type": "url",
        "url": "https://...",
        "media_type": "image/png"
      }
    },
    {
      "type": "image",
      "source": {
        "type": "base64",
        "media_type": "image/jpeg",
        "data": "base64..."
      }
    },
    {
      "type": "code",
      "language": "python",
      "code": "def hello(): pass"
    },
    {
      "type": "file",
      "name": "report.pdf",
      "mime_type": "application/pdf",
      "source": {
        "type": "url",
        "url": "https://..."
      }
    }
  ]
}

纯文本简写

{
  "content": "纯文本消息"
}

等价于：

{
  "content": [{"type": "text", "text": "纯文本消息"}]
}

---

Agent Card（身份与能力）

每个 Agent 提供静态的 card 端点，用于身份识别和能力发现。

端点

GET https://your-agent.com/mamp/v1/card

响应格式

{
  "protocol": "mamp/1.0",
  "agent_id": "agent://your-domain.com/agent-123",
  "name": "Code Analyst",
  "description": "专注于代码分析的 Agent",

  "owner": {
    "user_id": "user-789",
    "user_name": "张三",
    "organization": "YourCompany"
  },

  "device": {
    "device_id": "device-mac-001",
    "device_name": "MacBook Pro",
    "location": "Beijing Office",
    "platform": "darwin"
  },

  "capabilities": {
    "content_types": ["text", "image", "code"],
    "max_message_size": 10485760,
    "streaming": true,
    "async": true,
    "tools": ["code_analysis", "file_read", "web_search"]
  },

  "access": {
    "public": false,
    "allowed_agents": ["agent://trusted.com/*"],
    "require_auth": true
  }
}

---

传输层

HTTP REST（最简实现）

发送消息：

POST https://other-agent.com/mamp/v1/messages
Content-Type: application/json
Authorization: Bearer {api_key}

{MAMP 消息体}

响应：

{
  "conversation_id": "conv-abc-123",
  "message_id": "msg-xyz-456",
  "status": "received"
}

错误响应：

{
  "error": "conversation_not_found",
  "message": "Conversation conv-xxx not found",
  "status_code": 404
}

可选扩展

• WebSocket：实时双向通信
• Server-Sent Events：流式响应
• Message Queue：异步消息（NATS/Redis）

---

寻址方案

使用 URI 格式：agent://domain/agent-id

示例：

• agent://your-domain.com/trace-123 - 你的 Agent
• agent://claude.ai/session-456 - Claude
• agent://openai.com/assistant-789 - OpenAI Assistant

每个 Agent 系统自己决定如何解析 agent-id 部分。

---

系统集成

三层架构

┌─────────────────────────────────────────────────┐
│ Layer 3: 内部 Agent 逻辑                         │
│ (Trace, Goal, Messages...)                      │
└─────────────────────────────────────────────────┘
                    ↕
┌─────────────────────────────────────────────────┐
│ Layer 2: MAMP 适配器                             │
│ - 内部格式 ↔ MAMP 格式转换                        │
│ - conversation_id ↔ trace_id 映射                │
└─────────────────────────────────────────────────┘
                    ↕
┌─────────────────────────────────────────────────┐
│ Layer 1: 传输层（HTTP/WebSocket/MQ）             │
└─────────────────────────────────────────────────┘

接收端实现

实现位置：agent/trace/mamp_api.py

@app.post("/mamp/v1/messages")
async def receive_mamp_message(msg: dict):
    """接收外部 Agent 的 MAMP 消息"""

    conv_id = msg.get("conversation_id")

    if not conv_id:
        # 新建对话
        conv_id = f"conv-{generate_uuid()}"

        # 创建新 Trace
        async for item in runner.run(
            messages=[{"role": "user", "content": msg["content"]}],
            config=RunConfig(
                context={
                    "mamp_conversation_id": conv_id,
                    "mamp_from": msg["from"]
                }
            )
        ):
            if isinstance(item, Trace):
                await store_conversation_mapping(conv_id, item.trace_id)

        return {
            "conversation_id": conv_id,
            "message_id": msg["message_id"],
            "status": "received"
        }

    else:
        # 续跑对话
        trace_id = await get_trace_by_conversation_id(conv_id)
        if not trace_id:
            raise HTTPException(404, f"Conversation {conv_id} not found")

        await runner.run(
            messages=[{"role": "user", "content": msg["content"]}],
            config=RunConfig(trace_id=trace_id)
        )

        return {
            "conversation_id": conv_id,
            "message_id": msg["message_id"],
            "status": "received"
        }

发送端实现

实现位置：agent/tools/builtin/mamp_adapter.py

@tool(description="与外部 Agent 通信")
async def send_to_agent(
    target_agent: str,
    message: str,
    conversation_id: Optional[str] = None,
    ctx: ToolContext = None
) -> ToolResult:
    """
    发送消息到外部 Agent

    Args:
        target_agent: 目标 Agent URI (agent://domain/id)
        message: 消息内容
        conversation_id: 对话 ID（可选，不提供则新建）
    """

    # 构建 MAMP 消息
    mamp_msg = {
        "protocol": "mamp/1.0",
        "message_id": generate_uuid(),
        "from": f"agent://{config.domain}/{ctx.trace_id}",
        "to": target_agent,
        "content": message,
        "metadata": {"timestamp": datetime.now().isoformat()}
    }

    if conversation_id:
        mamp_msg["conversation_id"] = conversation_id

    # 发送
    agent_url = parse_agent_url(target_agent)
    response = await http_post(f"{agent_url}/mamp/v1/messages", mamp_msg)

    # 新建时存储映射
    if not conversation_id:
        await store_conversation_mapping(
            response["conversation_id"],
            ctx.trace_id
        )

    return ToolResult(
        title=f"已发送到 {target_agent}",
        output=f"Conversation ID: {response['conversation_id']}",
        long_term_memory=f"与 {target_agent} 的对话 {response['conversation_id']}"
    )

conversation_id 映射存储

实现位置：agent/trace/conversation_store.py

class ConversationStore:
    """管理 MAMP conversation_id 和内部 trace_id 的映射"""

    def __init__(self, base_dir: str = ".trace"):
        self.mapping_file = Path(base_dir) / "mamp_conversations.json"

    async def store_mapping(self, conversation_id: str, trace_id: str):
        """存储映射关系"""
        mappings = await self._load_mappings()
        mappings[conversation_id] = {
            "trace_id": trace_id,
            "created_at": datetime.now().isoformat(),
            "last_message_at": datetime.now().isoformat()
        }
        await self._save_mappings(mappings)

    async def get_trace_id(self, conversation_id: str) -> Optional[str]:
        """根据 conversation_id 查找 trace_id"""
        mappings = await self._load_mappings()
        mapping = mappings.get(conversation_id)
        return mapping["trace_id"] if mapping else None

---

使用示例

新建对话

# 调用外部 Agent
result = await send_to_agent(
    target_agent="agent://other.com/code-analyst",
    message="帮我分析这段代码的性能"
)
# 返回: {"conversation_id": "conv-abc-123", ...}

续跑对话

# 继续之前的对话
result = await send_to_agent(
    target_agent="agent://other.com/code-analyst",
    message="那如果用异步方案呢？",
    conversation_id="conv-abc-123"
)

多模态消息

# 发送图片
result = await send_to_agent(
    target_agent="agent://other.com/image-analyst",
    message={
        "content": [
            {"type": "text", "text": "分析这张图片"},
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": "image/png",
                    "data": encode_image_base64("screenshot.png")
                }
            }
        ]
    }
)

---

与现有标准的关系

MAMP 可以作为其他标准的"翻译层"：

• MCP (Model Context Protocol) → 写 MCP ↔ MAMP 适配器
• OpenAI Assistant API → 写 OpenAI ↔ MAMP 适配器
• 自定义协议 → 写对应的适配器

核心思想：不要试图统一所有 Agent 的内部实现，而是提供一个最薄的互操作层。

---

可选扩展

如果需要更丰富的功能，可以逐步添加：

• 认证：在 metadata 中加 auth_token
• 流式传输：使用 Server-Sent Events 或 WebSocket
• 异步回调：加 callback_url 字段
• 能力协商：通过 /mamp/v1/card 端点
• 错误处理：标准化错误码

---

实现路线图

Phase 1：基础协议（1-2 周）

目标：实现最简 MAMP 协议

任务：

1. 实现 /mamp/v1/messages 端点（接收消息）
2. 实现 /mamp/v1/card 端点（Agent 身份）
3. 实现 send_to_agent 工具（发送消息）
4. 实现 ConversationStore（映射管理）
5. 支持纯文本消息

Phase 2：多模态支持（1 周）

目标：支持图片、代码等多模态内容

任务：

1. 扩展 content 格式处理
2. 集成现有多模态实现（agent/llm/prompts/wrapper.py）
3. 支持 base64 和 URL 两种图片传输方式

Phase 3：增强功能（可选）

目标：认证、流式、异步等高级功能

任务：

1. API Key 认证
2. WebSocket 流式传输
3. 异步回调机制
4. 错误处理和重试

---

相关文档

• A2A 跨设备通信：内部 Agent 间通信方案
• 多模态支持：图片、PDF 处理
• 工具系统：工具定义、注册
• Agent 框架：核心 Agent 能力