Registre et découverte d'agents

Trois façons de trouver un agent

Méthode	Principe	Quand l'utiliser
URL en dur	l'adresse est dans le code	prototype, un seul agent
Registre (fichier, base, service)	une liste d'agents autorisés, maintenue à part	entreprise : catalogue contrôlé d'agents
Well-known seul	on connaît un domaine, on lit `/.well-known/agent-card.json`	agents publics d'un partenaire

La spécification A2A décrit la carte et son adresse bien connue, mais laisse le registre à votre charge. Dans ce lab, c'est un fichier JSON : simple, versionné avec le code, relu en revue de code.

files/registry.json

{
  "agents": [
    {"name": "stock_agent", "url": "http://localhost:8101"},
    {"name": "pricing_agent", "url": "http://localhost:8102"}
  ]
}

La découverte au démarrage

flowchart TB
  R[("registry.json")] --> L["Pour chaque entrée"]
  L --> G["GET url/.well-known/agent-card.json (timeout 2 s)"]
  G -->|"200 + JSON valide"| F["found : nom, description, skills, card_url"]
  G -->|"erreur réseau, HTTP, JSON"| M["missing : nom"]
  F --> B["build_orchestrator(found, missing)"]
  M --> B
  B --> I["Instruction : agents disponibles / indisponibles"]

files/orchestrator.py

def discover(registry_path: Path = REGISTRY, timeout: float = 2.0) -> tuple[list[dict], list[str]]:
    """Lit le registre et récupère l'Agent Card de chaque agent joignable.

    Retourne (agents_disponibles, noms_indisponibles). Un agent qui ne répond pas
    en `timeout` secondes est écarté : l'orchestrateur démarre quand même.
    """
    found, missing = [], []
    for entry in json.loads(registry_path.read_text(encoding="utf-8"))["agents"]:
        url = entry["url"].rstrip("/") + CARD_PATH
        try:
            card = httpx.get(url, timeout=timeout).raise_for_status().json()
            found.append({"name": card["name"], "description": card.get("description", ""),
                          "skills": [s.get("name") for s in card.get("skills", [])], "card_url": url})
        except (httpx.HTTPError, ValueError, KeyError):
            missing.append(entry["name"])
    return found, missing

Deux choix de conception :

Le nom et la description viennent de la carte, pas du registre : c'est l'équipe propriétaire de l'agent qui décrit ce qu'il fait, et l'orchestrateur s'adapte à chaque redémarrage.
Un agent absent n'empêche pas le démarrage. Il est listé dans l'instruction comme indisponible, pour que le modèle l'annonce au lieu de chercher un outil qui n'existe pas.

L'agent tarification est arrêté quand l'orchestrateur démarre. Que contient l'orchestrateur construit ?

Tester la découverte sans réseau

Le test pointe un registre temporaire vers le port 9 (réservé, jamais ouvert) : la connexion échoue immédiatement.

files/test_lab.py

def test_decouverte_ecarte_les_agents_injoignables(tmp_path):
    reg = tmp_path / "registry.json"
    reg.write_text(json.dumps({"agents": [{"name": "fantome", "url": "http://127.0.0.1:9"}]}))
    found, missing = orchestrator.discover(reg, timeout=0.5)
    assert found == [] and missing == ["fantome"]

Résultat obtenu

Scénario 4 de e2e_a2a.py (2026-10-15), après l'arrêt réel du serveur tarification :

4) Découverte avec un agent absent du réseau
   disponibles : ['stock_agent'] | absents : ['pricing_agent']

Limites de la découverte « au démarrage »

Limite	Conséquence	Amélioration
La liste est figée au démarrage	un agent revenu en ligne n'est pas vu	redécouvrir périodiquement, ou à chaque requête avec un cache court
Un agent peut tomber après la découverte	l'outil existe mais échoue	gestion d'erreurs à l'exécution (leçon suivante)
La carte n'est pas signée	un serveur usurpé pourrait publier une fausse carte	HTTPS, registre contrôlé, et cartes signées si votre version du protocole le permet

Pourquoi garder le registre sous forme de fichier versionné plutôt que laisser chaque agent s'inscrire tout seul ?

Exercice : redécouverte paresseuse

Modifiez l'orchestrateur pour qu'un agent absent au démarrage soit retenté au plus une fois par minute. Esquissez le code.

Voir la solution

L'agent ADK est construit une fois ; le plus simple est de reconstruire l'orchestrateur quand la liste change :

import time
_state = {"at": 0.0, "agent": None, "names": None}

def current_orchestrator(token, ttl=60):
    if time.monotonic() - _state["at"] > ttl:
        found, missing = discover()
        names = tuple(a["name"] for a in found)
        if names != _state["names"]:
            _state["agent"] = build_orchestrator(found, missing, token)
            _state["names"] = names
        _state["at"] = time.monotonic()
    return _state["agent"]

Il faut ensuite un runner qui utilise cet agent à chaque requête (par exemple un agent_executor_factory personnalisé passé à to_a2a). C'est plus de code à maintenir : dans beaucoup de cas, un redémarrage de l'orchestrateur après un déploiement suffit.

Je sais faire

Décrire les trois façons de trouver un agent
Écrire une découverte qui tolère les agents absents
Tester la découverte sans réseau
Expliquer pourquoi le registre est une décision de sécurité