Riferimento SDK Agent - Python

Riferimento API completo per Python Agent SDK, incluse tutte le funzioni, i tipi e le classi.

Installazione

pip install claude-agent-sdk

Scelta tra query() e ClaudeSDKClient

Python SDK fornisce due modi per interagire con Claude Code:

Confronto rapido

Quando usare query() (nuova sessione ogni volta)

Migliore per:

• Domande una tantum dove non hai bisogno della cronologia della conversazione
• Attività indipendenti che non richiedono contesto da scambi precedenti
• Script di automazione semplici
• Quando vuoi un nuovo inizio ogni volta

Quando usare ClaudeSDKClient (conversazione continua)

Migliore per:

• Continuare conversazioni - Quando hai bisogno che Claude ricordi il contesto
• Domande di follow-up - Costruire su risposte precedenti
• Applicazioni interattive - Interfacce chat, REPL
• Logica guidata dalla risposta - Quando l'azione successiva dipende dalla risposta di Claude
• Controllo della sessione - Gestire il ciclo di vita della conversazione in modo esplicito

Funzioni

query()

Crea una nuova sessione per ogni interazione con Claude Code. Restituisce un iteratore asincrono che produce messaggi man mano che arrivano. Ogni chiamata a query() inizia da zero senza memoria di interazioni precedenti.

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

Parametri

Restituisce

Restituisce un AsyncIterator[Message] che produce messaggi dalla conversazione.

Esempio - Con opzioni

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions


async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are an expert Python developer",
        permission_mode="acceptEdits",
        cwd="/home/user/project",
    )

    async for message in query(prompt="Create a Python web server", options=options):
        print(message)


asyncio.run(main())

tool()

Decoratore per definire strumenti MCP con sicurezza dei tipi.

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

Parametri

Opzioni dello schema di input

1. Mappatura di tipo semplice (consigliato):

   {"text": str, "count": int, "enabled": bool}
   

2. Formato JSON Schema (per validazione complessa):

   {
       "type": "object",
       "properties": {
           "text": {"type": "string"},
           "count": {"type": "integer", "minimum": 0},
       },
       "required": ["text"],
   }
   

Restituisce

Una funzione decoratore che avvolge l'implementazione dello strumento e restituisce un'istanza SdkMcpTool.

Esempio

from claude_agent_sdk import tool
from typing import Any


@tool("greet", "Greet a user", {"name": str})
async def greet(args: dict[str, Any]) -> dict[str, Any]:
    return {"content": [{"type": "text", "text": f"Hello, {args['name']}!"}]}

ToolAnnotations

Riesportato da mcp.types (disponibile anche come from claude_agent_sdk import ToolAnnotations). Tutti i campi sono suggerimenti opzionali; i client non dovrebbero fare affidamento su di essi per decisioni di sicurezza.

from claude_agent_sdk import tool, ToolAnnotations
from typing import Any


@tool(
    "search",
    "Search the web",
    {"query": str},
    annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
)
async def search(args: dict[str, Any]) -> dict[str, Any]:
    return {"content": [{"type": "text", "text": f"Results for: {args['query']}"}]}

create_sdk_mcp_server()

Crea un server MCP in-process che viene eseguito all'interno della tua applicazione Python.

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

Parametri

Restituisce

Restituisce un oggetto McpSdkServerConfig che può essere passato a ClaudeAgentOptions.mcp_servers.

Esempio

from claude_agent_sdk import tool, create_sdk_mcp_server


@tool("add", "Add two numbers", {"a": float, "b": float})
async def add(args):
    return {"content": [{"type": "text", "text": f"Sum: {args['a'] + args['b']}"}]}


@tool("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply(args):
    return {"content": [{"type": "text", "text": f"Product: {args['a'] * args['b']}"}]}


calculator = create_sdk_mcp_server(
    name="calculator",
    version="2.0.0",
    tools=[add, multiply],  # Pass decorated functions
)

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

list_sessions()

Elenca le sessioni passate con metadati. Filtra per directory di progetto o elenca le sessioni in tutti i progetti. Sincrono; restituisce immediatamente.

def list_sessions(
    directory: str | None = None,
    limit: int | None = None,
    include_worktrees: bool = True
) -> list[SDKSessionInfo]

Parametri

Tipo di ritorno: SDKSessionInfo

Esempio

Stampa le 10 sessioni più recenti per un progetto. I risultati sono ordinati per last_modified decrescente, quindi il primo elemento è il più recente. Ometti directory per cercare in tutti i progetti.

from claude_agent_sdk import list_sessions

for session in list_sessions(directory="/path/to/project", limit=10):
    print(f"{session.summary} ({session.session_id})")

get_session_messages()

Recupera i messaggi da una sessione passata. Sincrono; restituisce immediatamente.

def get_session_messages(
    session_id: str,
    directory: str | None = None,
    limit: int | None = None,
    offset: int = 0
) -> list[SessionMessage]

Parametri

Tipo di ritorno: SessionMessage

Esempio

from claude_agent_sdk import list_sessions, get_session_messages

sessions = list_sessions(limit=1)
if sessions:
    messages = get_session_messages(sessions[0].session_id)
    for msg in messages:
        print(f"[{msg.type}] {msg.uuid}")

get_session_info()

Legge i metadati per una singola sessione per ID senza scansionare la directory del progetto completo. Sincrono; restituisce immediatamente.

def get_session_info(
    session_id: str,
    directory: str | None = None,
) -> SDKSessionInfo | None

Parametri

Restituisce SDKSessionInfo, o None se la sessione non viene trovata.

Esempio

Cerca i metadati di una singola sessione senza scansionare la directory del progetto. Utile quando hai già un ID di sessione da un'esecuzione precedente.

from claude_agent_sdk import get_session_info

info = get_session_info("550e8400-e29b-41d4-a716-446655440000")
if info:
    print(f"{info.summary} (branch: {info.git_branch}, tag: {info.tag})")

rename_session()

Rinomina una sessione aggiungendo una voce di titolo personalizzato. Le chiamate ripetute sono sicure; il titolo più recente vince. Sincrono.

def rename_session(
    session_id: str,
    title: str,
    directory: str | None = None,
) -> None

Parametri

Genera ValueError se session_id non è un UUID valido o title è vuoto; FileNotFoundError se la sessione non può essere trovata.

Esempio

Rinomina la sessione più recente in modo che sia più facile da trovare in seguito. Il nuovo titolo appare in SDKSessionInfo.custom_title nelle letture successive.

from claude_agent_sdk import list_sessions, rename_session

sessions = list_sessions(directory="/path/to/project", limit=1)
if sessions:
    rename_session(sessions[0].session_id, "Refactor auth module")

tag_session()

Etichetta una sessione. Passa None per cancellare l'etichetta. Le chiamate ripetute sono sicure; l'etichetta più recente vince. Sincrono.

def tag_session(
    session_id: str,
    tag: str | None,
    directory: str | None = None,
) -> None

Parametri

Genera ValueError se session_id non è un UUID valido o tag è vuoto dopo la sanitizzazione; FileNotFoundError se la sessione non può essere trovata.

Esempio

Etichetta una sessione, quindi filtra per quell'etichetta in una lettura successiva. Passa None per cancellare un'etichetta esistente.

from claude_agent_sdk import list_sessions, tag_session

# Tag a session
tag_session("550e8400-e29b-41d4-a716-446655440000", "needs-review")

# Later: find all sessions with that tag
for session in list_sessions(directory="/path/to/project"):
    if session.tag == "needs-review":
        print(session.summary)

Classi

ClaudeSDKClient

Mantiene una sessione di conversazione in più scambi. Questo è l'equivalente Python di come la funzione query() di TypeScript SDK funziona internamente - crea un oggetto client che può continuare le conversazioni.

Caratteristiche principali

• Continuità della sessione: Mantiene il contesto della conversazione in più chiamate query()
• Stessa conversazione: La sessione conserva i messaggi precedenti
• Supporto per interruzioni: Può interrompere l'esecuzione a metà attività
• Ciclo di vita esplicito: Controlli quando la sessione inizia e termina
• Flusso guidato dalla risposta: Può reagire alle risposte e inviare follow-up
• Strumenti personalizzati e hooks: Supporta strumenti personalizzati (creati con il decoratore @tool) e hooks

class ClaudeSDKClient:
    def __init__(self, options: ClaudeAgentOptions | None = None, transport: Transport | 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 set_permission_mode(self, mode: str) -> None
    async def set_model(self, model: str | None = None) -> None
    async def rewind_files(self, user_message_id: str) -> None
    async def get_mcp_status(self) -> McpStatusResponse
    async def reconnect_mcp_server(self, server_name: str) -> None
    async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None
    async def stop_task(self, task_id: str) -> None
    async def get_server_info(self) -> dict[str, Any] | None
    async def disconnect(self) -> None

Metodi

Supporto Context Manager

Il client può essere utilizzato come context manager asincrono per la gestione automatica della connessione:

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

Importante: Quando iteri sui messaggi, evita di usare break per uscire anticipatamente poiché questo può causare problemi di pulizia asyncio. Invece, lascia che l'iterazione si completi naturalmente o usa flag per tracciare quando hai trovato quello che cerchi.

Esempio - Continuare una conversazione

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


async def main():
    async with ClaudeSDKClient() as client:
        # First question
        await client.query("What's the capital of France?")

        # Process response
        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}")

        # Follow-up question - the session retains the previous context
        await client.query("What's the population of that city?")

        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}")

        # Another follow-up - still in the same conversation
        await client.query("What are some famous landmarks there?")

        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())

