Construire un système multi-agents avec Claude : guide pratique (SDK + MCP)

Cet article est la suite directe de Systèmes agentiques avec Claude : architecture et théorie. On y a vu le quoi et le pourquoi. Ici, on attaque le comment.

Tu vas trouver deux voies parallèles :

• Voie A : le Claude Agent SDK — la voie officielle Anthropic, opinionated, rapide à mettre en place.
• Voie B : l'API brute + MCP — orchestration custom, plus de contrôle, plus pédagogique pour comprendre la mécanique interne.

Les deux voies aboutissent au même cas pratique : un système "veille marché" composé d'un orchestrateur et de 3 sub-agents qui collaborent. À la fin, un tableau comparatif te permet de choisir laquelle adopter selon ton contexte.

0. Setup commun aux deux voies

Avant d'attaquer, on prépare l'environnement.

Structure de projet recommandée

mon-systeme-agents/
├── .env                    # ANTHROPIC_API_KEY=sk-ant-...
├── requirements.txt
├── voie_a_sdk/
│   ├── orchestrateur.py
│   ├── sub_agents.py
│   └── tools/
└── voie_b_custom/
    ├── orchestrateur.py
    ├── mcp_server.py
    └── workers.py

Installation

python -m venv .venv
source .venv/bin/activate  # ou .venv\Scripts\activate sur Windows

# Voie A : SDK officiel
pip install claude-agent-sdk python-dotenv

# Voie B : API brute + MCP
pip install anthropic mcp httpx python-dotenv

Le fichier .env

ANTHROPIC_API_KEY=sk-ant-api03-...

Et le chargement dans tes scripts :

from dotenv import load_dotenv
load_dotenv()

1. Voie A — Claude Agent SDK

Pourquoi cette voie

Le Claude Agent SDK (anciennement Claude Code SDK, renommé en 2025) est l'infrastructure qu'Anthropic utilise en interne pour Claude Code, exposée comme une bibliothèque. Tu hérites de leur travail sur la gestion de la boucle nO, le compactage automatique du contexte, le spawn de sub-agents avec contexte isolé, l'intégration MCP native, les hooks de cycle de vie, et le budget en USD plafonné par session.

Le SDK n'est pas une bibliothèque API pure : il bundle un binaire CLI Claude Code qu'il lance comme subprocess. Ça pèse 270-340 Mo par release, à prendre en compte pour les images Docker et les pipelines CI.

Ton premier agent en 15 lignes

# voie_a_sdk/premier_agent.py
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions
from dotenv import load_dotenv

load_dotenv()

async def main():
    options = ClaudeAgentOptions(
        system_prompt="Tu es un assistant qui répond avec précision.",
        max_budget_usd=0.50,  # plafond dur, indispensable
    )
    async for message in query(
        prompt="Quelle est la différence entre un workflow et un agent ?",
        options=options,
    ):
        print(message)

anyio.run(main)

C'est tout. Pas de gestion manuelle de la boucle, pas de parsing de tool calls, pas de tracking du contexte. Le SDK fait tout.

Ajouter des tools (MCP intégré)

Le SDK accepte des tools via le Model Context Protocol (MCP). Tu peux pointer vers un MCP server existant (Asana, GitHub, Slack…) ou définir tes propres tools en quelques lignes.

from claude_agent_sdk import query, ClaudeAgentOptions, tool
from dotenv import load_dotenv
import anyio

load_dotenv()

@tool(
    name="get_weather",
    description="Récupère la météo actuelle pour une ville",
)
async def get_weather(city: str) -> dict:
    # Ici tu appellerais une vraie API météo
    return {"city": city, "temp": 18, "conditions": "nuageux"}

async def main():
    options = ClaudeAgentOptions(
        system_prompt="Tu es un assistant météo.",
        tools=[get_weather],
        max_budget_usd=0.50,
    )
    async for message in query(
        prompt="Quelle météo à Paris et à Lyon ?",
        options=options,
    ):
        print(message)

anyio.run(main)

Claude va appeler get_weather deux fois (une par ville), agréger les résultats et te répondre.

Spawn de sub-agents avec contexte isolé

C'est ici que le SDK brille. Pour faire de l'orchestrateur-workers, tu déclares des sub-agents que l'agent principal peut invoquer comme des tools.

# voie_a_sdk/orchestrateur.py
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, subagent
from dotenv import load_dotenv

load_dotenv()

# Définition d'un sub-agent spécialisé
researcher = subagent(
    name="web_researcher",
    description="Effectue une recherche web approfondie sur un sujet précis et retourne un résumé structuré.",
    system_prompt="""Tu es un chercheur web expert.
    Pour chaque requête, effectue 3-5 recherches ciblées.
    Retourne un résumé structuré en JSON avec : findings, sources, confidence_level.""",
    tools=["web_search"],
    max_turns=10,
)

