ACP 协议深度解析：IBM BeeAI 的 Agent 通信标准如何融入 A2A 生态

"""
ACP Server 完整实现
依赖: pip install fastapi uvicorn sse-starlette
"""
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from sse_starlette.sse import EventSourceResponse
from pydantic import BaseModel, Field
from typing import Optional, Literal
from datetime import datetime
import asyncio
import uuid
import json

app = FastAPI(title="ACP Server", version="1.0.0")

# ─── 数据模型 ───

class MessageContent(BaseModel):
    type: Literal["text", "image", "json", "embedding"]
    data: str  # 文本内容 / Base64 图片 / JSON 字符串 / 向量 JSON
    metadata: Optional[dict] = None

class MessageRequest(BaseModel):
    sender_id: str
    receiver_id: str
    content: MessageContent
    thread_id: Optional[str] = None
    priority: Literal["low", "normal", "high"] = "normal"
    timeout_seconds: int = 300

class AgentCard(BaseModel):
    agent_id: str
    name: str
    description: str
    capabilities: list[str]
    supported_content_types: list[str]
    auth_required: bool = False
    max_concurrent_messages: int = 10

# ─── 内存存储（生产环境替换为 Redis/DB） ───

messages_db: dict[str, dict] = {}
agents_db: dict[str, AgentCard] = {}
message_queues: dict[str, asyncio.Queue] = {}

# ─── 注册示例 Agent ───

@app.on_event("startup")
async def register_default_agents():
    agents_db["analyst-001"] = AgentCard(
        agent_id="analyst-001",
        name="数据分析 Agent",
        description="擅长 SQL 查询和数据可视化",
        capabilities=["sql_query", "data_visualization", "statistical_analysis"],
        supported_content_types=["text", "json"],
        auth_required=True,
    )
    agents_db["writer-001"] = AgentCard(
        agent_id="writer-001",
        name="写作 Agent",
        description="擅长技术文档和报告撰写",
        capabilities=["technical_writing", "report_generation", "summarization"],
        supported_content_types=["text"],
        auth_required=False,
    )

# ─── ACP 核心端点 ───

@app.get("/agents")
async def list_agents():
    """列出所有已注册的 Agent"""
    return {"agents": list(agents_db.values())}

@app.get("/agents/{agent_id}/card")
async def get_agent_card(agent_id: str):
    """获取指定 Agent 的能力描述"""
    if agent_id not in agents_db:
        raise HTTPException(status_code=404, detail=f"Agent {agent_id} not found")
    return agents_db[agent_id]

@app.post("/messages")
async def send_message(req: MessageRequest):
    """发送消息给指定 Agent（异步）"""
    if req.receiver_id not in agents_db:
        raise HTTPException(status_code=404, detail=f"Receiver {req.receiver_id} not found")

    msg_id = str(uuid.uuid4())
    thread_id = req.thread_id or str(uuid.uuid4())

    message = {
        "message_id": msg_id,
        "thread_id": thread_id,
        "sender_id": req.sender_id,
        "receiver_id": req.receiver_id,
        "content": req.content.model_dump(),
        "status": "pending",
        "priority": req.priority,
        "created_at": datetime.utcnow().isoformat(),
        "updated_at": datetime.utcnow().isoformat(),
    }
    messages_db[msg_id] = message

    # 创建消息队列用于 SSE 推送
    message_queues[msg_id] = asyncio.Queue()

    # 异步处理消息（模拟 Agent 处理）
    asyncio.create_task(process_message(msg_id, message))

    return {
        "message_id": msg_id,
        "thread_id": thread_id,
        "status": "pending",
        "stream_url": f"/messages/{msg_id}/stream",
    }

@app.get("/messages/{message_id}")
async def get_message_status(message_id: str):
    """查询消息状态"""
    if message_id not in messages_db:
        raise HTTPException(status_code=404, detail="Message not found")
    return messages_db[message_id]

@app.get("/messages/{message_id}/stream")
async def stream_message(message_id: str):
    """SSE 流式接收消息处理进度"""
    if message_id not in message_queues:
        raise HTTPException(status_code=404, detail="No stream for this message")

    async def event_generator():
        queue = message_queues[message_id]
        while True:
            event = await queue.get()
            yield {"event": event["type"], "data": json.dumps(event["data"])}
            if event["type"] in ("completed", "failed"):
                break

    return EventSourceResponse(event_generator())