Esempio - Input streaming con ClaudeSDKClient

import asyncio
from claude_agent_sdk import ClaudeSDKClient


async def message_stream():
    """Generate messages dynamically."""
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Analyze the following data:"},
    }
    await asyncio.sleep(0.5)
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Temperature: 25°C, Humidity: 60%"},
    }
    await asyncio.sleep(0.5)
    yield {
        "type": "user",
        "message": {"role": "user", "content": "What patterns do you see?"},
    }


async def main():
    async with ClaudeSDKClient() as client:
        # Stream input to Claude
        await client.query(message_stream())

        # Process response
        async for message in client.receive_response():
            print(message)

        # Follow-up in same session
        await client.query("Should we be concerned about these readings?")

        async for message in client.receive_response():
            print(message)


asyncio.run(main())

Esempio - Utilizzo di interruzioni

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, ResultMessage


async def interruptible_task():
    options = ClaudeAgentOptions(allowed_tools=["Bash"], permission_mode="acceptEdits")

    async with ClaudeSDKClient(options=options) as client:
        # Start a long-running task
        await client.query("Count from 1 to 100 slowly, using the bash sleep command")

        # Let it run for a bit
        await asyncio.sleep(2)

        # Interrupt the task
        await client.interrupt()
        print("Task interrupted!")

        # Drain the interrupted task's messages (including its ResultMessage)
        async for message in client.receive_response():
            if isinstance(message, ResultMessage):
                print(f"Interrupted task finished with subtype={message.subtype!r}")
                # subtype is "error_during_execution" for interrupted tasks

        # Send a new command
        await client.query("Just say hello instead")

        # Now receive the new response
        async for message in client.receive_response():
            if isinstance(message, ResultMessage) and message.subtype == "success":
                print(f"New result: {message.result}")


asyncio.run(interruptible_task())

Esempio - Controllo avanzato delle autorizzazioni

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from claude_agent_sdk.types import (
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext,
)


async def custom_permission_handler(
    tool_name: str, input_data: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    """Custom logic for tool permissions."""

    # Block writes to system directories
    if tool_name == "Write" and input_data.get("file_path", "").startswith("/system/"):
        return PermissionResultDeny(
            message="System directory write not allowed", interrupt=True
        )

    # Redirect sensitive file operations
    if tool_name in ["Write", "Edit"] and "config" in input_data.get("file_path", ""):
        safe_path = f"./sandbox/{input_data['file_path']}"
        return PermissionResultAllow(
            updated_input={**input_data, "file_path": safe_path}
        )

    # Allow everything else
    return PermissionResultAllow(updated_input=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("Update the system config file")

        async for message in client.receive_response():
            # Will use sandbox path instead
            print(message)


asyncio.run(main())

Tipi

SdkMcpTool

Definizione per uno strumento SDK MCP creato con il decoratore @tool.

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

Transport

Classe base astratta per implementazioni di trasporto personalizzate. Usala per comunicare con il processo Claude su un canale personalizzato (ad esempio, una connessione remota invece di un subprocess locale).

from abc import ABC, abstractmethod
from collections.abc import AsyncIterator
from typing import Any


class Transport(ABC):
    @abstractmethod
    async def connect(self) -> None: ...

    @abstractmethod
    async def write(self, data: str) -> None: ...

    @abstractmethod
    def read_messages(self) -> AsyncIterator[dict[str, Any]]: ...

    @abstractmethod
    async def close(self) -> None: ...

    @abstractmethod
    def is_ready(self) -> bool: ...

    @abstractmethod
    async def end_input(self) -> None: ...

Importazione: from claude_agent_sdk import Transport

ClaudeAgentOptions

Dataclass di configurazione per le query Claude Code.

@dataclass
class ClaudeAgentOptions:
    tools: list[str] | ToolsPreset | None = None
    allowed_tools: list[str] = field(default_factory=list)
    system_prompt: str | SystemPromptPreset | None = None
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    strict_mcp_config: bool = False
    permission_mode: PermissionMode | None = None
    continue_conversation: bool = False
    resume: str | None = None
    max_turns: int | None = None
    max_budget_usd: float | None = None
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    fallback_model: str | None = None
    betas: list[SdkBeta] = field(default_factory=list)
    output_format: dict[str, Any] | None = None
    permission_prompt_tool_name: str | None = None
    cwd: str | Path | None = None
    cli_path: 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)
    max_buffer_size: int | None = None
    debug_stderr: Any = sys.stderr  # Deprecated
    stderr: Callable[[str], None] | None = None
    can_use_tool: CanUseTool | None = None
    hooks: dict[HookEvent, list[HookMatcher]] | None = None
    user: str | None = None
    include_partial_messages: bool = False
    include_hook_events: bool = False
    fork_session: bool = False
    agents: dict[str, AgentDefinition] | None = None
    setting_sources: list[SettingSource] | None = None
    sandbox: SandboxSettings | None = None
    plugins: list[SdkPluginConfig] = field(default_factory=list)
    max_thinking_tokens: int | None = None  # Deprecated: use thinking instead
    thinking: ThinkingConfig | None = None
    effort: Literal["low", "medium", "high", "xhigh", "max"] | None = None
    enable_file_checkpointing: bool = False
    session_store: SessionStore | None = None
    session_store_flush: SessionStoreFlushMode = "batched"

Gestisci risposte API lente o bloccate

Il subprocess CLI legge diverse variabili di ambiente che controllano i timeout dell'API e il rilevamento dei blocchi. Passale attraverso ClaudeAgentOptions.env:

options = ClaudeAgentOptions(
    env={
        "API_TIMEOUT_MS": "120000",
        "CLAUDE_CODE_MAX_RETRIES": "2",
        "CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS": "120000",
    },
)

• API_TIMEOUT_MS: timeout per richiesta sul client Anthropic, in millisecondi. Predefinito 600000. Si applica al ciclo principale e a tutti i subagenti.
• CLAUDE_CODE_MAX_RETRIES: numero massimo di tentativi API. Predefinito 10. Ogni tentativo ottiene la propria finestra API_TIMEOUT_MS, quindi il tempo di parete nel caso peggiore è approssimativamente API_TIMEOUT_MS × (CLAUDE_CODE_MAX_RETRIES + 1) più backoff.
• CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS: watchdog di blocco per i subagenti lanciati con run_in_background. Predefinito 600000. Si ripristina su ogni evento di flusso; in caso di blocco interrompe il subagente, contrassegna l'attività come non riuscita e presenta l'errore al genitore con qualsiasi risultato parziale. Non si applica ai subagenti sincroni.
• CLAUDE_ENABLE_STREAM_WATCHDOG=1 con CLAUDE_STREAM_IDLE_TIMEOUT_MS: interrompe la richiesta quando le intestazioni sono arrivate ma il corpo della risposta smette di trasmettere. Disabilitato per impostazione predefinita. CLAUDE_STREAM_IDLE_TIMEOUT_MS predefinito a 300000 e viene bloccato a quel minimo. La richiesta interrotta passa attraverso il percorso di ripetizione normale.

OutputFormat

Configurazione per la validazione dell'output strutturato. Passa questo come dict al campo output_format su ClaudeAgentOptions:

# Expected dict shape for output_format
{
    "type": "json_schema",
    "schema": {...},  # Your JSON Schema definition
}

SystemPromptPreset

Configurazione per l'utilizzo del prompt di sistema preset di Claude Code con aggiunte opzionali.

class SystemPromptPreset(TypedDict):
    type: Literal["preset"]
    preset: Literal["claude_code"]
    append: NotRequired[str]
    exclude_dynamic_sections: NotRequired[bool]

SettingSource

Controlla quali fonti di configurazione basate su filesystem l'SDK carica le impostazioni da.

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

Comportamento predefinito

Quando setting_sources è omesso o None, query() carica le stesse impostazioni del filesystem della CLI di Claude Code: utente, progetto e locale. Le impostazioni della politica gestita vengono caricate in tutti i casi. Vedi Cosa settingSources non controlla per gli input che vengono letti indipendentemente da questa opzione, e come disabilitarli.

Perché usare setting_sources

Disabilita le impostazioni del filesystem:

# Do not load user, project, or local settings from disk
from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="Analyze this code",
    options=ClaudeAgentOptions(
        setting_sources=[]
    ),
):
    print(message)

