Trasmetti risposte in tempo reale

Per impostazione predefinita, l'Agent SDK restituisce oggetti AssistantMessage completi dopo che Claude ha finito di generare ogni risposta. Per ricevere aggiornamenti incrementali mentre il testo e le chiamate di strumenti vengono generati, abilita lo streaming di messaggi parziali impostando include_partial_messages (Python) o includePartialMessages (TypeScript) su true nelle tue opzioni.

Questa pagina copre lo streaming di output (ricezione di token in tempo reale). Per le modalità di input (come invii messaggi), vedi Invia messaggi agli agenti. Puoi anche trasmettere risposte utilizzando l'Agent SDK tramite la CLI.

Abilita lo streaming di output

Per abilitare lo streaming, imposta include_partial_messages (Python) o includePartialMessages (TypeScript) su true nelle tue opzioni. Questo fa sì che l'SDK restituisca messaggi StreamEvent contenenti eventi API grezzi mentre arrivano, oltre ai soliti AssistantMessage e ResultMessage.

Il tuo codice deve quindi:

Controllare il tipo di ogni messaggio per distinguere StreamEvent da altri tipi di messaggio
Per StreamEvent, estrarre il campo event e controllare il suo type
Cercare eventi content_block_delta dove delta.type è text_delta, che contengono i chunk di testo effettivi

L'esempio seguente abilita lo streaming e stampa i chunk di testo mentre arrivano. Nota i controlli di tipo annidati: prima per StreamEvent, poi per content_block_delta, poi per text_delta:

Riferimento StreamEvent

Quando i messaggi parziali sono abilitati, ricevi eventi di streaming API Claude grezzi avvolti in un oggetto. Il tipo ha nomi diversi in ogni SDK:

Python: StreamEvent (importa da claude_agent_sdk.types)
TypeScript: SDKPartialAssistantMessage con type: 'stream_event'

Entrambi contengono eventi API Claude grezzi, non testo accumulato. Devi estrarre e accumulare i delta di testo tu stesso. Ecco la struttura di ogni tipo:

@dataclass
class StreamEvent:
    uuid: str                      # Unique identifier for this event
    session_id: str                # Session identifier
    event: dict[str, Any]          # The raw Claude API stream event
    parent_tool_use_id: str | None # Parent tool ID if from a subagent

Il campo event contiene l'evento di streaming grezzo dall'API Claude. I tipi di evento comuni includono:

Tipo di evento	Descrizione
`message_start`	Inizio di un nuovo messaggio
`content_block_start`	Inizio di un nuovo blocco di contenuto (testo o uso di strumento)
`content_block_delta`	Aggiornamento incrementale al contenuto
`content_block_stop`	Fine di un blocco di contenuto
`message_delta`	Aggiornamenti a livello di messaggio (motivo di arresto, utilizzo)
`message_stop`	Fine del messaggio

Flusso dei messaggi

Con i messaggi parziali abilitati, ricevi i messaggi in questo ordine:

StreamEvent (message_start)
StreamEvent (content_block_start) - text block
StreamEvent (content_block_delta) - text chunks...
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_use block
StreamEvent (content_block_delta) - tool input chunks...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage - complete message with all content
... tool executes ...
... more streaming events for next turn ...
ResultMessage - final result

Senza i messaggi parziali abilitati (include_partial_messages in Python, includePartialMessages in TypeScript), ricevi tutti i tipi di messaggio tranne StreamEvent. I tipi comuni includono SystemMessage (inizializzazione della sessione), AssistantMessage (risposte complete), ResultMessage (risultato finale) e CompactBoundaryMessage (indica quando la cronologia della conversazione è stata compattata).

Trasmetti risposte di testo

Per visualizzare il testo mentre viene generato, cerca eventi content_block_delta dove delta.type è text_delta. Questi contengono i chunk di testo incrementali. L'esempio seguente stampa ogni chunk mentre arriva:

Trasmetti chiamate di strumenti