# ─── 消息处理模拟 ───

async def process_message(msg_id: str, message: dict):
    """模拟 Agent 处理消息的异步流程"""
    queue = message_queues[msg_id]

    # 状态: processing
    messages_db[msg_id]["status"] = "processing"
    await queue.put({"type": "status", "data": {"status": "processing", "progress": 0}})

    # 模拟处理步骤
    for step in range(1, 4):
        await asyncio.sleep(1)
        progress = step * 33
        await queue.put({
            "type": "progress",
            "data": {"status": "processing", "progress": progress, "step": step}
        })

    # 状态: completed
    result_content = {
        "type": "text",
        "data": f"任务完成。处理了来自 {message['sender_id']} 的请求。",
    }
    messages_db[msg_id]["status"] = "completed"
    messages_db[msg_id]["result"] = result_content
    messages_db[msg_id]["updated_at"] = datetime.utcnow().isoformat()

    await queue.put({
        "type": "completed",
        "data": {
            "status": "completed",
            "result": result_content,
            "completed_at": datetime.utcnow().isoformat(),
        }
    })

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8100)

# 1. 启动 ACP Server
python acp_server.py &

# 2. 列出所有 Agent
curl -s http://localhost:8100/agents | python -m json.tool

# 3. 获取指定 Agent 的能力
curl -s http://localhost:8100/agents/analyst-001/card | python -m json.tool

# 4. 发送消息（异步）
curl -s -X POST http://localhost:8100/messages \
  -H "Content-Type: application/json" \
  -d '{
    "sender_id": "coordinator-001",
    "receiver_id": "analyst-001",
    "content": {
      "type": "text",
      "data": "分析上季度销售数据，找出表现最差的产品线"
    },
    "priority": "high",
    "timeout_seconds": 120
  }' | python -m json.tool

# 5. 查询消息状态
curl -s http://localhost:8100/messages/{message_id} | python -m json.tool

# 6. SSE 流式接收处理进度
curl -N http://localhost:8100/messages/{message_id}/stream

"""
三协议融合协调 Agent
- A2A: 委托任务给专业 Agent
- MCP: 不直接使用（由专业 Agent 使用）
- ACP: 接收中间结果的 SSE 流
"""
import httpx
import asyncio
import json
from typing import Optional

class CoordinatorAgent:
    """协调 Agent：接收用户请求，分解并委托给专业 Agent"""

    def __init__(self):
        self.a2a_base = "http://localhost:8200"  # A2A 协调端点
        self.acp_base = "http://localhost:8100"   # ACP 消息端点

    async def handle_user_request(self, user_query: str) -> dict:
        """处理用户请求的完整流程"""

        # Step 1: 通过 A2A 委托数据分析任务
        analysis_task = await self._delegate_via_a2a(
            agent_id="analyst-001",
            task={
                "type": "data_analysis",
                "query": user_query,
                "output_format": "json",
            }
        )
        print(f"[A2A] 数据分析任务已委托: {analysis_task['task_id']}")

        # Step 2: 通过 ACP 发送消息通知报告 Agent 准备接收数据
        acp_msg = await self._notify_via_acp(
            receiver_id="writer-001",
            content=f"即将接收数据分析结果，请准备生成报告。用户查询: {user_query}"
        )
        print(f"[ACP] 报告 Agent 已通知: {acp_msg['message_id']}")

        # Step 3: 通过 ACP SSE 监听分析 Agent 的中间结果
        results = await self._stream_results_via_acp(
            message_id=analysis_task.get("acp_message_id", acp_msg["message_id"])
        )

        # Step 4: 将最终结果通过 A2A 委托报告 Agent
        report_task = await self._delegate_via_a2a(
            agent_id="writer-001",
            task={
                "type": "report_generation",
                "analysis_data": results,
                "format": "markdown",
            }
        )

        return {
            "analysis_task_id": analysis_task["task_id"],
            "report_task_id": report_task["task_id"],
            "status": "delegated",
        }

    async def _delegate_via_a2a(self, agent_id: str, task: dict) -> dict:
        """通过 A2A 协议委托任务"""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.a2a_base}/tasks",
                json={
                    "assignee": agent_id,
                    "task": task,
                    "priority": "high",
                }
            )
            return resp.json()

    async def _notify_via_acp(self, receiver_id: str, content: str) -> dict:
        """通过 ACP 协议发送通知消息"""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.acp_base}/messages",
                json={
                    "sender_id": "coordinator-001",
                    "receiver_id": receiver_id,
                    "content": {"type": "text", "data": content},
                    "priority": "normal",
                }
            )
            return resp.json()

    async def _stream_results_via_acp(self, message_id: str) -> list:
        """通过 ACP SSE 流式接收中间结果"""
        results = []
        async with httpx.AsyncClient(timeout=None) as client:
            async with client.stream(
                "GET",
                f"{self.acp_base}/messages/{message_id}/stream"
            ) as response:
                async for line in response.aiter_lines():
                    if line.startswith("data:"):
                        event = json.loads(line[5:].strip())
                        if event.get("type") == "progress":
                            print(f"  [ACP] 进度: {event['data']['progress']}%")
                        elif event.get("type") == "completed":
                            results.append(event["data"].get("result"))
                            break
        return results