Carica tutte le impostazioni del filesystem in modo esplicito:

from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="Analyze this code",
    options=ClaudeAgentOptions(
        setting_sources=["user", "project", "local"]
    ),
):
    print(message)

Carica solo fonti di impostazioni specifiche:

# Load only project settings, ignore user and local
async for message in query(
    prompt="Run CI checks",
    options=ClaudeAgentOptions(
        setting_sources=["project"]  # Only .claude/settings.json
    ),
):
    print(message)

Ambienti di test e CI:

# Ensure consistent behavior in CI by excluding local settings
async for message in query(
    prompt="Run tests",
    options=ClaudeAgentOptions(
        setting_sources=["project"],  # Only team-shared settings
        permission_mode="bypassPermissions",
    ),
):
    print(message)

Applicazioni solo SDK:

# Define everything programmatically.
# Pass [] to opt out of filesystem setting sources.
async for message in query(
    prompt="Review this PR",
    options=ClaudeAgentOptions(
        setting_sources=[],
        agents={...},
        mcp_servers={...},
        allowed_tools=["Read", "Grep", "Glob"],
    ),
):
    print(message)

Caricamento delle istruzioni del progetto CLAUDE.md:

# Load project settings to include CLAUDE.md files
async for message in query(
    prompt="Add a new feature following project conventions",
    options=ClaudeAgentOptions(
        system_prompt={
            "type": "preset",
            "preset": "claude_code",  # Use Claude Code's system prompt
        },
        setting_sources=["project"],  # Loads CLAUDE.md from project
        allowed_tools=["Read", "Write", "Edit"],
    ),
):
    print(message)

Precedenza delle impostazioni

Quando più fonti vengono caricate, le impostazioni vengono unite con questa precedenza (da più alta a più bassa):

1. Impostazioni locali (.claude/settings.local.json)
2. Impostazioni di progetto (.claude/settings.json)
3. Impostazioni utente (~/.claude/settings.json)

Le opzioni programmatiche come agents e allowed_tools sovrascrivono le impostazioni del filesystem utente, progetto e locale. Le impostazioni della politica gestita hanno la precedenza sulle opzioni programmatiche.

AgentDefinition

Configurazione per un subagente definito programmaticamente.

@dataclass
class AgentDefinition:
    description: str
    prompt: str
    tools: list[str] | None = None
    disallowedTools: list[str] | None = None
    model: str | None = None
    skills: list[str] | None = None
    memory: Literal["user", "project", "local"] | None = None
    mcpServers: list[str | dict[str, Any]] | None = None
    initialPrompt: str | None = None
    maxTurns: int | None = None
    background: bool | None = None
    effort: Literal["low", "medium", "high", "xhigh", "max"] | int | None = None
    permissionMode: PermissionMode | None = None

PermissionMode

Modalità di autorizzazione per controllare l'esecuzione dello strumento.

PermissionMode = Literal[
    "default",  # Standard permission behavior
    "acceptEdits",  # Auto-accept file edits
    "plan",  # Planning mode - read-only tools only
    "dontAsk",  # Deny anything not pre-approved instead of prompting
    "bypassPermissions",  # Bypass all permission checks (use with caution)
]

CanUseTool

Alias di tipo per le funzioni di callback di autorizzazione dello strumento.

CanUseTool = Callable[
    [str, dict[str, Any], ToolPermissionContext], Awaitable[PermissionResult]
]

Il callback riceve:

• tool_name: Nome dello strumento che viene chiamato
• input_data: I parametri di input dello strumento
• context: Un ToolPermissionContext con informazioni aggiuntive

Restituisce un PermissionResult (sia PermissionResultAllow che PermissionResultDeny).

ToolPermissionContext

Informazioni di contesto passate ai callback di autorizzazione dello strumento.

@dataclass
class ToolPermissionContext:
    signal: Any | None = None  # Future: abort signal support
    suggestions: list[PermissionUpdate] = field(default_factory=list)
    blocked_path: str | None = None
    decision_reason: str | None = None
    title: str | None = None
    display_name: str | None = None
    description: str | None = None

PermissionResult

Tipo di unione per i risultati del callback di autorizzazione.

PermissionResult = PermissionResultAllow | PermissionResultDeny

PermissionResultAllow

Risultato che indica che la chiamata dello strumento deve essere consentita.

@dataclass
class PermissionResultAllow:
    behavior: Literal["allow"] = "allow"
    updated_input: dict[str, Any] | None = None
    updated_permissions: list[PermissionUpdate] | None = None

PermissionResultDeny

Risultato che indica che la chiamata dello strumento deve essere negata.

@dataclass
class PermissionResultDeny:
    behavior: Literal["deny"] = "deny"
    message: str = ""
    interrupt: bool = False

PermissionUpdate

Configurazione per l'aggiornamento delle autorizzazioni a livello di programmazione.

@dataclass
class PermissionUpdate:
    type: Literal[
        "addRules",
        "replaceRules",
        "removeRules",
        "setMode",
        "addDirectories",
        "removeDirectories",
    ]
    rules: list[PermissionRuleValue] | None = None
    behavior: Literal["allow", "deny", "ask"] | None = None
    mode: PermissionMode | None = None
    directories: list[str] | None = None
    destination: (
        Literal["userSettings", "projectSettings", "localSettings", "session"] | None
    ) = None

PermissionRuleValue

Una regola da aggiungere, sostituire o rimuovere in un aggiornamento delle autorizzazioni.

@dataclass
class PermissionRuleValue:
    tool_name: str
    rule_content: str | None = None

ToolsPreset

Configurazione degli strumenti preset per l'utilizzo del set di strumenti predefinito di Claude Code.

class ToolsPreset(TypedDict):
    type: Literal["preset"]
    preset: Literal["claude_code"]

ThinkingConfig

Controlla il comportamento del pensiero esteso. Un'unione di tre configurazioni:

class ThinkingConfigAdaptive(TypedDict):
    type: Literal["adaptive"]


class ThinkingConfigEnabled(TypedDict):
    type: Literal["enabled"]
    budget_tokens: int


class ThinkingConfigDisabled(TypedDict):
    type: Literal["disabled"]


ThinkingConfig = ThinkingConfigAdaptive | ThinkingConfigEnabled | ThinkingConfigDisabled

