Atelier

Lección 3 — Construire pas à pas

L'agent commandes et le test sur 2 ports

5 min de lectura

Este contenido aún no está traducido. Se muestra la versión original.

L'agent commandes et le test sur 2 ports

Un agent distant, vu comme un agent local

RemoteA2aAgent est un agent ADK dont l'exécution se fait ailleurs : il lit la carte, envoie un message A2A et transforme la tâche reçue en événements ADK.

files/commandes_agent.py
STOCK_URL = os.environ.get("STOCK_URL", "http://localhost:8001").rstrip("/")

agent_stock_distant = RemoteA2aAgent(
    name="agent_stock",
    description="Agent distant (A2A) qui connaît la disponibilité et le prix unitaire des références.",
    agent_card=f"{STOCK_URL}/.well-known/agent-card.json",
    httpx_client=client_http(STOCK_URL),
)

Outil ou sous-agent ?

On pourrait le placer dans sub_agents=[...]. Mais un sous-agent ADK reçoit le contrôle de la conversation par transfert : c'est lui qui répondrait au commercial, et agent_commandes ne calculerait pas le devis. On l'enveloppe donc dans un AgentTool : l'agent distant est appelé comme une fonction, puis agent_commandes reprend la main.

files/commandes_agent.py
root_agent = Agent(
    name="agent_commandes",
    model=modele(),
    instruction=(
        "Tu prépares des devis pour une entreprise fictive, en français. "
        "Pour connaître la disponibilité et le prix unitaire d'une référence, appelle l'outil agent_stock (un agent distant). "
        "Ensuite, si la quantité demandée est disponible, appelle calculer_devis et donne HT, TVA et TTC ; "
        "sinon, indique la quantité disponible et propose une livraison partielle. N'invente aucun prix."
    ),
    tools=[calculer_devis, AgentTool(agent=agent_stock_distant)],
)
sub_agents=[distant]tools=[AgentTool(distant)]
Qui répond à l'utilisateur ?le sous-agent, après transfertl'agent principal
Enchaîner avec un autre outilnon, l'agent principal a passé la mainoui (ici calculer_devis)
Bon pouraiguiller vers un spécialisteconsulter un spécialiste et continuer

Avec sub_agents=[agent_stock_distant], que risque-t-on sur la demande « devis pour 500 VIS-M6 » ?

Le client HTTP : prêt pour Cloud Run

files/commandes_agent.py
class JetonIdentite(httpx.Auth):
    def __init__(self, audience: str):
        self.audience, self._jeton, self._expire = audience, None, 0.0

    def auth_flow(self, request):
        if time.time() > self._expire:
            self._jeton = id_token.fetch_id_token(google.auth.transport.requests.Request(), self.audience)
            self._expire = time.time() + 45 * 60  # les jetons d'identité vivent 1 h
        request.headers["Authorization"] = f"Bearer {self._jeton}"
        yield request

def client_http(url: str) -> httpx.AsyncClient:
    auth = JetonIdentite(url) if url.startswith("https://") else None
    return httpx.AsyncClient(auth=auth, timeout=120)

En local (http://), aucun jeton. Sur Cloud Run (https://), chaque requête porte un jeton d'identité dont l'audience est l'URL du service stock. On le détaille au module 3.

Le test sur deux ports

sequenceDiagram
  participant T as "client_a2a.py"
  participant C as "commandes :8002"
  participant S as "stock :8001"
  T->>C: GET carte puis SendMessage("devis 500 VIS-M6")
  C->>S: GET /.well-known/agent-card.json
  C->>S: POST / SendMessage("stock et prix VIS-M6 ?")
  S-->>C: Task COMPLETED (1200 dispo, 0.08 €)
  C->>C: calculer_devis(500, 0.08)
  C-->>T: Task COMPLETED (HT, TVA, TTC)
terminal 1
uvicorn stock_agent:app --port 8001
terminal 2
STOCK_URL=http://localhost:8001 uvicorn commandes_agent:app --port 8002
terminal 3
python client_a2a.py http://localhost:8002 "Je voudrais un devis pour 500 VIS-M6."
python client_a2a.py http://localhost:8002 "Devis pour 100 rondelles RON-M6 ?"

client_a2a.py lit la carte, envoie SendMessage à l'URL annoncée, puis extrait l'état, les outils appelés et le texte des artefacts.

Résultat obtenu

Sorties réelles du 2026-10-15 (Vertex AI, gemini-2.5-flash) :

terminal 3 (2026-10-15)
AGENT   : agent_commandes
ETAT    : TASK_STATE_COMPLETED
OUTILS  : ['agent_stock', 'calculer_devis']
REPONSE : Voici votre devis pour 500 VIS-M6 :
*   Prix unitaire HT : 0.08€
*   Total HT : 40.00€
*   TVA (20%) : 8.00€
*   Total TTC : 48.00€
AGENT   : agent_commandes
ETAT    : TASK_STATE_COMPLETED
OUTILS  : ['agent_stock']
REPONSE : Il n'y a pas de rondelles RON-M6 en stock. Le prix unitaire est de 0.02€ HT.

Et le journal du serveur stock prouve que l'appel est bien passé par le réseau, en A2A :

terminal 1 (2026-10-15)
INFO:     127.0.0.1:48030 - "GET /.well-known/agent-card.json HTTP/1.1" 200 OK
INFO:     127.0.0.1:48030 - "POST / HTTP/1.1" 200 OK

Les tests sans clé (pytest -q) : 6 passed, 6 warnings in 0.94s (les avertissements sont les [EXPERIMENTAL] d'ADK A2A). Ils couvrent les deux outils, la carte servie par l'app Starlette, le choix « jeton seulement vers https », la lecture d'une tâche par le client et la structure de l'agent (la solution de l'exercice ci-dessous, déjà dans test_a2a.py).

Sur la demande RON-M6, l'agent n'a pas appelé calculer_devis. Est-ce un bug ?

Exercice

Écrivez un test sans clé qui vérifie que agent_commandes déclare bien deux outils, dont un AgentTool qui enveloppe un RemoteA2aAgent pointant vers la carte de STOCK_URL.

Voir la solution 
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.adk.tools.agent_tool import AgentTool

import commandes_agent

def test_agent_stock_est_un_outil_distant():
    outils = commandes_agent.root_agent.tools
    assert len(outils) == 2
    distant = next(t for t in outils if isinstance(t, AgentTool))
    assert isinstance(distant.agent, RemoteA2aAgent)
    assert distant.agent.name == "agent_stock"

L'import ne contacte pas le réseau : la carte n'est lue qu'au premier appel de l'agent distant.

Je sais faire

  • Brancher un agent distant avec RemoteA2aAgent
  • Choisir entre sub_agents et AgentTool
  • Préparer un client HTTP qui s'authentifie seulement en https
  • Tester deux agents A2A en local sur deux ports
Siguiente