Exercices — Session 3 (Avancé) : Claude Agent SDK

Programme : Applied AI — Yann Isola Public : architectes solutions — préparation Claude Certified Architect Prérequis techniques : Python ≥ 3.10, pip install claude-agent-sdk ⚠ (vérifier le nom exact du paquet dans la documentation officielle), clé API (Application Programming Interface — interface de programmation applicative) valide.

Chaque exercice comporte : contexte métier, cahier des charges, contraintes, critères d’évaluation, et corrigé commenté. Ne lisez le corrigé qu’après votre tentative.

Exercice 1 — Construire un agent avec outils (≈ 45 min)

Contexte

Vous équipez le support interne d’un courtier en instruments financiers tokenisés. L’agent doit répondre aux questions des opérateurs sur l’état des règlements-livraisons (settlement) en interrogeant deux systèmes internes (simulés ici par des fonctions Python).

Cahier des charges

Créer deux outils avec le décorateur @tool :
- statut_reglement(reference: str) -> str — retourne le statut d’une opération (simulez avec un dictionnaire : "SETL-001": "réglée", "SETL-002": "en attente de collatéral", "SETL-003": "suspens — incident dépositaire").
- lister_suspens(gravite: str = "toutes") -> str — liste les opérations en suspens, filtrables par gravité ("critique" / "mineure" / "toutes").
Créer un Agent nommé ops-settlement avec des instructions imposant : réponse en français, citation systématique de la référence d’opération, refus poli de toute question hors périmètre règlement-livraison.
Exécuter via Runner.run() sur trois requêtes de test :
- « Où en est SETL-002 ? »
- « Liste-moi les suspens critiques. »
- « Quel temps fait-il à Genève ? » (doit être refusée)

Contraintes de qualité (évaluées)

Docstrings d’outils : chaque docstring doit indiquer quand utiliser l’outil et décrire chaque paramètre. Une docstring d’une ligne vague = échec du critère.
Typage complet : annotations sur tous les paramètres et le retour.
Bornage : lister_suspens doit valider gravite et retourner un message d’erreur textuel actionnable si la valeur est invalide (pas d’exception pour une erreur métier).

Critères d’évaluation (10 pts)

Critère	Pts
Outils fonctionnels, schémas corrects	3
Docstrings de qualité « prompt engineering »	2
Instructions : périmètre + refus hors-sujet effectif	2
Erreur métier retournée en texte (pas levée)	2
Les 3 requêtes de test se comportent comme attendu	1

Corrigé commenté

from claude_agent_sdk import Agent, Runner, tool

REGLEMENTS = {
    "SETL-001": {"statut": "réglée", "gravite": None},
    "SETL-002": {"statut": "en attente de collatéral", "gravite": "mineure"},
    "SETL-003": {"statut": "suspens — incident dépositaire", "gravite": "critique"},
}

@tool
def statut_reglement(reference: str) -> str:
    """Retourne le statut d'une opération de règlement-livraison.

    À utiliser dès que l'utilisateur mentionne une référence d'opération
    (format SETL-XXX) et demande son état.

    Args:
        reference: référence de l'opération, ex. "SETL-002".
    """
    op = REGLEMENTS.get(reference.strip().upper())
    if op is None:
        # Erreur MÉTIER → texte actionnable, pas d'exception :
        # le modèle peut rebondir (demander la bonne référence).
        return (f"Aucune opération trouvée pour '{reference}'. "
                f"Vérifier le format (SETL-XXX) ou la date de valeur.")
    return f"{reference} : {op['statut']}"

@tool
def lister_suspens(gravite: str = "toutes") -> str:
    """Liste les opérations en suspens (non réglées).

    À utiliser pour toute demande de vue d'ensemble des suspens,
    éventuellement filtrée par gravité.

    Args:
        gravite: "critique", "mineure" ou "toutes" (défaut).
    """
    if gravite not in ("critique", "mineure", "toutes"):
        return ("Valeur de gravité invalide. "
                "Valeurs acceptées : critique, mineure, toutes.")
    lignes = [
        f"{ref} : {op['statut']} (gravité : {op['gravite']})"
        for ref, op in REGLEMENTS.items()
        if op["gravite"] and (gravite == "toutes" or op["gravite"] == gravite)
    ]
    return "\n".join(lignes) if lignes else "Aucun suspens pour ce filtre."