Poiché queste sono classi TypedDict, sono dicts semplici in fase di esecuzione. Costruiscile come letterali dict o chiama la classe come costruttore; entrambi producono un dict. Accedi ai campi con config["budget_tokens"], non config.budget_tokens:

from claude_agent_sdk import ClaudeAgentOptions, ThinkingConfigEnabled

# Option 1: dict literal (recommended, no import needed)
options = ClaudeAgentOptions(thinking={"type": "enabled", "budget_tokens": 20000})

# Option 2: constructor-style (returns a plain dict)
config = ThinkingConfigEnabled(type="enabled", budget_tokens=20000)
print(config["budget_tokens"])  # 20000
# config.budget_tokens would raise AttributeError

SdkBeta

Tipo letterale per le funzionalità beta dell'SDK.

SdkBeta = Literal["context-1m-2025-08-07"]

Usa con il campo betas in ClaudeAgentOptions per abilitare le funzionalità beta.

McpSdkServerConfig

Configurazione per i server MCP dell'SDK creati con create_sdk_mcp_server().

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

McpServerConfig

Tipo di unione per le configurazioni del server MCP.

McpServerConfig = (
    McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig
)

McpStdioServerConfig

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]  # Optional for backwards compatibility
    command: str
    args: NotRequired[list[str]]
    env: NotRequired[dict[str, str]]

McpSSEServerConfig

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

McpHttpServerConfig

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

McpServerStatusConfig

La configurazione di un server MCP come riportato da get_mcp_status(). Questa è l'unione di tutte le varianti di trasporto McpServerConfig più una variante di output-only claudeai-proxy per i server proxy attraverso claude.ai.

McpServerStatusConfig = (
    McpStdioServerConfig
    | McpSSEServerConfig
    | McpHttpServerConfig
    | McpSdkServerConfigStatus
    | McpClaudeAIProxyServerConfig
)

McpSdkServerConfigStatus è la forma serializzabile di McpSdkServerConfig con solo i campi type ("sdk") e name (str); l'instance in-process viene omesso. McpClaudeAIProxyServerConfig ha i campi type ("claudeai-proxy"), url (str), e id (str).

McpStatusResponse

Risposta da ClaudeSDKClient.get_mcp_status(). Avvolge l'elenco degli stati del server sotto la chiave mcpServers.

class McpStatusResponse(TypedDict):
    mcpServers: list[McpServerStatus]

McpServerStatus

Stato di un server MCP connesso, contenuto in McpStatusResponse.

class McpServerStatus(TypedDict):
    name: str
    status: McpServerConnectionStatus  # "connected" | "failed" | "needs-auth" | "pending" | "disabled"
    serverInfo: NotRequired[McpServerInfo]
    error: NotRequired[str]
    config: NotRequired[McpServerStatusConfig]
    scope: NotRequired[str]
    tools: NotRequired[list[McpToolInfo]]

SdkPluginConfig

Configurazione per il caricamento dei plugin nell'SDK.

class SdkPluginConfig(TypedDict):
    type: Literal["local"]
    path: str

Esempio:

plugins = [
    {"type": "local", "path": "./my-plugin"},
    {"type": "local", "path": "/absolute/path/to/plugin"},
]

Per informazioni complete sulla creazione e l'utilizzo dei plugin, vedi Plugin.

Tipi di messaggio

Message

Tipo di unione di tutti i possibili messaggi.

Message = (
    UserMessage
    | AssistantMessage
    | SystemMessage
    | ResultMessage
    | StreamEvent
    | RateLimitEvent
)

UserMessage

Messaggio di input dell'utente.

@dataclass
class UserMessage:
    content: str | list[ContentBlock]
    uuid: str | None = None
    parent_tool_use_id: str | None = None
    tool_use_result: dict[str, Any] | None = None

AssistantMessage

Messaggio di risposta dell'assistente con blocchi di contenuto.

@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str
    parent_tool_use_id: str | None = None
    error: AssistantMessageError | None = None
    usage: dict[str, Any] | None = None
    message_id: str | None = None

AssistantMessageError

Possibili tipi di errore per i messaggi dell'assistente.

AssistantMessageError = Literal[
    "authentication_failed",
    "billing_error",
    "rate_limit",
    "invalid_request",
    "server_error",
    "max_output_tokens",
    "unknown",
]

SystemMessage

Messaggio di sistema con metadati.

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

ResultMessage

Messaggio di risultato finale con informazioni su costo e utilizzo.

@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    stop_reason: str | None = None
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None
    structured_output: Any = None
    model_usage: dict[str, Any] | None = None
    permission_denials: list[Any] | None = None
    deferred_tool_use: DeferredToolUse | None = None
    errors: list[str] | None = None
    api_error_status: int | None = None
    uuid: str | None = None

Il dict usage contiene le seguenti chiavi quando presenti:

Il dict model_usage mappa i nomi dei modelli all'utilizzo per modello. Le chiavi del dict interno usano camelCase perché il valore viene passato senza modifiche dal processo CLI sottostante, corrispondendo al tipo TypeScript ModelUsage:

StreamEvent

Evento di flusso per aggiornamenti di messaggi parziali durante lo streaming. Ricevuto solo quando include_partial_messages=True in ClaudeAgentOptions. Importa tramite from claude_agent_sdk.types import StreamEvent.

@dataclass
class StreamEvent:
    uuid: str
    session_id: str
    event: dict[str, Any]  # The raw Claude API stream event
    parent_tool_use_id: str | None = None

RateLimitEvent

Emesso quando lo stato del limite di velocità cambia (ad esempio, da "allowed" a "allowed_warning"). Usalo per avvertire gli utenti prima che raggiungano un limite rigido, o per fare backoff quando lo stato è "rejected".

@dataclass
class RateLimitEvent:
    rate_limit_info: RateLimitInfo
    uuid: str
    session_id: str

RateLimitInfo

Stato del limite di velocità trasportato da RateLimitEvent.

RateLimitStatus = Literal["allowed", "allowed_warning", "rejected"]
RateLimitType = Literal[
    "five_hour", "seven_day", "seven_day_opus", "seven_day_sonnet", "overage"
]


@dataclass
class RateLimitInfo:
    status: RateLimitStatus
    resets_at: int | None = None
    rate_limit_type: RateLimitType | None = None
    utilization: float | None = None
    overage_status: RateLimitStatus | None = None
    overage_resets_at: int | None = None
    overage_disabled_reason: str | None = None
    raw: dict[str, Any] = field(default_factory=dict)

TaskStartedMessage

Emesso quando un'attività in background inizia. Un'attività in background è qualsiasi cosa tracciata al di fuori del turno principale: un comando Bash in background, un watch Monitor, un subagente generato tramite lo strumento Agent, o un agente remoto. Il campo task_type ti dice quale. Questo nome non è correlato al rinomina dello strumento Task-to-Agent.

@dataclass
class TaskStartedMessage(SystemMessage):
    task_id: str
    description: str
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    task_type: str | None = None

TaskUsage

Dati di token e timing per un'attività in background.

class TaskUsage(TypedDict):
    total_tokens: int
    tool_uses: int
    duration_ms: int

TaskProgressMessage

Emesso periodicamente con aggiornamenti di progresso per un'attività in background in esecuzione.

@dataclass
class TaskProgressMessage(SystemMessage):
    task_id: str
    description: str
    usage: TaskUsage
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    last_tool_name: str | None = None

TaskNotificationMessage

Emesso quando un'attività in background si completa, fallisce o viene interrotta. Le attività in background includono comandi Bash run_in_background, watch Monitor e subagenti in background.

@dataclass
class TaskNotificationMessage(SystemMessage):
    task_id: str
    status: TaskNotificationStatus  # "completed" | "failed" | "stopped"
    output_file: str
    summary: str
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    usage: TaskUsage | None = None

Tipi di blocco di contenuto

ContentBlock

Tipo di unione di tutti i blocchi di contenuto.

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

TextBlock

Blocco di contenuto di testo.

@dataclass
class TextBlock:
    text: str

ThinkingBlock

