Multi-Agent Communication Protocols — Deep Dive

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel
import asyncio
import json
import uuid
from datetime import datetime

app = FastAPI()

# ── Agent Card ──────────────────────────────────────────────────────────────
AGENT_CARD = {
    "name": "ResearchAgent",
    "description": "Searches the web and summarizes findings for a given query",
    "url": "https://research-agent.example.com",
    "version": "1.0.0",
    "capabilities": {
        "streaming": True,
        "pushNotifications": False,
        "stateTransitionHistory": True,
    },
    "defaultInputModes": ["text/plain"],
    "defaultOutputModes": ["text/plain", "application/json"],
    "skills": [
        {
            "id": "web-research",
            "name": "Web Research",
            "description": "Search the web and synthesize findings",
            "inputModes": ["text/plain"],
            "outputModes": ["text/plain"],
        }
    ],
    "authentication": {
        "schemes": ["Bearer"]
    },
}

@app.get("/.well-known/agent.json")
async def agent_card():
    return JSONResponse(AGENT_CARD)

# ── Task Models ──────────────────────────────────────────────────────────────
class Message(BaseModel):
    role: str  # "user" | "agent"
    parts: list[dict]

class TaskRequest(BaseModel):
    id: str | None = None
    message: Message
    sessionId: str | None = None

# In-memory task store (use Redis in production)
tasks: dict = {}

# ── Task Submission (non-streaming) ─────────────────────────────────────────
@app.post("/tasks/send")
async def send_task(task_req: TaskRequest):
    task_id = task_req.id or str(uuid.uuid4())
    query = task_req.message.parts[0].get("text", "")

    tasks[task_id] = {
        "id": task_id,
        "status": {"state": "working"},
        "history": [{"role": "user", "parts": task_req.message.parts}],
    }

    # Run research (simplified)
    result = await run_research(query)

    tasks[task_id]["status"]["state"] = "completed"
    tasks[task_id]["artifacts"] = [
        {"name": "summary", "parts": [{"type": "text", "text": result}]}
    ]

    return tasks[task_id]

# ── Task Streaming (SSE) ─────────────────────────────────────────────────────
@app.post("/tasks/sendSubscribe")
async def send_task_subscribe(task_req: TaskRequest):
    task_id = task_req.id or str(uuid.uuid4())
    query = task_req.message.parts[0].get("text", "")

    async def event_stream():
        # State: working
        yield f"data: {json.dumps({'id': task_id, 'status': {'state': 'working'}, 'final': False})}\n\n"

        # Stream intermediate results
        async for chunk in stream_research(query):
            yield f"data: {json.dumps({'id': task_id, 'artifact': {'parts': [{'type': 'text', 'text': chunk}]}, 'final': False})}\n\n"
            await asyncio.sleep(0.1)

        # State: completed
        yield f"data: {json.dumps({'id': task_id, 'status': {'state': 'completed'}, 'final': True})}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

async def run_research(query: str) -> str:
    await asyncio.sleep(0.5)  # simulate web search
    return f"Research findings for: {query} — [synthesized summary here]"

async def stream_research(query: str):
    steps = [f"Searching for '{query}'...", "Analyzing results...", "Synthesizing summary..."]
    for step in steps:
        yield step
        await asyncio.sleep(0.3)

import httpx
import json
import asyncio