agent_ops = Agent(
    name="ops-settlement",
    model="claude-sonnet-4-5",  # ⚠ identifiant de modèle volatile
    instructions=(
        "Tu es l'assistant des opérateurs règlement-livraison d'un courtier. "
        "Règles impératives :\n"
        "1. Réponds exclusivement en français.\n"
        "2. Cite TOUJOURS la référence d'opération (SETL-XXX) dans ta réponse.\n"
        "3. Ton périmètre est STRICTEMENT le règlement-livraison. Pour toute "
        "autre demande, décline poliment en une phrase et rappelle ton périmètre."
    ),
    tools=[statut_reglement, lister_suspens],
)

for question in ["Où en est SETL-002 ?",
                 "Liste-moi les suspens critiques.",
                 "Quel temps fait-il à Genève ?"]:
    print(Runner.run(agent_ops, question).final_output, "\n")

Points de correction à discuter :

La validation de gravite retourne un texte : le modèle peut alors reformuler sa demande — comportement auto-correctif impossible avec une exception brute.
Le refus hors-périmètre est porté par les instructions ; en production on ajouterait un output guardrail (exercice 3) pour le garantir.

Exercice 2 — Implémenter des handoffs (≈ 60 min)

Contexte

Guichet unique de support d’une plateforme de tokenisation : les demandes arrivent mélangées (technique, conformité/KYC — Know Your Customer, connaissance client —, facturation). Vous devez construire un agent de triage qui route par handoff vers trois spécialistes.

Cahier des charges

Trois agents spécialistes : tech, conformite, facturation — chacun avec des instructions propres et au moins un outil factice pertinent.
Un agent triage :
- modèle léger (ex. claude-haiku-4-5 ⚠),
- handoffs vers les trois spécialistes,
- instructions interdisant explicitement de résoudre lui-même.
Un callback on_handoff (via l’option du handoff ou un hook) qui journalise chaque transfert : triage → conformite (raison).
Tests : trois demandes, une par spécialité, plus une demande ambiguë (« Ma vérification d’identité bloque et en plus j’ai été facturé deux fois ») — observer et commenter le choix du modèle.

Contraintes

L’agent cible doit exploiter l’historique : le spécialiste ne doit PAS redemander ce que l’utilisateur a déjà dit (vérifiez dans la sortie).
Prévoir un handoff retour : chaque spécialiste peut retransférer au triage s’il est hors de son périmètre.
Borner le run (max_turns) pour éviter le ping-pong triage ↔ spécialiste.

Critères d’évaluation (10 pts)

Critère	Pts
Routage correct des 3 demandes claires	3
Journalisation des handoffs opérationnelle	2
Historique exploité (pas de re-questionnement)	2
Handoff retour + bornage anti-boucle	2
Analyse écrite du cas ambigu (5–10 lignes)	1

Corrigé commenté (extraits clés)

from claude_agent_sdk import Agent, Runner, handoff, tool

@tool
def verifier_dossier_kyc(client_id: str) -> str:
    """Vérifie l'état du dossier KYC (Know Your Customer) d'un client.

    Args:
        client_id: identifiant client, ex. "C-1024".
    """
    return f"Dossier {client_id} : pièce d'identité expirée, à renouveler."

def log_handoff(source: str, cible: str):
    def _cb(ctx):
        print(f"[HANDOFF] {source} → {cible}")
    return _cb

agent_conformite = Agent(
    name="conformite",
    model="claude-sonnet-4-5",  # ⚠ volatile
    instructions=(
        "Spécialiste conformité/KYC. Tu disposes de l'historique complet : "
        "ne redemande jamais une information déjà fournie. "
        "Si la demande sort de la conformité, retransfère au triage."
    ),
    tools=[verifier_dossier_kyc],
    # handoff retour — ajouté après création du triage (voir note)
)

# ... agents tech et facturation analogues ...

