Vai al contenuto principale
Per impostazione predefinita, l’Agent SDK restituisce oggetti AssistantMessage completi dopo che Claude ha terminato 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 Inviare 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:
  1. Controllare il tipo di ogni messaggio per distinguere StreamEvent da altri tipi di messaggio
  2. Per StreamEvent, estrarre il campo event e controllare il suo type
  3. Cercare eventi content_block_delta dove delta.type è text_delta, che contengono i veri frammenti di testo
L’esempio seguente abilita lo streaming e stampa i frammenti di testo mentre arrivano. Nota i controlli di tipo annidati: prima per StreamEvent, poi per content_block_delta, poi per text_delta: 
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())

​
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 da solo. 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 eventoDescrizione
message_startInizio di un nuovo messaggio
content_block_startInizio di un nuovo blocco di contenuto (testo o uso di strumento)
content_block_deltaAggiornamento incrementale al contenuto
content_block_stopFine di un blocco di contenuto
message_deltaAggiornamenti a livello di messaggio (motivo di arresto, utilizzo)
message_stopFine del messaggio

​
Flusso di messaggi

Con i messaggi parziali abilitati, ricevi 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 un messaggio di confine compatto che indica quando la cronologia della conversazione è stata compattata (SDKCompactBoundaryMessage in TypeScript; SystemMessage con sottotipo "compact_boundary" in Python).

​
Trasmettere risposte di testo

Per visualizzare il testo mentre viene generato, cerca eventi content_block_delta dove delta.type è text_delta. Questi contengono i frammenti di testo incrementali. L’esempio seguente stampa ogni frammento mentre arriva: 
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())

​
Trasmettere 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: i frammenti di input arrivano
  • content_block_stop: la chiamata dello strumento è completa
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())

​
Costruire 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. 
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())

​
Limitazioni note

  • 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: