a2a-protocol-guide

Guide du Protocole A2A

Concepts fondamentaux

Composant	Rôle
Agent Card	Fichier JSON décrivant l'agent (skills, endpoint, auth) — publié sur /.well-known/agent.json
Task	Unité de travail délégué ; cycle : submitted → working → completed / failed / canceled
Artifact	Résultat produit par une tâche (fichier, JSON, message)
Part	Fragment d'un message ou artifact : TextPart, FilePart, DataPart
Client / Server	Un agent peut être les deux selon la direction de l'appel

---

Workflow en étapes

1. Définir l'Agent Card

Fichier minimal exposé sur GET /.well-known/agent.json :

{
  "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"]
    }
  ]
}

Critère de décision — streaming : active streaming: true si les tâches durent > 5 s ou produisent des résultats progressifs. Sinon, mode synchrone suffit.

---

2. Implémenter le serveur A2A (JSON-RPC 2.0)

Endpoint unique POST /a2a — méthodes requises :

Méthode	Usage
tasks/send	Tâche synchrone (réponse directe)
tasks/sendSubscribe	Tâche longue — SSE stream
tasks/get	Consulter statut + artifacts
tasks/cancel	Annuler une tâche en cours

# 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"}})