khalilbenaz/claude-skills-collection/agent-a2a-protocol-guide

{
  "name": "DataExtractorAgent",
  "description": "Extrait des données structurées depuis des documents PDF",
  "url": "https://agent.example.com/a2a",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "authentication": {
    "schemes": ["Bearer"]
  },
  "skills": [
    {
      "id": "extract-invoice",
      "name": "Invoice Extraction",
      "description": "Retourne les champs clés d'une facture PDF",
      "examples": ["Extrais les montants de cette facture"],
      "inputModes": ["application/pdf"],
      "outputModes": ["application/json"]
    }
  ]
}

# FastAPI — squelette minimal
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse
import uuid, asyncio
app = FastAPI()
tasks_store = {}  # En prod : Redis ou DB
@app.post("/a2a")
async def a2a_handler(request: Request):
    body = await request.json()
    method = body.get("method")
    params = body.get("params", {})
    rpc_id = body.get("id")
    if method == "tasks/send":
        task_id = str(uuid.uuid4())
        result = await run_task(params)
        tasks_store[task_id] = {"status": "completed", "result": result}
        return JSONResponse({
            "jsonrpc": "2.0", "id": rpc_id,
            "result": {"id": task_id, "status": {"state": "completed"}, "artifacts": [result]}
        })
    if method == "tasks/get":
        task_id = params["id"]
        task = tasks_store.get(task_id)
        if not task:
            return JSONResponse({"jsonrpc": "2.0", "id": rpc_id,
                                 "error": {"code": -32001, "message": "Task not found"}})
        return JSONResponse({"jsonrpc": "2.0", "id": rpc_id, "result": task})
    return JSONResponse({"jsonrpc": "2.0", "id": rpc_id,
                         "error": {"code": -32601, "message": "Method not found"}})

import httpx, json
async def send_task(agent_url: str, message: str, token: str) -> dict:
    payload = {
        "jsonrpc": "2.0", "id": "1", "method": "tasks/send",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": message}]
            }
        }
    }
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            agent_url,
            json=payload,
            headers={"Authorization": f"Bearer {token}"},
            timeout=30
        )
        resp.raise_for_status()
        return resp.json()["result"]

from fastapi.responses import StreamingResponse
@app.post("/a2a/stream")
async def stream_task(request: Request):
    body = await request.json()
    async def event_generator():
        # Mise à jour intermédiaire
        yield f"data: {json.dumps({'type': 'TaskStatusUpdateEvent', 'status': {'state': 'working'}})}\n\n"
        await asyncio.sleep(1)
        # Résultat final
        artifact = {"type": "DataPart", "data": {"key": "value"}}
        yield f"data: {json.dumps({'type': 'TaskArtifactUpdateEvent', 'artifact': artifact})}\n\n"
        yield f"data: {json.dumps({'type': 'TaskStatusUpdateEvent', 'status': {'state': 'completed'}})}\n\n"
    return StreamingResponse(event_generator(), media_type="text/event-stream")

async def discover_agent(base_url: str) -> dict:
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{base_url}/.well-known/agent.json", timeout=5)
        r.raise_for_status()
        return r.json()
# Critère de sélection
def select_agent(agents: list[dict], required_skill: str) -> dict | None:
    for a in agents:
        skills = [s["id"] for s in a.get("skills", [])]
        if required_skill in skills:
            return a
    return None

# Middleware auth minimal
from fastapi import Depends, HTTPException, Header
async def verify_token(authorization: str = Header(...)):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401)
    token = authorization[7:]
    if token not in ALLOWED_TOKENS:
        raise HTTPException(status_code=403)

async def orchestrate(agents: list[str], prompt: str) -> list:
    tasks = [send_task(url, prompt, TOKEN) for url in agents]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [r for r in results if not isinstance(r, Exception)]