# Exercises — Session 3 (Advanced): Claude Agent SDK

> **Program:** Applied AI — Yann Isola
> **Public:** solutions architects — preparation *Claude Certified Architect*
> **Technical prerequisites:** Python ≥ 3.10, `pip install claude-agent-sdk` ⚠ (check the exact package name in the official documentation), valid API (Application Programming Interface) key.

Each exercise includes: business context, specifications, constraints, evaluation criteria, and commented answer key. **Read the answer key only after your attempt.**

---

## Exercise 1 — Building an agent with tools (≈ 45 min)

### Context

You equip the internal support of a broker with tokenized financial instruments. The agent must answer operators' questions on the status of settlements by querying two internal systems (simulated here by Python functions).

### Specifications

1. Create two tools with the `@tool` decorator:
- `statut_reglement(reference: str) -> str` — returns the status of an operation (simulate with a dictionary: `"SETL-001": "réglée"`, `"SETL-002": "en attente de collatéral"`, `"SETL-003": "suspens — incident dépositaire"`).
- `lister_suspens(gravite: str = "toutes") -> str` — lists pending operations, filterable by severity (`"critique"` / `"mineure"` / `"toutes"`).
2. Create a `Agent` named `ops-settlement` with imposing `instructions`: response in French, systematic citation of the operation reference, polite refusal of any question outside the payment-delivery scope.
3. Run through `Runner.run()` on three test queries:
- “Where is SETL-002?” »
- “List me the critical suspenses. »
- “What is the weather like in Geneva?” » (must be refused)

### Quality constraints (evaluated)

- **Tool Docstrings**: Each docstring must indicate *when* to use the tool and describe each parameter. A vague one-line docstring = criterion failure.
- **Complete typing**: annotations on all parameters and feedback.
- **Boundary**: `lister_suspens` must validate `gravite` and return an actionable textual error message if the value is invalid (no exception for a business error).

### Evaluation criteria (10 pts)

| Criterion | Pts |
|---|---|
| Functional tools, correct diagrams | 3 |
| Quality “prompt engineering” docstrings | 2 |
| Instructions: scope + effective off-topic refusal | 2 |
| Business error returned as text (not raised) | 2 |
| The 3 test queries behave as expected | 1 |

### Fixed commented```python
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")
```**Correction points to discuss:**
- Validation of `gravite` returns a text: the model can then reformulate its request — self-correcting behavior impossible with a raw exception.
- Out-of-scope refusal is carried by `instructions`; in production we would add an output guardrail (exercise 3) to guarantee this.

---

## Exercise 2 — Implementing handoffs (≈ 60 min)

### Context

One-stop shop for support of a tokenization platform: requests arrive mixed (technical, compliance/KYC — Know Your Customer, billing). You must build a **triage agent** who routes by handoff to three specialists.

### Specifications

1. Three specialist agents: `tech`, `conformite`, `facturation` — each with their own `instructions` and **at least one dummy tool** relevant.
2. An agent `triage`:
- light model (e.g. `claude-haiku-4-5` ⚠),
- `handoffs` to the three specialists,
- instructions explicitly prohibiting resolving itself.
3. A callback `on_handoff` (via the handoff option or a hook) which logs each transfer: `triage → conformite (raison)`.
4. Tests: three requests, one per specialty, plus one **ambiguous** request (“My identity verification is blocked and on top of that I was billed twice”) — observe and comment on the choice of model.

### Constraints

- The target agent must exploit the history: the specialist must NOT repeat what the user has already said (check in the output).
- Provide a **return handoff**: each specialist can transfer back to triage if they are outside their perimeter.
- Limit the run (`max_turns`) to avoid triage ↔ specialist ping-pong.

### Evaluation criteria (10 pts)

| Criterion | Pts |
|---|---|
| Correct routing of the 3 clear requests | 3 |
| Logging of operational handoffs | 2 |
| History used (no re-questioning) | 2 |
| Handoff return + anti-loop terminal | 2 |
| Written analysis of the ambiguous case (5–10 lines) | 1 |

### Commented answer key (key extracts)```python
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)
```**Correction points:**
- **Circular reference**: triage ↔ specialists reference each other; we wire the handoff back *after* the instantiation. Question of classical architecture.
- **Ambiguous case**: there is no single “right” answer — what we evaluate is that the candidate has *anticipated* the ambiguity in the triage instructions (rule “the most blocking first, mention the other subject”). Defensible alternative: request clarification from the user before routing.
- **`max_turns=12`**: without limit, a poorly educated specialist can retransfer indefinitely.

---

## Exercise 3 — Designing guardrails (≈ 60 min)

### Context

Your agent `facturation` from exercise 2 goes into production in a regulated context. Risk management: “no card data must enter, no promise of financial commitment must go out, and everything must be auditable”.

### Specifications

1. **Input guardrail** `bloquer_pan`: detects a PAN (Primary Account Number — bank card number, 16 digits optionally separated by spaces/dashes) in the incoming message and triggers the tripwire with a message directing to the secure channel.
2. **Output guardrail** `bloquer_engagement`: blocks any output containing a firm promise of reimbursement or a guaranteed amount. Two implementations to deliver and compare:
- v1: rules/regex (trigger words: “I guarantee you”, “you will be reimbursed for”, amounts + “within X days”…);
- v2: LLM-as-judge — a lightweight model classifies the output (`engagement_ferme` / `information`), with a judge prompt that you write.
3. **Audit Hooks**: `RunHooks` class logging `on_tool_start/end` and guardrails triggers (timestamp, agent name, reason) to a JSONL file (JSON Lines — one JSON object per line).
4. **Application management**: intercept tripwire exceptions and return a clean fallback message to the user (never stack trace).
5. **Test set**: ≥ 6 cases — 2 inputs with PAN (different formats), 1 healthy input, 2 engaging outputs, 1 informative output. Measure false positives/false negatives of v1 vs v2.

### Evaluation criteria (10 pts)

| Criterion | Pts |
|---|---|
| Input guardrail: detection of the 2 PAN formats, healthy input passing | 2 |
| Output guardrail v1 (rules) functional | 2 |
| Output guardrail v2 (LLM-as-judge): rigorous judge prompt | 2 |
| Encrypted v1/v2 comparison (table FP/FN) + reasoned recommendation | 2 |
| JSONL audit hooks + own application fallback | 2 |

### Commented answer key (key extracts)```python
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.")
```**Expected comparison v1 vs v2 (example result):**

| | False positives | False negatives | Latency | Cost |
|---|---|---|---|---|
| v1 rules | Means (blocks “you will be reimbursed *if* the fraud is confirmed”) | High (paraphrases not listed) | ~0ms | 0 |
| v2 judge LLM | Weak | Weak | +200–800ms ⚠ | ~1 light call/answer ⚠ |

**Typical recommendation (to be argued):** v1 in the first line (fast, free) **AND** v2 in the second line on the exits that v1 lets pass — defense in depth, cost of the judge paid only when necessary. Any answer justifying another arbitration (e.g. v2 alone for a low volume flow) is acceptable if the cost/latency/risk reasoning is explicit.

---

## Overall scale

- Exercise 1: /10 — Exercise 2: /10 — Exercise 3: /10.
- ≥ 24/30: expected level for certification in this area.
- Between 18 and 23: review handoffs and guardrails (sequences 3–4 of the course).
- < 18: repeat exercise 1 with the hidden answer key, then progress.