Référence du SDK Agent - Python

pip install claude-agent-sdk

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeAgentOptions | None = None
) -> AsyncIterator[Message]

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        system_prompt="Vous êtes un développeur Python expert",
        permission_mode='acceptEdits',
        cwd="/home/user/project"
    )

    async for message in query(
        prompt="Créez un serveur web Python",
        options=options
    ):
        print(message)


asyncio.run(main())

def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any]
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]

from claude_agent_sdk import tool
from typing import Any

@tool("greet", "Saluer un utilisateur", {"name": str})
async def greet(args: dict[str, Any]) -> dict[str, Any]:
    return {
        "content": [{
            "type": "text",
            "text": f"Bonjour, {args['name']} !"
        }]
    }

def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig

from claude_agent_sdk import tool, create_sdk_mcp_server

@tool("add", "Additionner deux nombres", {"a": float, "b": float})
async def add(args):
    return {
        "content": [{
            "type": "text",
            "text": f"Somme : {args['a'] + args['b']}"
        }]
    }

@tool("multiply", "Multiplier deux nombres", {"a": float, "b": float})
async def multiply(args):
    return {
        "content": [{
            "type": "text",
            "text": f"Produit : {args['a'] * args['b']}"
        }]
    }

calculator = create_sdk_mcp_server(
    name="calculator",
    version="2.0.0",
    tools=[add, multiply]  # Passer les fonctions décorées
)

# Utiliser avec Claude
options = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["mcp__calc__add", "mcp__calc__multiply"]
)

class ClaudeSDKClient:
    def __init__(self, options: ClaudeAgentOptions | None = None)
    async def connect(self, prompt: str | AsyncIterable[dict] | None = None) -> None
    async def query(self, prompt: str | AsyncIterable[dict], session_id: str = "default") -> None
    async def receive_messages(self) -> AsyncIterator[Message]
    async def receive_response(self) -> AsyncIterator[Message]
    async def interrupt(self) -> None
    async def disconnect(self) -> None

async with ClaudeSDKClient() as client:
    await client.query("Bonjour Claude")
    async for message in client.receive_response():
        print(message)

import asyncio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage

async def main():
    async with ClaudeSDKClient() as client:
        # Première question
        await client.query("Quelle est la capitale de la France ?")
        
        # Traiter la réponse
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude : {block.text}")
        
        # Question de suivi - Claude se souvient du contexte précédent
        await client.query("Quelle est la population de cette ville ?")
        
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude : {block.text}")
        
        # Autre suivi - toujours dans la même conversation
        await client.query("Quels sont quelques monuments célèbres là-bas ?")
        
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude : {block.text}")

asyncio.run(main())

import asyncio
from claude_agent_sdk import ClaudeSDKClient

async def message_stream():
    """Générer des messages dynamiquement."""
    yield {"type": "text", "text": "Analysez les données suivantes :"}
    await asyncio.sleep(0.5)
    yield {"type": "text", "text": "Température : 25°C"}
    await asyncio.sleep(0.5)
    yield {"type": "text", "text": "Humidité : 60%"}
    await asyncio.sleep(0.5)
    yield {"type": "text", "text": "Quels modèles voyez-vous ?"}

async def main():
    async with ClaudeSDKClient() as client:
        # Diffuser l'entrée vers Claude
        await client.query(message_stream())
        
        # Traiter la réponse
        async for message in client.receive_response():
            print(message)
        
        # Suivi dans la même session
        await client.query("Devrions-nous être préoccupés par ces lectures ?")
        
        async for message in client.receive_response():
            print(message)

asyncio.run(main())

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def interruptible_task():
    options = ClaudeAgentOptions(
        allowed_tools=["Bash"],
        permission_mode="acceptEdits"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # Démarrer une tâche de longue durée
        await client.query("Comptez de 1 à 100 lentement")
        
        # Laisser tourner un peu
        await asyncio.sleep(2)
        
        # Interrompre la tâche
        await client.interrupt()
        print("Tâche interrompue !")
        
        # Envoyer une nouvelle commande
        await client.query("Dites juste bonjour à la place")
        
        async for message in client.receive_response():
            # Traiter la nouvelle réponse
            pass

asyncio.run(interruptible_task())

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions
)

async def custom_permission_handler(
    tool_name: str,
    input_data: dict,
    context: dict
):
    """Logique personnalisée pour les permissions d'outils."""

    # Bloquer les écritures dans les répertoires système
    if tool_name == "Write" and input_data.get("file_path", "").startswith("/system/"):
        return {
            "behavior": "deny",
            "message": "Écriture dans le répertoire système non autorisée",
            "interrupt": True
        }

    # Rediriger les opérations de fichiers sensibles
    if tool_name in ["Write", "Edit"] and "config" in input_data.get("file_path", ""):
        safe_path = f"./sandbox/{input_data['file_path']}"
        return {
            "behavior": "allow",
            "updatedInput": {**input_data, "file_path": safe_path}
        }

    # Autoriser tout le reste
    return {
        "behavior": "allow",
        "updatedInput": input_data
    }

async def main():
    options = ClaudeAgentOptions(
        can_use_tool=custom_permission_handler,
        allowed_tools=["Read", "Write", "Edit"]
    )
    
    async with ClaudeSDKClient(options=options) as client:
        await client.query("Mettez à jour le fichier de configuration système")
        
        async for message in client.receive_response():
            # Utilisera le chemin sandbox à la place
            print(message)

asyncio.run(main())

@dataclass
class SdkMcpTool(Generic[T]):
    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[[T], Awaitable[dict[str, Any]]]

@dataclass
class ClaudeAgentOptions:
    allowed_tools: list[str] = field(default_factory=list)
    max_thinking_tokens: int = 8000
    system_prompt: str | None = None
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    permission_mode: PermissionMode | None = None
    continue_conversation: bool = False
    resume: str | None = None
    fork_session: bool = False
    max_turns: int | None = None
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    permission_prompt_tool_name: str | None = None
    cwd: str | Path | None = None
    settings: str | None = None
    add_dirs: list[str | Path] = field(default_factory=list)
    env: dict[str, str] = field(default_factory=dict)
    extra_args: dict[str, str | None] = field(default_factory=dict)

SettingSource = Literal["user", "project", "local"]

# Charger tous les paramètres comme le SDK v0.0.x le faisait
from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="Analysez ce code",
    options=ClaudeAgentOptions(
        setting_sources=["user", "project", "local"]  # Charger tous les paramètres
    )
):
    print(message)

# Charger seulement les paramètres de projet, ignorer utilisateur et local
async for message in query(
    prompt="Exécuter les vérifications CI",
    options=ClaudeAgentOptions(
        setting_sources=["project"]  # Seulement .claude/settings.json
    )
):
    print(message)

# Assurer un comportement cohérent en CI en excluant les paramètres locaux
async for message in query(
    prompt="Exécuter les tests",
    options=ClaudeAgentOptions(
        setting_sources=["project"],  # Seulement les paramètres partagés d'équipe
        permission_mode="bypassPermissions"
    )
):
    print(message)

# Définir tout programmatiquement (comportement par défaut)
# Aucune dépendance du système de fichiers - setting_sources par défaut à None
async for message in query(
    prompt="Examiner cette PR",
    options=ClaudeAgentOptions(
        # setting_sources=None est le défaut, pas besoin de spécifier
        agents={ /* ... */ },
        mcp_servers={ /* ... */ },
        allowed_tools=["Read", "Grep", "Glob"]
    )
):
    print(message)

@dataclass
class AgentDefinition:
    description: str
    prompt: str
    tools: list[str] | None = None
    model: Literal["sonnet", "opus", "haiku", "inherit"] | None = None

PermissionMode = Literal[
    "default",           # Comportement de permission standard
    "acceptEdits",       # Auto-accepter les modifications de fichiers
    "plan",              # Mode planification - pas d'exécution
    "bypassPermissions"  # Contourner toutes les vérifications de permission (utiliser avec prudence)
]

class McpSdkServerConfig(TypedDict):
    type: Literal["sdk"]
    name: str
    instance: Any  # Instance de serveur MCP

McpServerConfig = McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]  # Optionnel pour la compatibilité descendante
    command: str
    args: NotRequired[list[str]]
    env: NotRequired[dict[str, str]]

class McpSSEServerConfig(TypedDict):
    type: Literal["sse"]
    url: str
    headers: NotRequired[dict[str, str]]

class McpHttpServerConfig(TypedDict):
    type: Literal["http"]
    url: str
    headers: NotRequired[dict[str, str]]

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage

@dataclass
class UserMessage:
    content: str | list[ContentBlock]

@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str

@dataclass
class SystemMessage:
    subtype: str
    data: dict[str, Any]

@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

@dataclass
class TextBlock:
    text: str

@dataclass
class ThinkingBlock:
    thinking: str
    signature: str

@dataclass
class ToolUseBlock:
    id: str
    name: str
    input: dict[str, Any]

@dataclass
class ToolResultBlock:
    tool_use_id: str
    content: str | list[dict[str, Any]] | None = None
    is_error: bool | None = None

class ClaudeSDKError(Exception):
    """Erreur de base pour Claude SDK."""

class CLINotFoundError(CLIConnectionError):
    def __init__(self, message: str = "Claude Code introuvable", cli_path: str | None = None):
        """
        Args:
            message: Message d'erreur (défaut: "Claude Code introuvable")
            cli_path: Chemin optionnel vers le CLI qui n'a pas été trouvé
        """

class CLIConnectionError(ClaudeSDKError):
    """Échec de connexion à Claude Code."""

class ProcessError(ClaudeSDKError):
    def __init__(self, message: str, exit_code: int | None = None, stderr: str | None = None):
        self.exit_code = exit_code
        self.stderr = stderr

class CLIJSONDecodeError(ClaudeSDKError):
    def __init__(self, line: str, original_error: Exception):
        """
        Args:
            line: La ligne qui a échoué à analyser
            original_error: L'exception de décodage JSON originale
        """
        self.line = line
        self.original_error = original_error

HookEvent = Literal[
    "PreToolUse",      # Appelé avant l'exécution d'outil
    "PostToolUse",     # Appelé après l'exécution d'outil
    "UserPromptSubmit", # Appelé quand l'utilisateur soumet un prompt
    "Stop",            # Appelé lors de l'arrêt de l'exécution
    "SubagentStop",    # Appelé quand un sous-agent s'arrête
    "PreCompact"       # Appelé avant la compaction de message
]

HookCallback = Callable[
    [dict[str, Any], str | None, HookContext],
    Awaitable[dict[str, Any]]
]

@dataclass
class HookContext:
    signal: Any | None = None  # Futur : support du signal d'abandon

@dataclass
class HookMatcher:
    matcher: str | None = None        # Nom d'outil ou motif à faire correspondre (ex: "Bash", "Write|Edit")
    hooks: list[HookCallback] = field(default_factory=list)  # Liste des rappels à exécuter

from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher, HookContext
from typing import Any

async def validate_bash_command(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """Valider et potentiellement bloquer les commandes bash dangereuses."""
    if input_data['tool_name'] == 'Bash':
        command = input_data['tool_input'].get('command', '')
        if 'rm -rf /' in command:
            return {
                'hookSpecificOutput': {
                    'hookEventName': 'PreToolUse',
                    'permissionDecision': 'deny',
                    'permissionDecisionReason': 'Commande dangereuse bloquée'
                }
            }
    return {}

async def log_tool_use(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """Enregistrer toute utilisation d'outil pour audit."""
    print(f"Outil utilisé : {input_data.get('tool_name')}")
    return {}

options = ClaudeAgentOptions(
    hooks={
        'PreToolUse': [
            HookMatcher(matcher='Bash', hooks=[validate_bash_command]),
            HookMatcher(hooks=[log_tool_use])  # S'applique à tous les outils
        ],
        'PostToolUse': [
            HookMatcher(hooks=[log_tool_use])
        ]
    }
)

async for message in query(
    prompt="Analysez cette base de code",
    options=options
):
    print(message)

{
    "description": str,      # Une description courte (3-5 mots) de la tâche
    "prompt": str,           # La tâche pour l'agent à effectuer
    "subagent_type": str     # Le type d'agent spécialisé à utiliser
}

{
    "result": str,                    # Résultat final du sous-agent
    "usage": dict | None,             # Statistiques d'utilisation de tokens
    "total_cost_usd": float | None,  # Coût total en USD
    "duration_ms": int | None         # Durée d'exécution en millisecondes
}

{
    "command": str,                  # La commande à exécuter
    "timeout": int | None,           # Timeout optionnel en millisecondes (max 600000)
    "description": str | None,       # Description claire et concise (5-10 mots)
    "run_in_background": bool | None # Définir à true pour exécuter en arrière-plan
}

{
    "output": str,              # Sortie combinée stdout et stderr
    "exitCode": int,            # Code de sortie de la commande
    "killed": bool | None,      # Si la commande a été tuée à cause du timeout
    "shellId": str | None       # ID de shell pour les processus en arrière-plan
}

{
    "file_path": str,           # Le chemin absolu vers le fichier à modifier
    "old_string": str,          # Le texte à remplacer
    "new_string": str,          # Le texte par lequel le remplacer
    "replace_all": bool | None  # Remplacer toutes les occurrences (défaut False)
}

{
    "message": str,      # Message de confirmation
    "replacements": int, # Nombre de remplacements effectués
    "file_path": str     # Chemin de fichier qui a été édité
}

{
    "file_path": str,     # Le chemin absolu vers le fichier à modifier
    "edits": [            # Tableau d'opérations d'édition
        {
            "old_string": str,          # Le texte à remplacer
            "new_string": str,          # Le texte par lequel le remplacer
            "replace_all": bool | None  # Remplacer toutes les occurrences
        }
    ]
}

{
    "message": str,       # Message de succès
    "edits_applied": int, # Nombre total d'éditions appliquées
    "file_path": str      # Chemin de fichier qui a été édité
}

{
    "file_path": str,       # Le chemin absolu vers le fichier à lire
    "offset": int | None,   # Le numéro de ligne à partir duquel commencer la lecture
    "limit": int | None     # Le nombre de lignes à lire
}

{
    "content": str,         # Contenu du fichier avec numéros de ligne
    "total_lines": int,     # Nombre total de lignes dans le fichier
    "lines_returned": int   # Lignes réellement retournées
}

{
    "image": str,       # Données d'image encodées en base64
    "mime_type": str,   # Type MIME de l'image
    "file_size": int    # Taille du fichier en octets
}

{
    "file_path": str,  # Le chemin absolu vers le fichier à écrire
    "content": str     # Le contenu à écrire dans le fichier
}

{
    "message": str,        # Message de succès
    "bytes_written": int,  # Nombre d'octets écrits
    "file_path": str       # Chemin de fichier qui a été écrit
}

{
    "pattern": str,       # Le motif glob pour faire correspondre les fichiers
    "path": str | None    # Le répertoire dans lequel chercher (par défaut cwd)
}

{
    "matches": list[str],  # Tableau des chemins de fichiers correspondants
    "count": int,          # Nombre de correspondances trouvées
    "search_path": str     # Répertoire de recherche utilisé
}

{
    "pattern": str,                    # Le motif d'expression régulière
    "path": str | None,                # Fichier ou répertoire dans lequel chercher
    "glob": str | None,                # Motif glob pour filtrer les fichiers
    "type": str | None,                # Type de fichier à chercher
    "output_mode": str | None,         # "content", "files_with_matches", ou "count"
    "-i": bool | None,                 # Recherche insensible à la casse
    "-n": bool | None,                 # Afficher les numéros de ligne
    "-B": int | None,                  # Lignes à afficher avant chaque correspondance
    "-A": int | None,                  # Lignes à afficher après chaque correspondance
    "-C": int | None,                  # Lignes à afficher avant et après
    "head_limit": int | None,          # Limiter la sortie aux N premières lignes/entrées
    "multiline": bool | None           # Activer le mode multiligne
}

{
    "matches": [
        {
            "file": str,
            "line_number": int | None,
            "line": str,
            "before_context": list[str] | None,
            "after_context": list[str] | None
        }
    ],
    "total_matches": int
}

{
    "files": list[str],  # Fichiers contenant des correspondances
    "count": int         # Nombre de fichiers avec correspondances
}

{
    "notebook_path": str,                     # Chemin absolu vers le notebook Jupyter
    "cell_id": str | None,                    # L'ID de la cellule à éditer
    "new_source": str,                        # La nouvelle source pour la cellule
    "cell_type": "code" | "markdown" | None,  # Le type de la cellule
    "edit_mode": "replace" | "insert" | "delete" | None  # Type d'opération d'édition
}

{
    "message": str, # Message de succès
    "edit_type": "replaced" | "inserted" | "deleted",  # Type d'édition effectuée
    "cell_id": str | None,                       # ID de cellule qui a été affectée
    "total_cells": int                           # Total de cellules dans le notebook après édition
}

{
    "url": str,     # L'URL pour récupérer le contenu
    "prompt": str   # Le prompt à exécuter sur le contenu récupéré
}

{
    "response": str,           # Réponse du modèle IA au prompt
    "url": str,                # URL qui a été récupérée
    "final_url": str | None,   # URL finale après redirections
    "status_code": int | None  # Code de statut HTTP
}

{
    "query": str,                        # La requête de recherche à utiliser
    "allowed_domains": list[str] | None, # Inclure seulement les résultats de ces domaines
    "blocked_domains": list[str] | None  # Ne jamais inclure les résultats de ces domaines
}

{
    "results": [
        {
            "title": str,
            "url": str,
            "snippet": str,
            "metadata": dict | None
        }
    ],
    "total_results": int,
    "query": str
}

{
    "todos": [
        {
            "content": str, # La description de la tâche
            "status": "pending" | "in_progress" | "completed",  # Statut de la tâche
            "activeForm": str                            # Forme active de la description
        }
    ]
}

{
    "message": str,  # Message de succès
    "stats": {
        "total": int,
        "pending": int,
        "in_progress": int,
        "completed": int
    }
}

{
    "bash_id": str,       # L'ID du shell en arrière-plan
    "filter": str | None  # Regex optionnel pour filtrer les lignes de sortie
}

{
    "output": str, # Nouvelle sortie depuis la dernière vérification
    "status": "running" | "completed" | "failed",       # Statut actuel du shell
    "exitCode": int | None # Code de sortie quand terminé
}

{
    "shell_id": str  # L'ID du shell en arrière-plan à tuer
}

{
    "message": str,  # Message de succès
    "shell_id": str  # ID du shell tué
}

{
    "plan": str  # Le plan à exécuter par l'utilisateur pour approbation
}

{
    "message": str,          # Message de confirmation
    "approved": bool | None  # Si l'utilisateur a approuvé le plan
}

{
    "server": str | None  # Nom de serveur optionnel pour filtrer les ressources
}

{
    "resources": [
        {
            "uri": str,
            "name": str,
            "description": str | None,
            "mimeType": str | None,
            "server": str
        }
    ],
    "total": int
}

{
    "server": str,  # Le nom du serveur MCP
    "uri": str      # L'URI de ressource à lire
}

{
    "contents": [
        {
            "uri": str,
            "mimeType": str | None,
            "text": str | None,
            "blob": str | None
        }
    ],
    "server": str
}

from claude_agent_sdk import ClaudeSDKClient, ClaudeCodeOptions, AssistantMessage, TextBlock
import asyncio

class ConversationSession:
    """Maintient une session de conversation unique avec Claude."""
    
    def __init__(self, options: ClaudeCodeOptions = None):
        self.client = ClaudeSDKClient(options)
        self.turn_count = 0
    
    async def start(self):
        await self.client.connect()
        print("Démarrage de la session de conversation. Claude se souviendra du contexte.")
        print("Commandes : 'exit' pour quitter, 'interrupt' pour arrêter la tâche actuelle, 'new' pour nouvelle session")
        
        while True:
            user_input = input(f"\n[Tour {self.turn_count + 1}] Vous : ")
            
            if user_input.lower() == 'exit':
                break
            elif user_input.lower() == 'interrupt':
                await self.client.interrupt()
                print("Tâche interrompue !")
                continue
            elif user_input.lower() == 'new':
                # Déconnecter et reconnecter pour une session fraîche
                await self.client.disconnect()
                await self.client.connect()
                self.turn_count = 0
                print("Nouvelle session de conversation démarrée (contexte précédent effacé)")
                continue
            
            # Envoyer message - Claude se souvient de tous les messages précédents dans cette session
            await self.client.query(user_input)
            self.turn_count += 1
            
            # Traiter la réponse
            print(f"[Tour {self.turn_count}] Claude : ", end="")
            async for message in self.client.receive_response():
                if isinstance(message, AssistantMessage):
                    for block in message.content:
                        if isinstance(block, TextBlock):
                            print(block.text, end="")
            print()  # Nouvelle ligne après la réponse
        
        await self.client.disconnect()
        print(f"Conversation terminée après {self.turn_count} tours.")

async def main():
    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits"
    )
    session = ConversationSession(options)
    await session.start()

# Exemple de conversation :
# Tour 1 - Vous : "Créez un fichier appelé hello.py"
# Tour 1 - Claude : "Je vais créer un fichier hello.py pour vous..."
# Tour 2 - Vous : "Qu'y a-t-il dans ce fichier ?"  
# Tour 2 - Claude : "Le fichier hello.py que je viens de créer contient..." (se souvient !)
# Tour 3 - Vous : "Ajoutez-y une fonction main"
# Tour 3 - Claude : "Je vais ajouter une fonction main à hello.py..." (sait quel fichier !)

asyncio.run(main())

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    HookMatcher,
    HookContext
)
import asyncio
from typing import Any

async def pre_tool_logger(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """Enregistrer toute utilisation d'outil avant l'exécution."""
    tool_name = input_data.get('tool_name', 'inconnu')
    print(f"[PRE-OUTIL] Sur le point d'utiliser : {tool_name}")

    # Vous pouvez modifier ou bloquer l'exécution d'outil ici
    if tool_name == "Bash" and "rm -rf" in str(input_data.get('tool_input', {})):
        return {
            'hookSpecificOutput': {
                'hookEventName': 'PreToolUse',
                'permissionDecision': 'deny',
                'permissionDecisionReason': 'Commande dangereuse bloquée'
            }
        }
    return {}

async def post_tool_logger(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """Enregistrer les résultats après l'exécution d'outil."""
    tool_name = input_data.get('tool_name', 'inconnu')
    print(f"[POST-OUTIL] Terminé : {tool_name}")
    return {}

async def user_prompt_modifier(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """Ajouter du contexte aux prompts utilisateur."""
    original_prompt = input_data.get('prompt', '')

    # Ajouter un horodatage à tous les prompts
    from datetime import datetime
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    return {
        'hookSpecificOutput': {
            'hookEventName': 'UserPromptSubmit',
            'updatedPrompt': f"[{timestamp}] {original_prompt}"
        }
    }

async def main():
    options = ClaudeAgentOptions(
        hooks={
            'PreToolUse': [
                HookMatcher(hooks=[pre_tool_logger]),
                HookMatcher(matcher='Bash', hooks=[pre_tool_logger])
            ],
            'PostToolUse': [
                HookMatcher(hooks=[post_tool_logger])
            ],
            'UserPromptSubmit': [
                HookMatcher(hooks=[user_prompt_modifier])
            ]
        },
        allowed_tools=["Read", "Write", "Bash"]
    )
    
    async with ClaudeSDKClient(options=options) as client:
        await client.query("Listez les fichiers dans le répertoire actuel")
        
        async for message in client.receive_response():
            # Les hooks enregistreront automatiquement l'utilisation d'outils
            pass

asyncio.run(main())

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    ToolUseBlock,
    ToolResultBlock,
    TextBlock
)
import asyncio

async def monitor_progress():
    options = ClaudeAgentOptions(
        allowed_tools=["Write", "Bash"],
        permission_mode="acceptEdits"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        await client.query(
            "Créez 5 fichiers Python avec différents algorithmes de tri"
        )
        
        # Surveiller la progression en temps réel
        files_created = []
        async for message in client.receive_messages():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, ToolUseBlock):
                        if block.name == "Write":
                            file_path = block.input.get("file_path", "")
                            print(f"🔨 Création : {file_path}")
                    elif isinstance(block, ToolResultBlock):
                        print(f"✅ Exécution d'outil terminée")
                    elif isinstance(block, TextBlock):
                        print(f"💭 Claude dit : {block.text[:100]}...")
            
            # Vérifier si nous avons reçu le résultat final
            if hasattr(message, 'subtype') and message.subtype in ['success', 'error']:
                print(f"\n🎯 Tâche terminée !")
                break

asyncio.run(monitor_progress())

from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ToolUseBlock
import asyncio

async def create_project():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode='acceptEdits',
        cwd="/home/user/project"
    )
    
    async for message in query(
        prompt="Créez une structure de projet Python avec setup.py",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    print(f"Utilisation d'outil : {block.name}")

asyncio.run(create_project())

from claude_agent_sdk import (
    query,
    CLINotFoundError,
    ProcessError,
    CLIJSONDecodeError
)

try:
    async for message in query(prompt="Bonjour"):
        print(message)
except CLINotFoundError:
    print("Veuillez installer Claude Code : npm install -g @anthropic-ai/claude-code")
except ProcessError as e:
    print(f"Processus échoué avec code de sortie : {e.exit_code}")
except CLIJSONDecodeError as e:
    print(f"Échec d'analyse de la réponse : {e}")

from claude_agent_sdk import ClaudeSDKClient
import asyncio

async def interactive_session():
    async with ClaudeSDKClient() as client:
        # Envoyer message initial
        await client.query("Quel temps fait-il ?")
        
        # Traiter les réponses
        async for msg in client.receive_response():
            print(msg)
        
        # Envoyer suivi
        await client.query("Dites-moi en plus à ce sujet")
        
        # Traiter la réponse de suivi
        async for msg in client.receive_response():
            print(msg)

asyncio.run(interactive_session())

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    tool,
    create_sdk_mcp_server,
    AssistantMessage,
    TextBlock
)
import asyncio
from typing import Any

# Définir des outils personnalisés avec le décorateur @tool
@tool("calculate", "Effectuer des calculs mathématiques", {"expression": str})
async def calculate(args: dict[str, Any]) -> dict[str, Any]:
    try:
        result = eval(args["expression"], {"__builtins__": {}})
        return {
            "content": [{
                "type": "text",
                "text": f"Résultat : {result}"
            }]
        }
    except Exception as e:
        return {
            "content": [{
                "type": "text",
                "text": f"Erreur : {str(e)}"
            }],
            "is_error": True
        }

@tool("get_time", "Obtenir l'heure actuelle", {})
async def get_time(args: dict[str, Any]) -> dict[str, Any]:
    from datetime import datetime
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return {
        "content": [{
            "type": "text",
            "text": f"Heure actuelle : {current_time}"
        }]
    }

async def main():
    # Créer un serveur SDK MCP avec des outils personnalisés
    my_server = create_sdk_mcp_server(
        name="utilities",
        version="1.0.0",
        tools=[calculate, get_time]
    )

    # Configurer les options avec le serveur
    options = ClaudeAgentOptions(
        mcp_servers={"utils": my_server},
        allowed_tools=[
            "mcp__utils__calculate",
            "mcp__utils__get_time"
        ]
    )
    
    # Utiliser ClaudeSDKClient pour l'utilisation interactive d'outils
    async with ClaudeSDKClient(options=options) as client:
        await client.query("Combien font 123 * 456 ?")
        
        # Traiter la réponse de calcul
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Calcul : {block.text}")
        
        # Suivi avec requête d'heure
        await client.query("Quelle heure est-il maintenant ?")
        
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Heure : {block.text}")

asyncio.run(main())