Blocco di contenuto di pensiero (per modelli con capacità di pensiero).

@dataclass
class ThinkingBlock:
    thinking: str
    signature: str

ToolUseBlock

Blocco di richiesta di utilizzo dello strumento.

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

ToolResultBlock

Blocco di risultato dell'esecuzione dello strumento.

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

Tipi di errore

ClaudeSDKError

Classe di eccezione base per tutti gli errori dell'SDK.

class ClaudeSDKError(Exception):
    """Base error for Claude SDK."""

CLINotFoundError

Generato quando Claude Code CLI non è installato o non viene trovato.

class CLINotFoundError(CLIConnectionError):
    def __init__(
        self, message: str = "Claude Code not found", cli_path: str | None = None
    ):
        """
        Args:
            message: Error message (default: "Claude Code not found")
            cli_path: Optional path to the CLI that was not found
        """

CLIConnectionError

Generato quando la connessione a Claude Code fallisce.

class CLIConnectionError(ClaudeSDKError):
    """Failed to connect to Claude Code."""

ProcessError

Generato quando il processo Claude Code fallisce.

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

CLIJSONDecodeError

Generato quando l'analisi JSON fallisce.

class CLIJSONDecodeError(ClaudeSDKError):
    def __init__(self, line: str, original_error: Exception):
        """
        Args:
            line: The line that failed to parse
            original_error: The original JSON decode exception
        """
        self.line = line
        self.original_error = original_error

Tipi di Hook

Per una guida completa sull'utilizzo degli hooks con esempi e modelli comuni, vedi la guida Hooks.

HookEvent

Tipi di evento hook supportati.

HookEvent = Literal[
    "PreToolUse",  # Called before tool execution
    "PostToolUse",  # Called after tool execution
    "PostToolUseFailure",  # Called when a tool execution fails
    "UserPromptSubmit",  # Called when user submits a prompt
    "Stop",  # Called when stopping execution
    "SubagentStop",  # Called when a subagent stops
    "PreCompact",  # Called before message compaction
    "Notification",  # Called for notification events
    "SubagentStart",  # Called when a subagent starts
    "PermissionRequest",  # Called when a permission decision is needed
]

HookCallback

Definizione di tipo per le funzioni di callback hook.

HookCallback = Callable[[HookInput, str | None, HookContext], Awaitable[HookJSONOutput]]

Parametri:

• input: Input hook fortemente tipizzato con unioni discriminate basate su hook_event_name (vedi HookInput)
• tool_use_id: Identificatore di utilizzo dello strumento opzionale (per hook correlati allo strumento)
• context: Contesto hook con informazioni aggiuntive

Restituisce un HookJSONOutput che può contenere:

• decision: "block" per bloccare l'azione
• systemMessage: Messaggio di avviso mostrato all'utente
• hookSpecificOutput: Dati di output specifici dell'hook

HookContext

Informazioni di contesto passate ai callback hook.

class HookContext(TypedDict):
    signal: Any | None  # Future: abort signal support

HookMatcher

Configurazione per l'abbinamento degli hook a eventi o strumenti specifici.

@dataclass
class HookMatcher:
    matcher: str | None = (
        None  # Tool name or pattern to match (e.g., "Bash", "Write|Edit")
    )
    hooks: list[HookCallback] = field(
        default_factory=list
    )  # List of callbacks to execute
    timeout: float | None = (
        None  # Timeout in seconds for all hooks in this matcher (default: 60)
    )

HookInput

Tipo di unione di tutti i tipi di input hook. Il tipo effettivo dipende dal campo hook_event_name.

HookInput = (
    PreToolUseHookInput
    | PostToolUseHookInput
    | PostToolUseFailureHookInput
    | UserPromptSubmitHookInput
    | StopHookInput
    | SubagentStopHookInput
    | PreCompactHookInput
    | NotificationHookInput
    | SubagentStartHookInput
    | PermissionRequestHookInput
)

BaseHookInput

Campi di base presenti in tutti i tipi di input hook.

class BaseHookInput(TypedDict):
    session_id: str
    transcript_path: str
    cwd: str
    permission_mode: NotRequired[str]

PreToolUseHookInput

Dati di input per gli eventi hook PreToolUse.

class PreToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PreToolUse"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_use_id: str
    agent_id: NotRequired[str]
    agent_type: NotRequired[str]

PostToolUseHookInput

Dati di input per gli eventi hook PostToolUse.

class PostToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PostToolUse"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_response: Any
    tool_use_id: str
    agent_id: NotRequired[str]
    agent_type: NotRequired[str]

PostToolUseFailureHookInput

Dati di input per gli eventi hook PostToolUseFailure. Chiamato quando l'esecuzione di uno strumento fallisce.

class PostToolUseFailureHookInput(BaseHookInput):
    hook_event_name: Literal["PostToolUseFailure"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_use_id: str
    error: str
    is_interrupt: NotRequired[bool]
    agent_id: NotRequired[str]
    agent_type: NotRequired[str]

UserPromptSubmitHookInput

Dati di input per gli eventi hook UserPromptSubmit.

class UserPromptSubmitHookInput(BaseHookInput):
    hook_event_name: Literal["UserPromptSubmit"]
    prompt: str

StopHookInput

Dati di input per gli eventi hook Stop.

class StopHookInput(BaseHookInput):
    hook_event_name: Literal["Stop"]
    stop_hook_active: bool

SubagentStopHookInput

Dati di input per gli eventi hook SubagentStop.

class SubagentStopHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStop"]
    stop_hook_active: bool
    agent_id: str
    agent_transcript_path: str
    agent_type: str

PreCompactHookInput

Dati di input per gli eventi hook PreCompact.

class PreCompactHookInput(BaseHookInput):
    hook_event_name: Literal["PreCompact"]
    trigger: Literal["manual", "auto"]
    custom_instructions: str | None

NotificationHookInput

Dati di input per gli eventi hook Notification.

class NotificationHookInput(BaseHookInput):
    hook_event_name: Literal["Notification"]
    message: str
    title: NotRequired[str]
    notification_type: str

SubagentStartHookInput

Dati di input per gli eventi hook SubagentStart.

class SubagentStartHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStart"]
    agent_id: str
    agent_type: str

PermissionRequestHookInput

Dati di input per gli eventi hook PermissionRequest. Consente agli hook di gestire le decisioni di autorizzazione a livello di programmazione.

class PermissionRequestHookInput(BaseHookInput):
    hook_event_name: Literal["PermissionRequest"]
    tool_name: str
    tool_input: dict[str, Any]
    permission_suggestions: NotRequired[list[Any]]

HookJSONOutput

Tipo di unione per i valori di ritorno del callback hook.

HookJSONOutput = AsyncHookJSONOutput | SyncHookJSONOutput

SyncHookJSONOutput

Output hook sincrono con campi di controllo e decisione.

class SyncHookJSONOutput(TypedDict):
    # Control fields
    continue_: NotRequired[bool]  # Whether to proceed (default: True)
    suppressOutput: NotRequired[bool]  # Hide stdout from transcript
    stopReason: NotRequired[str]  # Message when continue is False

    # Decision fields
    decision: NotRequired[Literal["block"]]
    systemMessage: NotRequired[str]  # Warning message for user
    reason: NotRequired[str]  # Feedback for Claude

    # Hook-specific output
    hookSpecificOutput: NotRequired[HookSpecificOutput]

HookSpecificOutput

Un TypedDict contenente il nome dell'evento hook e i campi specifici dell'evento. La forma dipende dal valore hookEventName. Per i dettagli completi sui campi disponibili per evento hook, vedi Controlla l'esecuzione con gli hooks.

Un'unione discriminata di tipi di output specifici dell'evento. Il campo hookEventName determina quali campi sono validi.

class PreToolUseHookSpecificOutput(TypedDict):
    hookEventName: Literal["PreToolUse"]
    permissionDecision: NotRequired[Literal["allow", "deny", "ask", "defer"]]
    permissionDecisionReason: NotRequired[str]
    updatedInput: NotRequired[dict[str, Any]]
    additionalContext: NotRequired[str]


class PostToolUseHookSpecificOutput(TypedDict):
    hookEventName: Literal["PostToolUse"]
    additionalContext: NotRequired[str]
    updatedToolOutput: NotRequired[Any]
    updatedMCPToolOutput: NotRequired[Any]


class PostToolUseFailureHookSpecificOutput(TypedDict):
    hookEventName: Literal["PostToolUseFailure"]
    additionalContext: NotRequired[str]


class UserPromptSubmitHookSpecificOutput(TypedDict):
    hookEventName: Literal["UserPromptSubmit"]
    additionalContext: NotRequired[str]


class NotificationHookSpecificOutput(TypedDict):
    hookEventName: Literal["Notification"]
    additionalContext: NotRequired[str]


class SubagentStartHookSpecificOutput(TypedDict):
    hookEventName: Literal["SubagentStart"]
    additionalContext: NotRequired[str]


class PermissionRequestHookSpecificOutput(TypedDict):
    hookEventName: Literal["PermissionRequest"]
    decision: dict[str, Any]


HookSpecificOutput = (
    PreToolUseHookSpecificOutput
    | PostToolUseHookSpecificOutput
    | PostToolUseFailureHookSpecificOutput
    | UserPromptSubmitHookSpecificOutput
    | NotificationHookSpecificOutput
    | SubagentStartHookSpecificOutput
    | PermissionRequestHookSpecificOutput
)

AsyncHookJSONOutput

Output hook asincrono che rinvia l'esecuzione dell'hook.

class AsyncHookJSONOutput(TypedDict):
    async_: Literal[True]  # Set to True to defer execution
    asyncTimeout: NotRequired[int]  # Timeout in milliseconds

Esempio di utilizzo di Hook

Questo esempio registra due hook: uno che blocca i comandi bash pericolosi come rm -rf /, e un altro che registra tutto l'utilizzo dello strumento per il controllo. L'hook di sicurezza viene eseguito solo sui comandi Bash (tramite il matcher), mentre l'hook di registrazione viene eseguito su tutti gli strumenti.

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]:
    """Validate and potentially block dangerous bash commands."""
    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": "Dangerous command blocked",
                }
            }
    return {}


