Claude Architect, épisode 4 : un coordinateur et deux sous-agents, en Python

import json
from functools import partial

import anthropic

client = anthropic.Anthropic()        # lit la clé dans ANTHROPIC_API_KEY
MODEL = "claude-haiku-4-5"
MAX_ITERATIONS = 10                   # soupape de sécurité, pas la condition d'arrêt

# --- Fausses données : aucune vraie API n'est appelée ---
CLIENTS = {
    "C001": {"name": "Alice", "email": "alice@example.com", "plan": "pro"},
    "C002": {"name": "Bob", "email": "bob@example.com", "plan": "free"},
}
COMMANDES = {
    "0100": {"customer_id": "C001", "item": "Annual Plan", "amount": 99.00, "status": "delivered"},
    "0101": {"customer_id": "C002", "item": "Monthly Plan", "amount": 9.00, "status": "refunded"},
}


# --- Les outils : du code ordinaire, que Claude ne peut que demander ---
def get_customer(customer_id: str) -> dict:
    fiche = CLIENTS.get(customer_id)
    if fiche:
        return {"found": True, "customer_id": customer_id, **fiche}
    return {"found": False, "customer_id": customer_id}   # résultat valide, pas une erreur


def lookup_order(order_id: str) -> dict:
    commande = COMMANDES.get(order_id)
    if commande:
        return {"found": True, "order_id": order_id, **commande}
    return {"found": False, "order_id": order_id}


def process_refund(order_id: str, amount: float, *, client_verifie: str) -> dict:
    # client_verifie est fixé par le coordinateur, jamais par le modèle.
    commande = COMMANDES.get(order_id)
    if not commande:
        return {"ok": False, "reason": f"no order {order_id}"}
    if commande["customer_id"] != client_verifie:
        return {"ok": False, "reason": "order does not belong to the verified customer"}
    if commande["status"] == "refunded":
        return {"ok": False, "reason": "order already refunded"}
    if amount > commande["amount"]:
        return {"ok": False, "reason": "amount exceeds the order total"}
    commande["status"] = "refunded"
    return {"ok": True, "order_id": order_id, "refunded": amount}

# --- Les schémas envoyés à l'API : ce que Claude sait des outils ---
SCHEMAS = {
    "get_customer": {
        "name": "get_customer",
        "description": "Fetch a customer record by ID. Returns found, name, email and plan.",
        "input_schema": {
            "type": "object",
            "properties": {"customer_id": {"type": "string", "description": "e.g. C001"}},
            "required": ["customer_id"],
        },
    },
    "lookup_order": {
        "name": "lookup_order",
        "description": "Fetch an order by ID. Returns item, amount, status and the linked customer ID.",
        "input_schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string", "description": "e.g. 0100"}},
            "required": ["order_id"],
        },
    },
    "process_refund": {
        "name": "process_refund",
        "description": "Refund an order for the verified customer. Fails if the order belongs "
                       "to someone else, is already refunded, or the amount is too high.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string"},
                "amount": {"type": "number", "description": "Amount to refund, in USD"},
            },
            "required": ["order_id", "amount"],
        },
    },
}

# --- Épisode 1 : la boucle agentique, avec ses garde-fous ---
def run_agentic_loop(system_prompt: str, user_message: str, schemas: list, fonctions: dict):
    """Renvoie le texte final de Claude ET le journal des appels d'outils."""
    messages = [{"role": "user", "content": user_message}]
    journal = []
    for _ in range(MAX_ITERATIONS):
        response = client.messages.create(
            model=MODEL, max_tokens=1024, system=system_prompt,
            tools=schemas, messages=messages,
        )
        if response.stop_reason == "end_turn":
            texte = "".join(b.text for b in response.content if b.type == "text")
            return texte, journal
        if response.stop_reason != "tool_use":
            # max_tokens, refusal… : s'arrêter plutôt que renvoyer la même requête
            raise RuntimeError(f"arrêt inattendu : {response.stop_reason}")

        messages.append({"role": "assistant", "content": response.content})
        resultats = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            fonction = fonctions.get(block.name)
            try:
                if fonction is None:
                    raise PermissionError(f"tool {block.name} is not allowed here")
                sortie, en_erreur = fonction(**block.input), False
            except Exception as exc:          # l'outil a échoué : on le dit à Claude
                sortie, en_erreur = {"error": str(exc)}, True
            journal.append({"tool": block.name, "input": block.input, "output": sortie})
            resultats.append({
                "type": "tool_result",
                "tool_use_id": block.id,      # doit correspondre au bloc tool_use
                "content": json.dumps(sortie),
                "is_error": en_erreur,
            })
        messages.append({"role": "user", "content": resultats})
    raise RuntimeError("MAX_ITERATIONS atteint")