agent_triage = Agent(
    name="triage",
    model="claude-haiku-4-5",   # ⚠ volatile — router = tâche simple, modèle léger
    instructions=(
        "Tu es un aiguilleur. Analyse la demande et transfère au bon "
        "spécialiste : tech (bugs, API), conformite (KYC, identité), "
        "facturation (paiements, factures). Tu ne résous JAMAIS toi-même. "
        "Si plusieurs sujets coexistent, choisis le plus bloquant pour "
        "le client et mentionne l'autre sujet dans ton transfert."
    ),
    handoffs=[
        handoff(agent_tech,        on_handoff=log_handoff("triage", "tech")),
        handoff(agent_conformite,  on_handoff=log_handoff("triage", "conformite")),
        handoff(agent_facturation, on_handoff=log_handoff("triage", "facturation")),
    ],
)

# Handoff retour : câblé après coup pour éviter la référence circulaire.
agent_conformite.handoffs = [handoff(agent_triage,
                                     on_handoff=log_handoff("conformite", "triage"))]

res = Runner.run(agent_triage,
                 "Ma vérification d'identité bloque et j'ai été facturé deux fois.",
                 max_turns=12)   # ⬅ anti-boucle
print(res.final_output)

Points de correction :

Référence circulaire : triage ↔ spécialistes s’entre-référencent ; on câble le handoff retour après l’instanciation. Question d’architecture classique.
Cas ambigu : il n’y a pas de « bonne » réponse unique — ce qu’on évalue, c’est que le candidat a anticipé l’ambiguïté dans les instructions du triage (règle « le plus bloquant d’abord, mentionner l’autre sujet »). Alternative défendable : demander une clarification à l’utilisateur avant de router.
max_turns=12 : sans borne, un spécialiste mal instruit peut retransférer indéfiniment.

Exercice 3 — Concevoir des guardrails (≈ 60 min)

Contexte

Votre agent facturation de l’exercice 2 part en production dans un contexte régulé. Direction des risques : « aucune donnée de carte ne doit entrer, aucune promesse d’engagement financier ne doit sortir, et tout doit être auditable ».

Cahier des charges

Input guardrail bloquer_pan : détecte un PAN (Primary Account Number — numéro de carte bancaire, 16 chiffres éventuellement séparés par espaces/tirets) dans le message entrant et déclenche le tripwire avec un message d’orientation vers le canal sécurisé.
Output guardrail bloquer_engagement : bloque toute sortie contenant une promesse ferme de remboursement ou un montant garanti. Deux implémentations à livrer et comparer :
- v1 : règles/regex (mots déclencheurs : « je vous garantis », « vous serez remboursé de », montants + « sous X jours »…) ;
- v2 : LLM-as-judge — un modèle léger classifie la sortie (engagement_ferme / information), avec un prompt de juge que vous rédigez.
Hooks d’audit : classe RunHooks journalisant on_tool_start/end et les déclenchements de guardrails (horodatage, nom d’agent, motif) dans un fichier JSONL (JSON Lines — un objet JSON par ligne).
Gestion applicative : intercepter les exceptions tripwire et retourner à l’utilisateur un message de repli propre (jamais de stack trace).
Jeu de tests : ≥ 6 cas — 2 entrées avec PAN (formats différents), 1 entrée saine, 2 sorties engageantes, 1 sortie informative. Mesurer faux positifs / faux négatifs de v1 vs v2.

Critères d’évaluation (10 pts)

Critère	Pts
Input guardrail : détection des 2 formats de PAN, entrée saine passante	2
Output guardrail v1 (règles) fonctionnel	2
Output guardrail v2 (LLM-as-judge) : prompt de juge rigoureux	2
Comparaison v1/v2 chiffrée (tableau FP/FN) + recommandation argumentée	2
Hooks d’audit JSONL + repli applicatif propre	2

Corrigé commenté (extraits clés)

import re, json, datetime
from claude_agent_sdk import (Agent, Runner, RunHooks,
                              input_guardrail, output_guardrail,
                              GuardrailTripwire,
                              InputGuardrailTripwireTriggered,
                              OutputGuardrailTripwireTriggered)