async def log_tool_use(
    input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
    """Log all tool usage for auditing."""
    print(f"Tool used: {input_data.get('tool_name')}")
    return {}


options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash", hooks=[validate_bash_command], timeout=120
            ),  # 2 min for validation
            HookMatcher(
                hooks=[log_tool_use]
            ),  # Applies to all tools (default 60s timeout)
        ],
        "PostToolUse": [HookMatcher(hooks=[log_tool_use])],
    }
)

async for message in query(prompt="Analyze this codebase", options=options):
    print(message)

Tipi di input/output dello strumento

Documentazione degli schemi di input/output per tutti gli strumenti Claude Code integrati. Mentre Python SDK non esporta questi come tipi, rappresentano la struttura degli input e output dello strumento nei messaggi.

Agent

Nome dello strumento: Agent (precedentemente Task, che è ancora accettato come alias)

Input:

{
    "description": str,  # Una breve descrizione del compito (3-5 parole)
    "prompt": str,  # Il compito che l'agente deve eseguire
    "subagent_type": str,  # Il tipo di agente specializzato da utilizzare
}

Output:

{
    "result": str,  # Risultato finale dal subagente
    "usage": dict | None,  # Statistiche di utilizzo dei token
    "total_cost_usd": float | None,  # Costo totale stimato in USD
    "duration_ms": int | None,  # Durata dell'esecuzione in millisecondi
}

AskUserQuestion

Nome dello strumento: AskUserQuestion

Chiede all'utente domande di chiarimento durante l'esecuzione. Vedi Gestisci approvazioni e input dell'utente per i dettagli di utilizzo.

Input:

{
    "questions": [  # Domande da porre all'utente (1-4 domande)
        {
            "question": str,  # La domanda completa da porre all'utente
            "header": str,  # Etichetta molto breve visualizzata come chip/tag (max 12 caratteri)
            "options": [  # Le scelte disponibili (2-4 opzioni)
                {
                    "label": str,  # Testo visualizzato per questa opzione (1-5 parole)
                    "description": str,  # Spiegazione di cosa significa questa opzione
                }
            ],
            "multiSelect": bool,  # Impostare su true per consentire selezioni multiple
        }
    ],
    "answers": dict[str, str | list[str]] | None,
    # Risposte dell'utente popolate dal sistema di autorizzazione. Le risposte
    # multi-select possono essere un elenco di etichette o una stringa unita da virgole
}

Output:

{
    "questions": [  # Le domande che sono state poste
        {
            "question": str,
            "header": str,
            "options": [{"label": str, "description": str}],
            "multiSelect": bool,
        }
    ],
    "answers": dict[str, str],  # Mappa il testo della domanda alla stringa di risposta
    # Le risposte multi-select sono separate da virgole
}

Bash

Nome dello strumento: Bash

Input:

{
    "command": str,  # Il comando da eseguire
    "timeout": int | None,  # Timeout opzionale in millisecondi (max 600000)
    "description": str | None,  # Descrizione chiara e concisa (5-10 parole)
    "run_in_background": bool | None,  # Impostare su true per eseguire in background
}

Output:

{
    "output": str,  # Output combinato di stdout e stderr
    "exitCode": int,  # Codice di uscita del comando
    "killed": bool | None,  # Se il comando è stato terminato a causa del timeout
    "shellId": str | None,  # ID della shell per i processi in background
}

Monitor

Nome dello strumento: Monitor

Esegue uno script in background e fornisce ogni riga stdout a Claude come evento in modo che possa reagire senza polling. Monitor segue le stesse regole di autorizzazione di Bash. Vedi il riferimento dello strumento Monitor per il comportamento e la disponibilità del provider.

Input:

{
    "command": str,  # Script shell; ogni riga stdout è un evento, exit termina il monitoraggio
    "description": str,  # Breve descrizione mostrata nelle notifiche
    "timeout_ms": int | None,  # Termina dopo questa scadenza (default 300000, max 3600000)
    "persistent": bool | None,  # Esegui per la durata della sessione; ferma con TaskStop
}

Output:

{
    "taskId": str,  # ID dell'attività di monitoraggio in background
    "timeoutMs": int,  # Scadenza del timeout in millisecondi (0 quando persistente)
    "persistent": bool | None,  # True quando in esecuzione fino a TaskStop o fine sessione
}

Edit

Nome dello strumento: Edit

Input:

{
    "file_path": str,  # Il percorso assoluto del file da modificare
    "old_string": str,  # Il testo da sostituire
    "new_string": str,  # Il testo con cui sostituirlo
    "replace_all": bool | None,  # Sostituisci tutte le occorrenze (default False)
}

Output:

{
    "message": str,  # Messaggio di conferma
    "replacements": int,  # Numero di sostituzioni effettuate
    "file_path": str,  # Percorso del file che è stato modificato
}

Read

Nome dello strumento: Read

Input:

{
    "file_path": str,  # Il percorso assoluto del file da leggere
    "offset": int | None,  # Il numero di riga da cui iniziare la lettura
    "limit": int | None,  # Il numero di righe da leggere
}

Output (File di testo):

{
    "content": str,  # Contenuto del file con numeri di riga
    "total_lines": int,  # Numero totale di righe nel file
    "lines_returned": int,  # Righe effettivamente restituite
}

Output (Immagini):

{
    "image": str,  # Dati dell'immagine codificati in Base64
    "mime_type": str,  # Tipo MIME dell'immagine
    "file_size": int,  # Dimensione del file in byte
}

Write

Nome dello strumento: Write

Input:

{
    "file_path": str,  # Il percorso assoluto del file da scrivere
    "content": str,  # Il contenuto da scrivere nel file
}

Output:

{
    "message": str,  # Messaggio di successo
    "bytes_written": int,  # Numero di byte scritti
    "file_path": str,  # Percorso del file che è stato scritto
}

Glob

Nome dello strumento: Glob

Input:

