Exposer un agent ADK en serveur A2A

Installer l'extra A2A

Le support A2A d'ADK est une dépendance optionnelle. Dans ADK 2.10, pip install "google-adk[a2a]==2.10.0" installe a2a-sdk[http-server] (version 1.2.2 au moment du test). Sans cet extra, l'import échoue avec ModuleNotFoundError: No module named 'a2a'.

Un agent ordinaire d'abord

L'agent stock est un agent ADK classique avec un outil. Rien dans son code ne parle d'A2A :

def check_stock(sku: str) -> dict:
    """Retourne le stock disponible d'un article, par entrepôt.

    Args:
        sku: référence de l'article, par exemple "CHAISE-ERGO".
    """
    item = STOCK.get(sku.strip().upper())
    if item is None:
        return {"status": "error", "message": f"Référence inconnue : {sku}",
                "references_connues": sorted(STOCK)}
    entrepots = {k: v for k, v in item.items() if k != "nom"}
    return {"status": "ok", "sku": sku.upper(), "nom": item["nom"],
            "par_entrepot": entrepots, "total": sum(entrepots.values())}

root_agent = Agent(
    name="stock_agent", model=MODEL,
    description="Donne le stock disponible d'un article (par entrepôt) à partir de sa référence.",
    instruction="Tu es le service stock. Pour toute question de disponibilité, appelle check_stock. ...",
    tools=[check_stock],
)

app = serve(root_agent, port=int(os.environ.get("STOCK_PORT", 8101)))

L'outil renvoie une erreur exploitable (references_connues) plutôt qu'une exception : le modèle peut corriger une référence mal saisie au lieu d'échouer.

Ce que fait to_a2a

flowchart TB
  A["Agent ADK (root_agent)"] --> R["Runner + services en mémoire (sessions, artefacts, mémoire)"]
  R --> X["A2aAgentExecutor : traduit A2A vers événements ADK"]
  A --> C["AgentCardBuilder : carte depuis nom, description, outils"]
  X --> H["DefaultRequestHandler + task store"]
  C --> W["Route GET /.well-known/agent-card.json"]
  H --> J["Route POST / (JSON-RPC)"]
  W --> S["Application Starlette"]
  J --> S

to_a2a(agent, host, port, protocol) renvoie une application Starlette (le micro-framework web sous FastAPI) prête pour uvicorn. Les routes sont ajoutées au démarrage (lifespan) : c'est là que la carte est construite, avec l'URL protocol://host:port/ qu'elle annoncera.

Le lab enveloppe cet appel dans serve() pour ajouter l'authentification (module 3) et l'URL publique (module 4) :

def serve(agent, port: int, token: str | None = None):
    """Transforme un agent ADK en application A2A (Starlette) protégée par jeton."""
    # Derrière Cloud Run : A2A_HOST=<service>.run.app, A2A_PROTOCOL=https, A2A_PUBLIC_PORT=443
    app = to_a2a(agent, host=HOST, port=int(os.environ.get("A2A_PUBLIC_PORT", port)),
                 protocol=os.environ.get("A2A_PROTOCOL", "http"))
    app.add_middleware(BearerAuth, token=token or a2a_token())
    return app

L'agent tarification, même recette

pricing_agent.py calcule un prix avec des paliers de remise. Le calcul est en Python, jamais confié au modèle :

PALIERS = [(20, 0.15), (10, 0.10), (5, 0.05)]

def quote_price(sku: str, quantity: int) -> dict:
    ...
    remise = next((r for seuil, r in PALIERS if quantity >= seuil), 0.0)
    total = round(PRIX_UNITAIRE[sku] * quantity * (1 - remise), 2)

Puis on lance chaque agent dans son terminal :

export A2A_TOKEN=$(python -c "import secrets; print(secrets.token_urlsafe(24))")
uvicorn stock_agent:app --port 8101
uvicorn pricing_agent:app --port 8102

Résultat obtenu

Tests sans clé du lab, le 2026-10-15 (pytest -q dans files/) :

.............                                                            [100%]
13 passed in 0.93s

Parmi eux, test_agent_card_publique démarre l'application A2A avec le TestClient de Starlette (ce qui exécute le lifespan et construit la carte) et vérifie le nom, l'URL http://localhost:8101 et la compétence check_stock, sans aucun appel au modèle.

Exercice : un troisième spécialiste

Écrivez delivery_agent.py : un outil delivery_delay(city: str) -> dict (délais fictifs : Lyon 2 jours, Lille 3 jours, autres 5 jours) et un serveur A2A sur le port 8103. Vérifiez sa carte.

import os
from google.adk.agents import Agent
from common import MODEL, serve

DELAIS = {"lyon": 2, "lille": 3}

def delivery_delay(city: str) -> dict:
    """Retourne le délai de livraison estimé, en jours ouvrés, pour une ville.

    Args:
        city: ville de livraison, par exemple "Lyon".
    """
    return {"status": "ok", "ville": city, "jours_ouvres": DELAIS.get(city.strip().lower(), 5)}

root_agent = Agent(name="delivery_agent", model=MODEL,
                   description="Estime le délai de livraison d'une commande selon la ville.",
                   instruction="Tu es le service livraison. Appelle delivery_delay et réponds en une phrase.",
                   tools=[delivery_delay])
app = serve(root_agent, port=int(os.environ.get("DELIVERY_PORT", 8103)))

Je sais faire

• Installer l'extra A2A d'ADK et épingler les versions
• Exposer un agent avec to_a2a et le lancer avec uvicorn
• Expliquer la différence entre URL annoncée et port d'écoute
• Garder les règles métier dans des outils testables