PAN_RE = re.compile(r"\b(?:\d[ -]?){15}\d\b")   # 16 chiffres, séparateurs tolérés

@input_guardrail
def bloquer_pan(ctx, agent, message: str) -> GuardrailTripwire:
    """Refuse tout message contenant un numéro de carte (PAN)."""
    if PAN_RE.search(message):
        return GuardrailTripwire(
            triggered=True,
            message=("Par sécurité, ne transmettez jamais de numéro de carte ici. "
                     "Utilisez le portail sécurisé : Espace client → Paiements."),
        )
    return GuardrailTripwire(triggered=False)

# ---- Output v1 : règles ----
MOTIFS_ENGAGEMENT = [
    r"je vous garantis", r"vous serez rembours",
    r"\d+ ?(€|EUR|CHF).{0,40}sous \d+ jours",
]
@output_guardrail
def bloquer_engagement_v1(ctx, agent, sortie: str) -> GuardrailTripwire:
    """Bloque toute promesse ferme d'engagement financier (règles)."""
    hit = any(re.search(p, sortie, re.IGNORECASE) for p in MOTIFS_ENGAGEMENT)
    return GuardrailTripwire(triggered=hit)

# ---- Output v2 : LLM-as-judge ----
PROMPT_JUGE = """Tu es un contrôleur conformité. Classe le texte suivant :
- "engagement_ferme" : promesse contraignante (remboursement garanti,
  montant précis dû, délai ferme engageant la société).
- "information" : explication de procédure, statut, conditionnel
  ("sous réserve de validation", "généralement").
Réponds par UN seul mot. Texte :
\"\"\"{sortie}\"\"\""""

@output_guardrail
def bloquer_engagement_v2(ctx, agent, sortie: str) -> GuardrailTripwire:
    """Bloque les engagements fermes (juge LLM léger)."""
    verdict = appel_modele_leger(PROMPT_JUGE.format(sortie=sortie))  # ⚠ coût/latence
    return GuardrailTripwire(triggered=verdict.strip() == "engagement_ferme")

# ---- Hooks d'audit JSONL ----
class AuditHooks(RunHooks):
    def _log(self, **champs):
        champs["ts"] = datetime.datetime.utcnow().isoformat()
        with open("audit.jsonl", "a") as f:
            f.write(json.dumps(champs, ensure_ascii=False) + "\n")
    async def on_tool_start(self, ctx, agent, tool):
        self._log(evt="tool_start", agent=agent.name, tool=tool.name)
    async def on_tool_end(self, ctx, agent, tool, result):
        self._log(evt="tool_end", agent=agent.name, tool=tool.name)

# ---- Repli applicatif ----
def repondre(agent, message):
    try:
        return Runner.run(agent, message, hooks=AuditHooks()).final_output
    except InputGuardrailTripwireTriggered as e:
        return e.guardrail_message          # message d'orientation, pas de trace
    except OutputGuardrailTripwireTriggered:
        return ("Votre demande nécessite une validation manuelle. "
                "Un conseiller vous recontacte sous 24 h ouvrées.")

Comparaison attendue v1 vs v2 (exemple de résultat) :

	Faux positifs	Faux négatifs	Latence	Coût
v1 règles	Moyens (bloque « vous serez remboursé si la fraude est confirmée »)	Élevés (paraphrases non listées)	~0 ms	0
v2 juge LLM	Faibles	Faibles	+200–800 ms ⚠	~1 appel léger/réponse ⚠

Recommandation type (à argumenter) : v1 en première ligne (rapide, gratuite) ET v2 en seconde ligne sur les sorties que v1 laisse passer — défense en profondeur, coût du juge payé seulement quand nécessaire. Toute réponse justifiant un autre arbitrage (ex. v2 seul pour un flux à faible volume) est acceptable si le raisonnement coût/latence/risque est explicite.

Barème global

Exercice 1 : /10 — Exercice 2 : /10 — Exercice 3 : /10.
≥ 24/30 : niveau attendu pour la certification sur ce domaine.
Entre 18 et 23 : revoir handoffs et guardrails (séquences 3–4 du cours).
< 18 : refaire l’exercice 1 avec le corrigé masqué, puis progresser.