class A2AClient:
    def __init__(self, agent_url: str, api_key: str | None = None):
        self.agent_url = agent_url.rstrip("/")
        self.headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}

    async def discover(self) -> dict:
        """Fetch and validate the remote agent's Agent Card."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"{self.agent_url}/.well-known/agent.json")
            resp.raise_for_status()
            card = resp.json()
            print(f"Connected to: {card['name']} v{card['version']}")
            print(f"Streaming: {card['capabilities'].get('streaming', False)}")
            return card

    async def send_task(self, message: str, task_id: str | None = None) -> dict:
        """Send a task and wait for completion."""
        payload = {
            "id": task_id,
            "message": {"role": "user", "parts": [{"type": "text", "text": message}]},
        }
        async with httpx.AsyncClient(timeout=60) as client:
            resp = await client.post(
                f"{self.agent_url}/tasks/send",
                json=payload,
                headers=self.headers,
            )
            resp.raise_for_status()
            return resp.json()

    async def send_task_streaming(self, message: str) -> None:
        """Send a task and stream results via SSE."""
        payload = {
            "message": {"role": "user", "parts": [{"type": "text", "text": message}]},
        }
        async with httpx.AsyncClient(timeout=120) as client:
            async with client.stream(
                "POST",
                f"{self.agent_url}/tasks/sendSubscribe",
                json=payload,
                headers=self.headers,
            ) as resp:
                async for line in resp.aiter_lines():
                    if line.startswith("data: "):
                        event = json.loads(line[6:])
                        if "artifact" in event:
                            print(f"[chunk] {event['artifact']['parts'][0]['text']}")
                        if event.get("final"):
                            print("[done]")
                            break

async def main():
    client = A2AClient("https://research-agent.example.com", api_key="sk-...")
    await client.discover()
    await client.send_task_streaming("Latest advances in multi-agent systems 2025")

asyncio.run(main())

# ACP Agent using the beeai-framework pattern
from dataclasses import dataclass
from typing import AsyncIterator
import httpx
import json

@dataclass
class ACPMessage:
    """ACP message with JSON-LD typed parts."""
    parts: list[dict]

    def text_content(self) -> str:
        for part in self.parts:
            if part.get("content_type") == "text/plain":
                return part.get("content", "")
        return ""

    def to_dict(self) -> dict:
        return {"parts": self.parts}

    @classmethod
    def from_text(cls, text: str) -> "ACPMessage":
        return cls(parts=[{
            "content_type": "text/plain",
            "content": text,
        }])

    @classmethod
    def from_structured(cls, data: dict, schema_type: str) -> "ACPMessage":
        return cls(parts=[
            {"content_type": "text/plain", "content": str(data)},
            {
                "content_type": "application/json",
                "content": json.dumps(data),
                "@type": schema_type,
            }
        ])

class ACPAgentClient:
    """Client for interacting with an ACP-compliant agent server."""

    def __init__(self, base_url: str, agent_name: str, api_key: str | None = None):
        self.base_url = base_url.rstrip("/")
        self.agent_name = agent_name
        self.headers = {}
        if api_key:
            self.headers["Authorization"] = f"Bearer {api_key}"

    async def run(self, input_message: ACPMessage) -> ACPMessage:
        """Create a run and wait for synchronous completion."""
        async with httpx.AsyncClient(timeout=60) as client:
            # Create run
            resp = await client.post(
                f"{self.base_url}/agents/{self.agent_name}/runs",
                json={"input": [input_message.to_dict()]},
                headers=self.headers,
            )
            resp.raise_for_status()
            run = resp.json()

            # Poll until done (for async runs)
            while run.get("status") not in ("completed", "failed"):
                await __import__("asyncio").sleep(1)
                poll = await client.get(
                    f"{self.base_url}/agents/{self.agent_name}/runs/{run['run_id']}",
                    headers=self.headers,
                )
                run = poll.json()

            if run["status"] == "failed":
                raise RuntimeError(f"Agent run failed: {run.get('error')}")

            return ACPMessage(parts=run["output"][0]["parts"])

    async def run_stream(self, input_message: ACPMessage) -> AsyncIterator[str]:
        """Stream output from an ACP run."""
        async with httpx.AsyncClient(timeout=120) as client:
            async with client.stream(
                "POST",
                f"{self.base_url}/agents/{self.agent_name}/runs/stream",
                json={"input": [input_message.to_dict()]},
                headers=self.headers,
            ) as resp:
                async for line in resp.aiter_lines():
                    if line.startswith("data:"):
                        data = json.loads(line[5:])
                        if chunk := data.get("chunk"):
                            for part in chunk.get("delta", {}).get("parts", []):
                                if part.get("content_type") == "text/plain":
                                    yield part["content"]

# Usage
async def acp_example():
    client = ACPAgentClient("https://acp-server.example.com", "research-agent")
    msg = ACPMessage.from_text("Summarize recent breakthroughs in quantum computing")
    async for chunk in client.run_stream(msg):
        print(chunk, end="", flush=True)

import json
import hashlib
import base64
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.serialization import (
    Encoding, PublicFormat, PrivateFormat, NoEncryption
)
from datetime import datetime, timezone
import httpx

class ANPAgent:
    """Agent with ANP-compliant DID-based identity."""

    def __init__(self, domain: str):
        self.domain = domain
        self.did = f"did:web:{domain}"
        # Generate Ed25519 key pair
        self._private_key = Ed25519PrivateKey.generate()
        self._public_key = self._private_key.public_key()

    @property
    def public_key_bytes(self) -> bytes:
        return self._public_key.public_bytes(Encoding.Raw, PublicFormat.Raw)

    @property
    def public_key_multibase(self) -> str:
        """Multibase-encoded public key (base58btc prefix 'z')."""
        import base58
        return "z" + base58.b58encode(self.public_key_bytes).decode()

    def did_document(self) -> dict:
        """Generate the DID Document served at /.well-known/did.json"""
        key_id = f"{self.did}#key-1"
        return {
            "@context": [
                "https://www.w3.org/ns/did/v1",
                "https://w3id.org/security/suites/ed25519-2020/v1",
            ],
            "id": self.did,
            "verificationMethod": [{
                "id": key_id,
                "type": "Ed25519VerificationKey2020",
                "controller": self.did,
                "publicKeyMultibase": self.public_key_multibase,
            }],
            "authentication": [key_id],
            "service": [{
                "id": f"{self.did}#anp-service",
                "type": "ANPService",
                "serviceEndpoint": f"https://{self.domain}/anp",
            }],
        }

    def sign_message(self, message: dict) -> dict:
        """Add a proof to a message using Ed25519 signature."""
        message_bytes = json.dumps(message, sort_keys=True).encode()
        signature = self._private_key.sign(message_bytes)
        signature_b64 = base64.urlsafe_b64encode(signature).decode()

        return {
            **message,
            "proof": {
                "type": "Ed25519Signature2020",
                "created": datetime.now(timezone.utc).isoformat(),
                "verificationMethod": f"{self.did}#key-1",
                "proofPurpose": "authentication",
                "proofValue": signature_b64,
            }
        }

    async def send_message(self, target_did: str, content: str) -> dict:
        """Send a signed ANP message to another agent."""
        # Resolve target DID to get endpoint
        target_domain = target_did.replace("did:web:", "")
        async with httpx.AsyncClient() as client:
            did_doc = (await client.get(f"https://{target_domain}/.well-known/did.json")).json()

        endpoint = next(
            s["serviceEndpoint"] for s in did_doc["service"]
            if s["type"] == "ANPService"
        )

        message = {
            "from": self.did,
            "to": target_did,
            "type": "https://anp.example.com/message",
            "body": {"content": content},
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
        signed = self.sign_message(message)

        async with httpx.AsyncClient() as client:
            resp = await client.post(f"{endpoint}/messages", json=signed)
            resp.raise_for_status()
            return resp.json()

# Usage
async def anp_example():
    alice = ANPAgent("alice.example.com")
    result = await alice.send_message(
        target_did="did:web:bob.example.com",
        content="Request: analyze this dataset and return anomalies",
    )
    print(result)

from enum import Enum
from dataclasses import dataclass
import httpx

class AgentProtocol(Enum):
    A2A = "a2a"
    ACP = "acp"
    ANP = "anp"
    UNKNOWN = "unknown"

@dataclass
class NormalizedMessage:
    """Protocol-agnostic internal message format."""
    sender_id: str
    content: str
    metadata: dict

async def detect_protocol(agent_url: str) -> AgentProtocol:
    """Probe an agent URL to detect which protocol it speaks."""
    async with httpx.AsyncClient(timeout=5) as client:
        try:
            resp = await client.get(f"{agent_url}/.well-known/agent.json")
            if resp.status_code == 200 and "skills" in resp.json():
                return AgentProtocol.A2A
        except Exception:
            pass
        try:
            resp = await client.get(f"{agent_url}/.well-known/did.json")
            if resp.status_code == 200 and "@context" in resp.json():
                return AgentProtocol.ANP
        except Exception:
            pass
        try:
            resp = await client.get(f"{agent_url}/agents")
            if resp.status_code == 200:
                return AgentProtocol.ACP
        except Exception:
            pass
    return AgentProtocol.UNKNOWN

async def route_message(
    target_url: str,
    message: NormalizedMessage,
    protocol: AgentProtocol | None = None,
) -> str:
    """Route a normalized message to a target agent, auto-detecting protocol."""
    if protocol is None:
        protocol = await detect_protocol(target_url)

    if protocol == AgentProtocol.A2A:
        return await send_via_a2a(target_url, message)
    elif protocol == AgentProtocol.ACP:
        return await send_via_acp(target_url, message)
    elif protocol == AgentProtocol.ANP:
        return await send_via_anp(target_url, message)
    else:
        raise ValueError(f"Could not detect protocol for {target_url}")

async def send_via_a2a(url: str, msg: NormalizedMessage) -> str:
    async with httpx.AsyncClient(timeout=60) as client:
        resp = await client.post(f"{url}/tasks/send", json={
            "message": {"role": "user", "parts": [{"type": "text", "text": msg.content}]}
        })
        result = resp.json()
        return result["artifacts"][0]["parts"][0]["text"]

async def send_via_acp(url: str, msg: NormalizedMessage) -> str:
    async with httpx.AsyncClient(timeout=60) as client:
        resp = await client.post(f"{url}/agents/default/runs", json={
            "input": [{"parts": [{"content_type": "text/plain", "content": msg.content}]}]
        })
        run = resp.json()
        return run["output"][0]["parts"][0]["content"]

async def send_via_anp(url: str, msg: NormalizedMessage) -> str:
    domain = url.replace("https://", "").split("/")[0]
    async with httpx.AsyncClient(timeout=60) as client:
        resp = await client.post(f"{url}/anp/messages", json={
            "from": msg.sender_id,
            "to": f"did:web:{domain}",
            "body": {"content": msg.content},
        })
        return resp.json().get("response", "")

# Example: send to three different agents without caring about their protocol
async def orchestrate():
    msg = NormalizedMessage(
        sender_id="did:web:orchestrator.example.com",
        content="Analyze Q1 sales data and identify top 3 anomalies",
        metadata={},
    )
    results = {}
    for name, url in [
        ("analyst", "https://analyst-agent.example.com"),    # speaks A2A
        ("auditor", "https://auditor-agent.enterprise.com"), # speaks ACP
        ("verifier", "https://verifier.defi.example.com"),   # speaks ANP
    ]:
        results[name] = await route_message(url, msg)
    return results