A2A — Agent-to-Agent Protocol

{
  "name": "Research Assistant Agent",
  "description": "Searches the web and synthesizes research reports on any topic",
  "url": "https://research-agent.example.com",
  "version": "1.0.0",
  "provider": {
    "organization": "Acme AI",
    "url": "https://acme.ai"
  },
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  "authentication": {
    "schemes": ["Bearer"]
  },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain", "application/json"],
  "skills": [
    {
      "id": "research-topic",
      "name": "Research a Topic",
      "description": "Search the web and produce a structured research report",
      "tags": ["research", "web-search", "summarization"],
      "examples": [
        "Research the current state of quantum computing",
        "Find recent developments in agentic AI frameworks"
      ],
      "inputModes": ["text/plain"],
      "outputModes": ["text/plain", "application/json"]
    },
    {
      "id": "fact-check",
      "name": "Fact Check a Claim",
      "description": "Verify a claim against authoritative sources",
      "tags": ["fact-checking", "verification"],
      "inputModes": ["text/plain"],
      "outputModes": ["text/plain"]
    }
  ]
}

# a2a_server.py — A minimal A2A-compliant agent server
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import Optional
import asyncio, uuid, json
from anthropic import Anthropic

app = FastAPI()
client = Anthropic()

# In-memory task store (use a database in production)
tasks: dict[str, dict] = {}

# -------------------------------------------------------
# A2A Data Models
# -------------------------------------------------------
class TextPart(BaseModel):
    type: str = "text"
    text: str

class Message(BaseModel):
    role: str  # "user" or "agent"
    parts: list[TextPart]

class TaskSendRequest(BaseModel):
    id: str
    message: Message
    sessionId: Optional[str] = None

class Artifact(BaseModel):
    name: str
    parts: list[TextPart]

# -------------------------------------------------------
# Agent Card endpoint (capability discovery)
# -------------------------------------------------------
@app.get("/.well-known/agent.json")
async def agent_card():
    return {
        "name": "Research Assistant",
        "description": "Researches topics and writes structured reports",
        "url": "http://localhost:8000",
        "version": "1.0.0",
        "capabilities": {"streaming": True, "pushNotifications": False},
        "skills": [
            {
                "id": "research",
                "name": "Research a Topic",
                "description": "Search and synthesize information on any topic",
                "tags": ["research", "web-search"],
            }
        ],
    }

# -------------------------------------------------------
# Task submission endpoint
# -------------------------------------------------------
@app.post("/tasks/send")
async def send_task(request: TaskSendRequest, background_tasks: BackgroundTasks):
    task_id = request.id or str(uuid.uuid4())
    user_message = request.message.parts[0].text

    # Initialize task
    tasks[task_id] = {
        "id": task_id,
        "status": {"state": "submitted"},
        "messages": [{"role": "user", "parts": [{"type": "text", "text": user_message}]}],
        "artifacts": [],
    }

    # Process in background
    background_tasks.add_task(run_agent, task_id, user_message)
    return tasks[task_id]

async def run_agent(task_id: str, user_message: str):
    tasks[task_id]["status"]["state"] = "working"

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system="You are a research assistant. Provide comprehensive, well-structured answers.",
        messages=[{"role": "user", "content": user_message}],
    )

    result = response.content[0].text

    # Add agent response to task
    tasks[task_id]["messages"].append({
        "role": "agent",
        "parts": [{"type": "text", "text": result}]
    })
    tasks[task_id]["artifacts"].append({
        "name": "research-report",
        "parts": [{"type": "text", "text": result}]
    })
    tasks[task_id]["status"]["state"] = "completed"

# -------------------------------------------------------
# Task polling and streaming endpoints
# -------------------------------------------------------
@app.get("/tasks/{task_id}")
async def get_task(task_id: str):
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail="Task not found")
    return tasks[task_id]

@app.get("/tasks/{task_id}/stream")
async def stream_task(task_id: str):
    async def event_generator():
        last_state = None
        for _ in range(30):  # poll for up to 30 seconds
            task = tasks.get(task_id, {})
            state = task.get("status", {}).get("state")
            if state != last_state:
                yield f"data: {json.dumps(task)}\n\n"
                last_state = state
            if state in ("completed", "failed", "canceled"):
                break
            await asyncio.sleep(1)

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

# Run: uvicorn a2a_server:app --reload

# a2a_client.py — A client that discovers and delegates to an A2A agent
import httpx, asyncio, time

A2A_AGENT_URL = "http://localhost:8000"

async def discover_agent(url: str) -> dict:
    """Fetch the Agent Card to understand the agent's capabilities."""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{url}/.well-known/agent.json")
        response.raise_for_status()
        card = response.json()
        print(f"Discovered agent: {card['name']}")
        print(f"Skills: {[s['name'] for s in card['skills']]}")
        return card

async def send_task(url: str, task_id: str, message: str) -> dict:
    """Send a task to the remote agent."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{url}/tasks/send",
            json={
                "id": task_id,
                "message": {
                    "role": "user",
                    "parts": [{"type": "text", "text": message}]
                }
            }
        )
        response.raise_for_status()
        return response.json()

async def poll_task(url: str, task_id: str, timeout: int = 60) -> dict:
    """Poll until the task completes."""
    async with httpx.AsyncClient() as client:
        start = time.time()
        while time.time() - start < timeout:
            response = await client.get(f"{url}/tasks/{task_id}")
            task = response.json()
            state = task["status"]["state"]
            print(f"  Task state: {state}")
            if state == "completed":
                return task
            if state in ("failed", "canceled"):
                raise RuntimeError(f"Task {state}")
            await asyncio.sleep(2)
    raise TimeoutError("Task did not complete in time")

async def main():
    import uuid

    # Step 1: Discover the remote agent
    card = await discover_agent(A2A_AGENT_URL)

    # Step 2: Send a task (delegate work to the remote agent)
    task_id = str(uuid.uuid4())
    print(f"\nSending task {task_id}...")
    await send_task(
        A2A_AGENT_URL,
        task_id,
        "Research the key differences between A2A, ACP, and MCP protocols for AI agents"
    )

    # Step 3: Wait for the result
    print("Polling for completion...")
    task = await poll_task(A2A_AGENT_URL, task_id)

    # Step 4: Extract the artifact
    for artifact in task.get("artifacts", []):
        print(f"\n=== {artifact['name']} ===")
        for part in artifact["parts"]:
            print(part["text"])

asyncio.run(main())