synthesizer = subagent(
    name="synthesizer",
    description="Prend plusieurs résumés de recherche et produit une synthèse cohérente.",
    system_prompt="Tu es un analyste qui croise des sources pour produire des synthèses neutres.",
    max_turns=5,
)

async def main():
    options = ClaudeAgentOptions(
        system_prompt="""Tu es un orchestrateur de recherche.
        Décompose la requête en sous-recherches indépendantes.
        Lance plusieurs web_researcher en parallèle si pertinent.
        Utilise synthesizer pour produire le rapport final.""",
        subagents=[researcher, synthesizer],
        max_budget_usd=5.00,
    )

    async for message in query(
        prompt="Fais-moi une veille sur les outils français d'observabilité d'agents IA en 2026.",
        options=options,
    ):
        print(message)

anyio.run(main)

Ce qui se passe en coulisses :

1. L'orchestrateur lit la requête et décide de spawner plusieurs web_researcher en parallèle (un par angle).
2. Chaque web_researcher tourne dans son propre contexte isolé avec son propre budget en tokens.
3. Les résumés condensés remontent à l'orchestrateur.
4. L'orchestrateur appelle synthesizer avec les résumés pour produire le rapport.
5. Le rapport final est retourné à l'utilisateur.

Hooks : observabilité et garde-fous

Le SDK expose des hooks de cycle de vie pour intercepter chaque étape :

async def log_pre_tool(tool_name, tool_input):
    print(f"[PRE] {tool_name} appelé avec {tool_input}")

async def log_post_tool(tool_name, tool_result):
    print(f"[POST] {tool_name} a retourné {len(str(tool_result))} chars")

options = ClaudeAgentOptions(
    # ... config ...
    pre_tool_use=log_pre_tool,
    post_tool_use=log_post_tool,
)

C'est ton observabilité minimale viable. En prod tu remplaces les print par des envois vers Datadog, Sentry, ou une base de tracing custom.

2. Voie B — API brute + MCP

Pourquoi cette voie

L'Agent SDK est élégant mais te cache des choses. Si tu veux comprendre précisément ce qui se passe dans la boucle, avoir un contrôle total sur l'orchestration (logique métier custom, états persistants exotiques), ne pas embarquer 300 Mo de binaire dans ton image Docker, ou simplement apprendre, alors tu codes l'orchestration toi-même sur l'API brute, en utilisant MCP comme protocole pour exposer tes tools.

Le rôle de MCP dans cette voie

MCP (Model Context Protocol) est un standard ouvert qui définit comment un agent et un outil communiquent. Tu écris un MCP server qui expose des capacités (lire un fichier, requêter une DB, appeler une API…), et n'importe quel agent compatible MCP peut s'y connecter.

L'intérêt : tes tools deviennent réutilisables entre agents et entre projets, tu peux versionner et tester un MCP server indépendamment, tu décourples ta logique métier du modèle.

Construire un MCP server custom

Voici un MCP server minimal qui expose un outil de recherche web :

# voie_b_custom/mcp_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx
import asyncio

app = Server("mcp-recherche")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Effectue une recherche web et retourne les 5 premiers résultats.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Requête de recherche"},
                },
                "required": ["query"],
            },
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_web":
        async with httpx.AsyncClient() as client:
            r = await client.get(
                "https://api.tavily.com/search",
                params={"query": arguments["query"], "max_results": 5},
            )
            results = r.json()
        return [TextContent(type="text", text=str(results))]
    raise ValueError(f"Tool inconnu : {name}")

async def main():
    async with stdio_server() as (read, write):
        await app.run(read, write, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

Tu lances ce server avec python mcp_server.py. Il communique via stdin/stdout selon le standard MCP.

Coder son propre orchestrateur

Maintenant la partie la plus instructive : reconstituer la boucle nO et le pattern orchestrateur-workers à la main.

# voie_b_custom/orchestrateur.py
import os, json, asyncio
from anthropic import AsyncAnthropic
from dotenv import load_dotenv

load_dotenv()
client = AsyncAnthropic()

MODEL_LEAD = "claude-opus-4-7"
MODEL_WORKER = "claude-sonnet-4-6"

# --- Définition d'un worker ---
async def run_worker(worker_id: str, mission: str, tools: list) -> dict:
    """Un worker tourne dans son propre contexte isolé."""
    messages = [{"role": "user", "content": mission}]
    iterations = 0
    MAX_ITER = 10

    while iterations < MAX_ITER:
        iterations += 1
        response = await client.messages.create(
            model=MODEL_WORKER,
            max_tokens=4096,
            system=f"Tu es le worker {worker_id}. Mission unique : {mission}. Sois concis.",
            tools=tools,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            final = next((b.text for b in response.content if hasattr(b, "text")), "")
            return {"worker_id": worker_id, "summary": final, "iterations": iterations}

        messages.append({"role": "assistant", "content": response.content})
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = await execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result),
                })
        messages.append({"role": "user", "content": tool_results})

    return {"worker_id": worker_id, "summary": "MAX_ITER atteint", "iterations": iterations}