# ─── 运行示例 ───

async def main():
    coordinator = CoordinatorAgent()
    result = await coordinator.handle_user_request(
        "分析上季度华东区销售数据，找出下滑最明显的产品线"
    )
    print(f"\n最终结果: {json.dumps(result, indent=2)}")

if __name__ == "__main__":
    asyncio.run(main())

/**
 * ACP TypeScript 客户端
 * 依赖: npm install eventsource-parser
 */
import EventSource from 'eventsource';

interface ACPMessage {
  message_id: string;
  thread_id: string;
  sender_id: string;
  receiver_id: string;
  content: {
    type: 'text' | 'image' | 'json' | 'embedding';
    data: string;
    metadata?: Record<string, unknown>;
  };
  status: 'pending' | 'processing' | 'completed' | 'failed';
  created_at: string;
}

interface AgentCard {
  agent_id: string;
  name: string;
  description: string;
  capabilities: string[];
  supported_content_types: string[];
  auth_required: boolean;
}

export class ACPClient {
  private baseUrl: string;

  constructor(baseUrl: string = 'http://localhost:8100') {
    this.baseUrl = baseUrl;
  }

  /** 列出所有可用 Agent */
  async listAgents(): Promise<AgentCard[]> {
    const res = await fetch(`${this.baseUrl}/agents`);
    const data = await res.json();
    return data.agents;
  }

  /** 获取指定 Agent 的能力描述 */
  async getAgentCard(agentId: string): Promise<AgentCard> {
    const res = await fetch(`${this.baseUrl}/agents/${agentId}/card`);
    if (!res.ok) throw new Error(`Agent ${agentId} not found`);
    return res.json();
  }

  /** 发送消息 */
  async sendMessage(params: {
    senderId: string;
    receiverId: string;
    content: { type: string; data: string };
    priority?: 'low' | 'normal' | 'high';
  }): Promise<{ message_id: string; thread_id: string; status: string }> {
    const res = await fetch(`${this.baseUrl}/messages`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        sender_id: params.senderId,
        receiver_id: params.receiverId,
        content: params.content,
        priority: params.priority ?? 'normal',
      }),
    });
    return res.json();
  }

  /** SSE 流式接收消息进度 */
  streamProgress(
    messageId: string,
    onEvent: (type: string, data: unknown) => void
  ): EventSource {
    const es = new EventSource(
      `${this.baseUrl}/messages/${messageId}/stream`
    );

    es.addEventListener('status', (e) => {
      onEvent('status', JSON.parse(e.data));
    });

    es.addEventListener('progress', (e) => {
      onEvent('progress', JSON.parse(e.data));
    });

    es.addEventListener('completed', (e) => {
      onEvent('completed', JSON.parse(e.data));
      es.close();
    });

    es.addEventListener('failed', (e) => {
      onEvent('failed', JSON.parse(e.data));
      es.close();
    });

    return es;
  }

  /** 完整流程：发送消息并等待结果 */
  async sendAndWait(params: {
    senderId: string;
    receiverId: string;
    content: { type: string; data: string };
  }): Promise<unknown> {
    // 发送消息
    const { message_id } = await this.sendMessage(params);
    console.log(`[ACP] 消息已发送: ${message_id}`);

    // 等待结果
    return new Promise((resolve, reject) => {
      this.streamProgress(message_id, (type, data) => {
        console.log(`[ACP] 事件: ${type}`, data);
        if (type === 'completed') resolve(data);
        if (type === 'failed') reject(new Error('Message processing failed'));
      });
    });
  }
}