{
    "pattern": str,  # Il pattern glob per abbinare i file
    "path": str | None,  # La directory da cercare (default cwd)
}

Output:

{
    "matches": list[str],  # Array dei percorsi dei file corrispondenti
    "count": int,  # Numero di corrispondenze trovate
    "search_path": str,  # Directory di ricerca utilizzata
}

Grep

Nome dello strumento: Grep

Input:

{
    "pattern": str,  # Il pattern di espressione regolare
    "path": str | None,  # File o directory da cercare
    "glob": str | None,  # Pattern glob per filtrare i file
    "type": str | None,  # Tipo di file da cercare
    "output_mode": str | None,  # "content", "files_with_matches", o "count"
    "-i": bool | None,  # Ricerca senza distinzione maiuscole/minuscole
    "-n": bool | None,  # Mostra i numeri di riga
    "-B": int | None,  # Righe da mostrare prima di ogni corrispondenza
    "-A": int | None,  # Righe da mostrare dopo ogni corrispondenza
    "-C": int | None,  # Righe da mostrare prima e dopo
    "head_limit": int | None,  # Limita l'output alle prime N righe/voci
    "multiline": bool | None,  # Abilita la modalità multilinea
}

Output (modalità content):

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

Output (modalità files_with_matches):

{
    "files": list[str],  # File contenenti corrispondenze
    "count": int,  # Numero di file con corrispondenze
}

NotebookEdit

Nome dello strumento: NotebookEdit

Input:

{
    "notebook_path": str,  # Percorso assoluto del notebook Jupyter
    "cell_id": str | None,  # L'ID della cella da modificare
    "new_source": str,  # La nuova sorgente per la cella
    "cell_type": "code" | "markdown" | None,  # Il tipo della cella
    "edit_mode": "replace" | "insert" | "delete" | None,  # Tipo di operazione di modifica
}

Output:

{
    "message": str,  # Messaggio di successo
    "edit_type": "replaced" | "inserted" | "deleted",  # Tipo di modifica eseguita
    "cell_id": str | None,  # ID della cella interessata
    "total_cells": int,  # Numero totale di celle nel notebook dopo la modifica
}

WebFetch

Nome dello strumento: WebFetch

Input:

{
    "url": str,  # L'URL da cui recuperare il contenuto
    "prompt": str,  # Il prompt da eseguire sul contenuto recuperato
}

Output:

{
    "bytes": int,  # Dimensione del contenuto recuperato in byte
    "code": int,  # Codice di risposta HTTP
    "codeText": str,  # Testo del codice di risposta HTTP
    "result": str,  # Risultato elaborato dall'applicazione del prompt al contenuto
    "durationMs": int,  # Tempo per recuperare ed elaborare il contenuto, in millisecondi
    "url": str,  # URL che è stato recuperato
}

WebSearch

Nome dello strumento: WebSearch

Input:

{
    "query": str,  # La query di ricerca da utilizzare
    "allowed_domains": list[str] | None,  # Includi solo risultati da questi domini
    "blocked_domains": list[str] | None,  # Non includere mai risultati da questi domini
}

Output:

{
    "query": str,  # La query di ricerca
    "results": list[str | {"tool_use_id": str, "content": list[{"title": str, "url": str}]}],
    "durationSeconds": float,  # Durata della ricerca in secondi
}

TodoWrite

Nome dello strumento: TodoWrite

Input:

{
    "todos": [
        {
            "content": str,  # La descrizione del compito
            "status": "pending" | "in_progress" | "completed",  # Stato del compito
            "activeForm": str,  # Forma attiva della descrizione
        }
    ]
}

Output:

{
    "message": str,  # Messaggio di successo
    "stats": {"total": int, "pending": int, "in_progress": int, "completed": int},
}

TaskCreate

Nome dello strumento: TaskCreate

Input:

{
    "subject": str,  # Titolo breve del compito
    "description": str,  # Corpo dettagliato del compito
    "activeForm": str | None,  # Etichetta al tempo presente mostrata mentre in corso
    "metadata": dict | None,  # Metadati arbitrari del chiamante
}

Output:

{
    "task": {"id": str, "subject": str},  # Compito creato con ID assegnato
}

TaskUpdate

Nome dello strumento: TaskUpdate

Input:

{
    "taskId": str,  # ID del compito da modificare
    "status": Literal["pending", "in_progress", "completed", "deleted"] | None,
    "subject": str | None,
    "description": str | None,
    "activeForm": str | None,
    "addBlocks": list[str] | None,  # ID dei compiti che questo compito ora blocca
    "addBlockedBy": list[str] | None,  # ID dei compiti che ora bloccano questo compito
    "owner": str | None,
    "metadata": dict | None,
}

Output:

{
    "success": bool,
    "taskId": str,
    "updatedFields": list[str],  # Nomi dei campi che sono cambiati
    "error": str | None,
    "statusChange": {"from": str, "to": str} | None,
}

TaskGet

Nome dello strumento: TaskGet

Input:

{
    "taskId": str,  # ID del compito da leggere
}

Output:

{
    "task": {
        "id": str,
        "subject": str,
        "description": str,
        "status": Literal["pending", "in_progress", "completed"],
        "blocks": list[str],
        "blockedBy": list[str],
    } | None,  # None quando l'ID non viene trovato
}

TaskList

Nome dello strumento: TaskList

Input:

{}

Output:

{
    "tasks": [
        {
            "id": str,
            "subject": str,
            "status": Literal["pending", "in_progress", "completed"],
            "owner": str | None,
            "blockedBy": list[str],
        }
    ],
}

BashOutput

Nome dello strumento: BashOutput

Input:

{
    "bash_id": str,  # L'ID della shell in background
    "filter": str | None,  # Regex opzionale per filtrare le righe di output
}

Output:

{
    "output": str,  # Nuovo output dall'ultimo controllo
    "status": "running" | "completed" | "failed",  # Stato attuale della shell
    "exitCode": int | None,  # Codice di uscita al completamento
}

KillBash

Nome dello strumento: KillBash

Input:

{
    "shell_id": str  # L'ID della shell in background da terminare
}

Output:

{
    "message": str,  # Messaggio di successo
    "shell_id": str,  # ID della shell terminata
}

ExitPlanMode

Nome dello strumento: ExitPlanMode

Input:

{
    "plan": str  # Il piano da eseguire dall'utente per l'approvazione
}

Output:

{
    "message": str,  # Messaggio di conferma
    "approved": bool | None,  # Se l'utente ha approvato il piano
}

ListMcpResources

Nome dello strumento: ListMcpResources

Input:

{
    "server": str | None  # Nome del server opzionale per filtrare le risorse
}

Output:

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

ReadMcpResource

Nome dello strumento: ReadMcpResource

Input:

{
    "server": str,  # Il nome del server MCP
    "uri": str,  # L'URI della risorsa da leggere
}

Output:

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

Funzionalità avanzate con ClaudeSDKClient

Costruire un'interfaccia di conversazione continua

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


class ConversationSession:
    """Maintains a single conversation session with Claude."""

    def __init__(self, options: ClaudeAgentOptions | None = None):
        self.client = ClaudeSDKClient(options)
        self.turn_count = 0

    async def start(self):
        await self.client.connect()
        print("Starting conversation session. Claude will remember context.")
        print(
            "Commands: 'exit' to quit, 'interrupt' to stop current task, 'new' for new session"
        )

        while True:
            user_input = input(f"\n[Turn {self.turn_count + 1}] You: ")

            if user_input.lower() == "exit":
                break
            elif user_input.lower() == "interrupt":
                await self.client.interrupt()
                print("Task interrupted!")
                continue
            elif user_input.lower() == "new":
                # Disconnect and reconnect for a fresh session
                await self.client.disconnect()
                await self.client.connect()
                self.turn_count = 0
                print("Started new conversation session (previous context cleared)")
                continue

            # Send message - the session retains all previous messages
            await self.client.query(user_input)
            self.turn_count += 1

            # Process response
            print(f"[Turn {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()  # New line after response

        await self.client.disconnect()
        print(f"Conversation ended after {self.turn_count} turns.")


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


# Example conversation:
# Turn 1 - You: "Create a file called hello.py"
# Turn 1 - Claude: "I'll create a hello.py file for you..."
# Turn 2 - You: "What's in that file?"
# Turn 2 - Claude: "The hello.py file I just created contains..." (remembers!)
# Turn 3 - You: "Add a main function to it"
# Turn 3 - Claude: "I'll add a main function to hello.py..." (knows which file!)

asyncio.run(main())

Utilizzo di Hook per la modifica del comportamento

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]:
    """Log all tool usage before execution."""
    tool_name = input_data.get("tool_name", "unknown")
    print(f"[PRE-TOOL] About to use: {tool_name}")

    # You can modify or block the tool execution here
    if tool_name == "Bash" and "rm -rf" in str(input_data.get("tool_input", {})):
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": "Dangerous command blocked",
            }
        }
    return {}