async def execute_tool(name: str, args: dict):
    """Dispatcher de tools — ici tu appelles ton MCP server ou tes fonctions."""
    if name == "search_web":
        return {"results": ["..."]}  # placeholder
    raise ValueError(f"Tool inconnu : {name}")


# --- L'orchestrateur ---
async def orchestrate(user_request: str) -> str:
    print(f"🎯 Requête : {user_request}\n")

    # ÉTAPE 1 : le lead décompose la requête en missions
    plan_response = await client.messages.create(
        model=MODEL_LEAD,
        max_tokens=2048,
        system="""Tu es un orchestrateur. Décompose la requête utilisateur en 2-4 missions
        INDÉPENDANTES et PARALLÉLISABLES. Retourne du JSON strict :
        {"missions": [{"id": "w1", "mission": "..."}, ...]}""",
        messages=[{"role": "user", "content": user_request}],
    )
    plan_text = plan_response.content[0].text
    plan = json.loads(plan_text[plan_text.find("{"):plan_text.rfind("}")+1])
    print(f"📋 Plan : {len(plan['missions'])} missions\n")

    # ÉTAPE 2 : lancer les workers EN PARALLÈLE
    tools = [{
        "name": "search_web",
        "description": "Recherche web.",
        "input_schema": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
    }]

    worker_tasks = [
        run_worker(m["id"], m["mission"], tools)
        for m in plan["missions"]
    ]
    worker_results = await asyncio.gather(*worker_tasks)

    print(f"✅ {len(worker_results)} workers terminés\n")

    # ÉTAPE 3 : le lead synthétise
    synthesis_input = "\n\n".join([
        f"### Worker {r['worker_id']} ({r['iterations']} itérations)\n{r['summary']}"
        for r in worker_results
    ])

    final = await client.messages.create(
        model=MODEL_LEAD,
        max_tokens=4096,
        system="Tu es un orchestrateur. Synthétise les résultats des workers en réponse cohérente.",
        messages=[{
            "role": "user",
            "content": f"Requête initiale : {user_request}\n\nRésultats des workers :\n{synthesis_input}",
        }],
    )
    return final.content[0].text


if __name__ == "__main__":
    request = "Fais-moi une veille sur les outils français d'observabilité d'agents IA en 2026."
    result = asyncio.run(orchestrate(request))
    print("\n" + "="*60)
    print(result)

C'est ~150 lignes pour reproduire ce que le SDK fait tout seul. Mais maintenant tu sais exactement ce qui se passe.

3. Cas pratique commun : un système "veille marché"

Mettons les deux voies au travail sur le même problème concret.

L'objectif

Une équipe marketing veut une veille hebdomadaire sur ses 3 concurrents. Elle ne veut pas lire 50 articles, elle veut un brief de 1 page par concurrent, livré le lundi matin.

L'architecture

Chaque worker fait : recherche web (5 articles récents) → extraction des annonces / produits / pricing → résumé en 5 bullets. L'orchestrateur agrège les 3 résumés et produit le brief final formaté.

Implémentation côté SDK (Voie A)

options = ClaudeAgentOptions(
    system_prompt="Tu pilotes une veille hebdo sur 3 concurrents listés par l'utilisateur. "
                  "Lance un competitor_watcher par concurrent en parallèle, puis appelle brief_writer.",
    subagents=[competitor_watcher, brief_writer],
    max_budget_usd=3.00,
)
async for msg in query(prompt="Concurrents : Acme, Globex, Initech", options=options):
    print(msg)

Le SDK orchestre tout, tu ne gères rien. ~30 lignes au total avec les définitions de sub-agents.

Implémentation côté Custom (Voie B)

Tu réutilises le orchestrate() que tu as vu plus haut, en spécifiant les missions :

plan = {"missions": [
    {"id": "wAcme", "mission": "Veille hebdo sur Acme : 5 articles récents, annonces, pricing, résumé 5 bullets"},
    {"id": "wGlobex", "mission": "Idem pour Globex"},
    {"id": "wInitech", "mission": "Idem pour Initech"},
]}

~200 lignes au total, mais tu contrôles chaque appel modèle, chaque tool, chaque format de retour.