Le chiamate di strumenti vengono trasmesse anche in modo incrementale. Puoi tracciare quando gli strumenti iniziano, ricevere il loro input mentre viene generato e vedere quando si completano. L'esempio seguente traccia lo strumento attualmente chiamato e accumula l'input JSON mentre viene trasmesso. Utilizza tre tipi di evento:

content_block_start: lo strumento inizia
content_block_delta con input_json_delta: arrivano i chunk di input
content_block_stop: chiamata dello strumento completata

Costruisci un'interfaccia utente di streaming

Questo esempio combina il testo e lo streaming di strumenti in un'interfaccia utente coerente. Traccia se l'agente sta attualmente eseguendo uno strumento (utilizzando un flag in_tool) per mostrare indicatori di stato come [Using Read...] mentre gli strumenti vengono eseguiti. Il testo viene trasmesso normalmente quando non è in uno strumento e il completamento dello strumento attiva un messaggio "done". Questo modello è utile per le interfacce di chat che devono mostrare lo stato di avanzamento durante attività di agenti multi-step.

Limitazioni note

Alcune funzionalità dell'SDK sono incompatibili con lo streaming:

Extended thinking: quando imposti esplicitamente max_thinking_tokens (Python) o maxThinkingTokens (TypeScript), i messaggi StreamEvent non vengono emessi. Riceverai solo messaggi completi dopo ogni turno. Nota che il thinking è disabilitato per impostazione predefinita nell'SDK, quindi lo streaming funziona a meno che non lo abiliti.
Structured output: il risultato JSON appare solo nel ResultMessage.structured_output finale, non come delta di streaming. Vedi structured outputs per i dettagli.

Passaggi successivi

Ora che puoi trasmettere testo e chiamate di strumenti in tempo reale, esplora questi argomenti correlati:

Interactive vs one-shot queries: scegli tra le modalità di input per il tuo caso d'uso
Structured outputs: ottieni risposte JSON tipizzate dall'agente
Permissions: controlla quali strumenti l'agente può utilizzare

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio

async def stream_response():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Bash", "Read"],
    )

    async for message in query(prompt="List the files in my project", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta.get("text", ""), end="", flush=True)

asyncio.run(stream_response())

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio

async def stream_text():
    options = ClaudeAgentOptions(include_partial_messages=True)

    async for message in query(prompt="Explain how databases work", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    # Print each text chunk as it arrives
                    print(delta.get("text", ""), end="", flush=True)

    print()  # Final newline

asyncio.run(stream_text())

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio

async def stream_tool_calls():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash"],
    )

    # Track the current tool and accumulate its input JSON
    current_tool = None
    tool_input = ""

    async for message in query(prompt="Read the README.md file", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                # New tool call is starting
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    current_tool = content_block.get("name")
                    tool_input = ""
                    print(f"Starting tool: {current_tool}")

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "input_json_delta":
                    # Accumulate JSON input as it streams in
                    chunk = delta.get("partial_json", "")
                    tool_input += chunk
                    print(f"  Input chunk: {chunk}")

            elif event_type == "content_block_stop":
                # Tool call complete - show final input
                if current_tool:
                    print(f"Tool {current_tool} called with: {tool_input}")
                    current_tool = None

asyncio.run(stream_tool_calls())

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
from claude_agent_sdk.types import StreamEvent
import asyncio
import sys

async def streaming_ui():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash", "Grep"],
    )

    # Track whether we're currently in a tool call
    in_tool = False

    async for message in query(
        prompt="Find all TODO comments in the codebase",
        options=options
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    # Tool call is starting - show status indicator
                    tool_name = content_block.get("name")
                    print(f"\n[Using {tool_name}...]", end="", flush=True)
                    in_tool = True

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                # Only stream text when not executing a tool
                if delta.get("type") == "text_delta" and not in_tool:
                    sys.stdout.write(delta.get("text", ""))
                    sys.stdout.flush()

            elif event_type == "content_block_stop":
                if in_tool:
                    # Tool call finished
                    print(" done", flush=True)
                    in_tool = False

        elif isinstance(message, ResultMessage):
            # Agent finished all work
            print(f"\n\n--- Complete ---")

asyncio.run(streaming_ui())