# MAMP：Minimal Agent Message Protocol

**更新日期：** 2026-03-04

## 设计目标

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

**与现有方案的关系**：
- [A2A 跨设备通信](./a2a-cross-device.md)：内部 Agent 间通信（基于 Trace API）
- **MAMP 协议**（本文档）：与外部 Agent 系统的通用交互

---

## 核心设计原则

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

---

## 消息格式

### 基础消息结构

```json
{
  "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`）：

```json
{
  "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://..."
      }
    }
  ]
}
```

### 纯文本简写

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

等价于：

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

---

## Agent Card（身份与能力）

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

### 端点

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

### 响应格式

```json
{
  "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（最简实现）

**发送消息**：

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

{MAMP 消息体}
```

**响应**：

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

**错误响应**：

```json
{
  "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`

```python
@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`

```python
@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`

```python
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
```

---

## 使用示例

### 新建对话

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

### 续跑对话

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

### 多模态消息

```python
# 发送图片
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 跨设备通信](./a2a-cross-device.md)：内部 Agent 间通信方案
- [多模态支持](../../agent/docs/multimodal.md)：图片、PDF 处理
- [工具系统](../../agent/docs/tools.md)：工具定义、注册
- [Agent 框架](../README.md)：核心 Agent 能力