4. Comparatif final : SDK vs Custom

| Critère | Claude Agent SDK | API brute + MCP | |---|---|---| | Temps de mise en place | 1-2 jours | 1-2 semaines | | Lignes de code (cas veille) | ~30 | ~200 | | Contrôle fin de la boucle | Limité (hooks) | Total | | Compactage contexte auto | ✅ Oui | ❌ À coder | | Sub-agents parallèles natifs | ✅ Oui | À coder avec asyncio.gather | | Plafond budget intégré | ✅ max_budget_usd | ❌ Compteur custom | | Poids déploiement | 270-340 Mo (binaire CLI) | Bibliothèque API seule | | Debug & observabilité | Hooks fournis | Tu instrumentes ce que tu veux | | Réutilisabilité des tools | Via MCP | Via MCP | | Coût en tokens | ~Identique | ~Identique | | Courbe d'apprentissage | Faible | Élevée | | Maintenance | Tu suis les versions SDK | Tu maintiens tout |

Quand choisir le SDK

• Tu démarres et tu veux un MVP en 48h.
• Ton cas d'usage colle bien aux patterns par défaut (recherche, génération, automation classique).
• Tu acceptes la dépendance forte à Anthropic.
• Le poids du binaire n'est pas un blocker.

Quand choisir le Custom

• Tu as une logique métier exotique (orchestration multi-modèles, états persistants en base, intégration profonde à un workflow existant).
• Tu veux pouvoir basculer entre fournisseurs (Claude, GPT, Gemini, Mistral) sans réécrire ton orchestration.
• Tu déploies dans des environnements contraints (Lambda, edge, conteneurs minimaux).
• Tu construis une plateforme dont l'orchestration est le cœur de valeur.

5. Mise en production : ce qui casse en vrai

Tu as un prototype qui marche. Voici ce qui va te tomber dessus en prod.

💸 Budget plafonné, toujours — Côté SDK : max_budget_usd sur chaque session. Côté custom : compteur de tokens incrémenté à chaque appel, kill switch si dépassement. Au-delà : système de quotas par utilisateur, par projet, par jour.

📊 Observabilité dès le jour 1 — Tu dois pouvoir répondre à : "Pourquoi cet agent a-t-il fait ce choix le 14 mars à 11h47 ?". Donc tracing complet : chaque message envoyé/reçu, chaque tool call avec ses arguments, chaque résultat, chaque erreur. Stocke dans une base requêtable (pas juste des logs texte).

🛂 Permissions et hooks de sécurité — Liste blanche des tools utilisables par sub-agent. Validation des arguments de tool avant exécution (un agent peut générer du SQL injection si tu ne valides pas). Sandbox pour l'exécution de code (si tu permets à l'agent d'exécuter).

🧪 Évaluation continue — Construis une suite de 20-50 scénarios de référence que tu rejoues à chaque modification de prompt ou de modèle. Mesure : taux de succès, coût moyen, durée moyenne. Sans ça, tu casses des comportements sans le savoir.

🔁 Compactage et mémoire externe — Pour les sessions longues (> 50 tours), implémente une mémoire externe (fichier, base, vecteur) où l'orchestrateur sauvegarde le plan et les findings. Quand le contexte sature, tu compactes vers un résumé condensé + rechargement de la mémoire externe.

⏱️ Timeouts à tous les étages — Timeout par appel modèle (30s par défaut), timeout par tool call (variable selon le tool), timeout global de session (max 10 min en prod typique). Sans timeouts, un bug réseau te laisse une session ouverte qui consomme indéfiniment.

6. Teste ta compréhension

Pour aller plus loin

• 📘 Doc officielle : Claude Agent SDK sur docs.claude.com
• 📘 Protocol MCP : modelcontextprotocol.io
• 📘 Le post de référence d'Anthropic : "How we built our multi-agent research system" (juin 2025)
• 📘 Exemples de code : github.com/anthropics/claude-agent-sdk-demos

En résumé

• Deux voies viables : SDK officiel (rapide, opinionated) ou API brute + MCP (contrôle total, plus de code).
• Le SDK te livre la boucle nO, le compactage, les sub-agents parallèles, les hooks et le budget plafonné gratuitement.
• En custom, tu reconstruis ces briques (~200 lignes pour un orchestrateur fonctionnel) mais tu maîtrises chaque détail.
• MCP est le standard commun : tes tools écrits en MCP sont portables entre SDK et custom.
• En production, le vrai travail commence : budget, observabilité, permissions, évaluation continue, mémoire externe, timeouts.
• Pattern recommandé : prototyper en SDK, puis migrer vers custom quand les besoins métier deviennent spécifiques.