async def post_tool_logger(
    input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
    """Log results after tool execution."""
    tool_name = input_data.get("tool_name", "unknown")
    print(f"[POST-TOOL] Completed: {tool_name}")
    return {}


async def user_prompt_modifier(
    input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
    """Add context to user prompts."""
    original_prompt = input_data.get("prompt", "")

    # Add a timestamp as additional context for Claude to see
    from datetime import datetime

    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": f"[Submitted at {timestamp}] Original prompt: {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("List files in current directory")

        async for message in client.receive_response():
            # Hooks will automatically log tool usage
            pass


asyncio.run(main())

Monitoraggio del progresso in tempo reale

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("Create 5 Python files with different sorting algorithms")

        # Monitor progress in real-time
        async for message in client.receive_response():
            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"Creating: {file_path}")
                    elif isinstance(block, ToolResultBlock):
                        print("Completed tool execution")
                    elif isinstance(block, TextBlock):
                        print(f"Claude says: {block.text[:100]}...")

        print("Task completed!")


asyncio.run(monitor_progress())

Utilizzo di esempio

Operazioni di file di base (usando query)

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="Create a Python project structure with setup.py", options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    print(f"Using tool: {block.name}")


asyncio.run(create_project())

Gestione degli errori

from claude_agent_sdk import query, CLINotFoundError, ProcessError, CLIJSONDecodeError

try:
    async for message in query(prompt="Hello"):
        print(message)
except CLINotFoundError:
    print(
        "Claude Code CLI not found. Try reinstalling: pip install --force-reinstall claude-agent-sdk"
    )
except ProcessError as e:
    print(f"Process failed with exit code: {e.exit_code}")
except CLIJSONDecodeError as e:
    print(f"Failed to parse response: {e}")

Modalità streaming con client

from claude_agent_sdk import ClaudeSDKClient
import asyncio


async def interactive_session():
    async with ClaudeSDKClient() as client:
        # Send initial message
        await client.query("What's the weather like?")

        # Process responses
        async for msg in client.receive_response():
            print(msg)

        # Send follow-up
        await client.query("Tell me more about that")

        # Process follow-up response
        async for msg in client.receive_response():
            print(msg)


asyncio.run(interactive_session())

Utilizzo di strumenti personalizzati con ClaudeSDKClient

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


# Define custom tools with @tool decorator
@tool("calculate", "Perform mathematical calculations", {"expression": str})
async def calculate(args: dict[str, Any]) -> dict[str, Any]:
    try:
        result = eval(args["expression"], {"__builtins__": {}})
        return {"content": [{"type": "text", "text": f"Result: {result}"}]}
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Error: {str(e)}"}],
            "is_error": True,
        }


@tool("get_time", "Get current time", {})
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"Current time: {current_time}"}]}


async def main():
    # Create SDK MCP server with custom tools
    my_server = create_sdk_mcp_server(
        name="utilities", version="1.0.0", tools=[calculate, get_time]
    )

    # Configure options with the server
    options = ClaudeAgentOptions(
        mcp_servers={"utils": my_server},
        allowed_tools=["mcp__utils__calculate", "mcp__utils__get_time"],
    )

    # Use ClaudeSDKClient for interactive tool usage
    async with ClaudeSDKClient(options=options) as client:
        await client.query("What's 123 * 456?")

        # Process calculation response
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Calculation: {block.text}")

        # Follow up with time query
        await client.query("What time is it now?")

        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Time: {block.text}")


asyncio.run(main())

Configurazione della Sandbox

SandboxSettings

Configurazione per il comportamento della sandbox. Usala per abilitare il sandboxing dei comandi e configurare le restrizioni di rete a livello di programmazione.

class SandboxSettings(TypedDict, total=False):
    enabled: bool
    autoAllowBashIfSandboxed: bool
    excludedCommands: list[str]
    allowUnsandboxedCommands: bool
    network: SandboxNetworkConfig
    ignoreViolations: SandboxIgnoreViolations
    enableWeakerNestedSandbox: bool

Utilizzo di esempio

from claude_agent_sdk import query, ClaudeAgentOptions, SandboxSettings

sandbox_settings: SandboxSettings = {
    "enabled": True,
    "autoAllowBashIfSandboxed": True,
    "network": {"allowLocalBinding": True},
}

async for message in query(
    prompt="Build and test my project",
    options=ClaudeAgentOptions(sandbox=sandbox_settings),
):
    print(message)

SandboxNetworkConfig

Configurazione specifica della rete per la modalità sandbox.

class SandboxNetworkConfig(TypedDict, total=False):
    allowedDomains: list[str]
    deniedDomains: list[str]
    allowManagedDomainsOnly: bool
    allowUnixSockets: list[str]
    allowAllUnixSockets: bool
    allowLocalBinding: bool
    allowMachLookup: list[str]
    httpProxyPort: int
    socksProxyPort: int

SandboxIgnoreViolations

Configurazione per ignorare violazioni specifiche della sandbox.

class SandboxIgnoreViolations(TypedDict, total=False):
    file: list[str]
    network: list[str]

Fallback delle autorizzazioni per i comandi senza sandbox

Quando allowUnsandboxedCommands è abilitato, il modello può richiedere di eseguire comandi al di fuori della sandbox impostando dangerouslyDisableSandbox: True nell'input dello strumento. Queste richieste ricadono nel sistema di autorizzazioni esistente, il che significa che il tuo handler can_use_tool verrà invocato, permettendoti di implementare una logica di autorizzazione personalizzata.

from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    HookMatcher,
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext,
)


async def can_use_tool(
    tool: str, input: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    # Check if the model is requesting to bypass the sandbox
    if tool == "Bash" and input.get("dangerouslyDisableSandbox"):
        # The model is requesting to run this command outside the sandbox
        print(f"Unsandboxed command requested: {input.get('command')}")

        if is_command_authorized(input.get("command")):
            return PermissionResultAllow()
        return PermissionResultDeny(
            message="Command not authorized for unsandboxed execution"
        )
    return PermissionResultAllow()


# Required: dummy hook keeps the stream open for can_use_tool
async def dummy_hook(input_data, tool_use_id, context):
    return {"continue_": True}


async def prompt_stream():
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Deploy my application"},
    }


async def main():
    async for message in query(
        prompt=prompt_stream(),
        options=ClaudeAgentOptions(
            sandbox={
                "enabled": True,
                "allowUnsandboxedCommands": True,  # Model can request unsandboxed execution
            },
            permission_mode="default",
            can_use_tool=can_use_tool,
            hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[dummy_hook])]},
        ),
    ):
        print(message)

Questo modello ti consente di:

• Controllare le richieste del modello: Registra quando il modello richiede l'esecuzione senza sandbox
• Implementare allowlist: Consenti solo comandi specifici di essere eseguiti senza sandbox
• Aggiungere flussi di lavoro di approvazione: Richiedi un'autorizzazione esplicita per operazioni privilegiate

Vedi anche

• Panoramica dell'SDK - Concetti generali dell'SDK
• Riferimento TypeScript SDK - Documentazione TypeScript SDK
• Riferimento CLI - Interfaccia della riga di comando
• Flussi di lavoro comuni - Guide passo dopo passo