// ─── 使用示例 ───
async function main() {
  const acp = new ACPClient('http://localhost:8100');

  // 列出 Agent
  const agents = await acp.listAgents();
  console.log('可用 Agent:', agents.map(a => a.name));

  // 发送消息并等待结果
  const result = await acp.sendAndWait({
    senderId: 'coordinator-001',
    receiverId: 'analyst-001',
    content: {
      type: 'text',
      data: '分析上季度华东区销售数据',
    },
  });
  console.log('分析结果:', result);
}

main().catch(console.error);

"""
ACP Agent 注册中心
支持离线发现、能力搜索、状态追踪
"""
from datetime import datetime
from typing import Optional
import json
import os

class ACPRegistry:
    """ACP Agent 注册中心（文件系统实现）"""

    def __init__(self, registry_dir: str = "./acp_registry"):
        self.registry_dir = registry_dir
        os.makedirs(registry_dir, exist_ok=True)

    def register(self, agent_card: dict) -> str:
        """注册 Agent Card 到注册中心"""
        agent_id = agent_card["agent_id"]
        card_path = os.path.join(self.registry_dir, f"{agent_id}.json")

        # 添加注册时间戳
        agent_card["registered_at"] = datetime.utcnow().isoformat()
        agent_card["last_seen"] = datetime.utcnow().isoformat()
        agent_card["availability"] = "online"

        with open(card_path, "w") as f:
            json.dump(agent_card, f, indent=2, ensure_ascii=False)

        return agent_id

    def discover(self, capabilities: list[str]) -> list[dict]:
        """按能力标签发现 Agent（支持离线 Agent）"""
        results = []
        for filename in os.listdir(self.registry_dir):
            if not filename.endswith(".json"):
                continue
            card_path = os.path.join(self.registry_dir, filename)
            with open(card_path) as f:
                card = json.load(f)

            # 检查能力匹配
            agent_caps = set(card.get("capabilities", []))
            required_caps = set(capabilities)
            if required_caps.issubset(agent_caps):
                results.append(card)

        return sorted(results, key=lambda c: c.get("registered_at", ""))

    def heartbeat(self, agent_id: str) -> bool:
        """更新 Agent 心跳"""
        card_path = os.path.join(self.registry_dir, f"{agent_id}.json")
        if not os.path.exists(card_path):
            return False

        with open(card_path) as f:
            card = json.load(f)

        card["last_seen"] = datetime.utcnow().isoformat()
        card["availability"] = "online"

        with open(card_path, "w") as f:
            json.dump(card, f, indent=2, ensure_ascii=False)

        return True

    def mark_offline(self, agent_id: str):
        """标记 Agent 为离线（Card 保留用于离线发现）"""
        card_path = os.path.join(self.registry_dir, f"{agent_id}.json")
        if not os.path.exists(card_path):
            return

        with open(card_path) as f:
            card = json.load(f)

        card["availability"] = "offline"

        with open(card_path, "w") as f:
            json.dump(card, f, indent=2, ensure_ascii=False)

# ─── 使用示例 ───

registry = ACPRegistry()

# 注册 Agent
registry.register({
    "agent_id": "analyst-001",
    "name": "数据分析 Agent",
    "description": "擅长 SQL 查询和数据可视化",
    "capabilities": ["sql_query", "data_visualization", "statistical_analysis"],
    "supported_content_types": ["text", "json"],
    "auth_required": True,
    "max_concurrent_messages": 10,
})

# 离线发现：查找具有 sql_query 能力的 Agent
results = registry.discover(["sql_query"])
print(f"找到 {len(results)} 个匹配的 Agent")
for agent in results:
    print(f"  - {agent['name']} (状态: {agent['availability']})")