# --- Épisode 2 : un sous-agent, c'est la même boucle avec un rôle étroit ---
def run_subagent(role: str, task_prompt: str, schemas: list, fonctions: dict) -> dict:
    system_prompt = (
        f"You are the {role} subagent in a customer support system. "
        "Use only your tools. Be concise. Never make up data."
    )
    texte, journal = run_agentic_loop(system_prompt, task_prompt, schemas, fonctions)
    return {"role": role, "summary": texte, "tool_calls": journal}


def dernier_resultat(sous_agent: dict, outil: str):
    sorties = [c["output"] for c in sous_agent["tool_calls"] if c["tool"] == outil]
    return sorties[-1] if sorties else None

# --- Le coordinateur : décomposer, contrôler, transmettre, agréger ---
def coordinator(customer_id: str, order_id: str) -> dict:
    # Sous-tâche 1 : le vérificateur ne reçoit que l'identifiant du client.
    verification = run_subagent(
        "Customer Verifier",
        f"Look up customer {customer_id} and confirm they exist. Return their name and plan.",
        [SCHEMAS["get_customer"]],
        {"get_customer": get_customer},
    )

    # Épisode 3 : le constat structuré vient du résultat de l'outil, pas de la prose.
    fiche = dernier_resultat(verification, "get_customer") or {}
    constat = {
        "customer_id": customer_id,
        "verified": fiche.get("found") is True,
        "plan": fiche.get("plan"),
    }

    # La porte : sans client vérifié, le remboursement ne démarre jamais.
    if not constat["verified"]:
        return {"status": "rejected", "reason": "customer not verified", "verification": constat}

    # Sous-tâche 2 : le constat est passé explicitement, en JSON.
    remboursement = run_subagent(
        "Refund Processor",
        f"Process a refund for order {order_id}.\n\n"
        f"PRIOR VERIFICATION (structured):\n{json.dumps(constat, indent=2)}\n\n"
        "Steps: 1. Look up the order. 2. Check it belongs to the verified customer. "
        "3. If so, refund the full order amount. Otherwise, explain why not.",
        [SCHEMAS["lookup_order"], SCHEMAS["process_refund"]],
        {
            "lookup_order": lookup_order,
            "process_refund": partial(process_refund, client_verifie=customer_id),
        },
    )

    # Agrégation : un statut lisible par du code, et le résumé pour un humain.
    refund = dernier_resultat(remboursement, "process_refund") or {}
    return {
        "status": "refunded" if refund.get("ok") else "not_refunded",
        "verification": constat,
        "refund": refund,
        "summary": remboursement["summary"],
    }


if __name__ == "__main__":
    for customer_id, order_id in [("C001", "0100"), ("C999", "0100"), ("C002", "0101")]:
        print(f"\n=== {customer_id} / commande {order_id}")
        print(json.dumps(coordinator(customer_id, order_id), indent=2, ensure_ascii=False))

from claude_agent_sdk import ClaudeAgentOptions, AgentDefinition

options = ClaudeAgentOptions(
    allowed_tools=["Agent"],          # "Task" dans les anciennes versions
    agents={
        "customer-verifier": AgentDefinition(
            description="Verifies that a customer exists. Use before any refund.",
            prompt="Look up the customer and report found, name and plan. Never make up data.",
            tools=["mcp__support__get_customer"],
            model="haiku",
        ),
        "refund-processor": AgentDefinition(
            description="Processes a refund for an already verified customer.",
            prompt="Check the order belongs to the verified customer, then refund it.",
            tools=["mcp__support__lookup_order", "mcp__support__process_refund"],
            model="haiku",
        ),
    },
)