Trasmettere risposte in tempo reale

Ricevere risposte in tempo reale dall'Agent SDK mentre il testo e le chiamate di strumenti vengono trasmessi

Per impostazione predefinita, l'Agent SDK restituisce un AssistantMessage completo per ogni blocco di contenuto non vuoto, come un blocco di testo o una chiamata di strumento, dopo che Claude ha terminato di generare quel blocco. Per ricevere aggiornamenti incrementali mentre il testo e le chiamate di strumenti vengono generati, abilita lo streaming di messaggi parziali.

[!TIP]

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

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "List the files in my project",
  options: {
    includePartialMessages: true,
    allowedTools: ["Bash", "Read"]
  }
})) {
  if (message.type === "stream_event") {
    const event = message.event;
    if (event.type === "content_block_delta") {
      if (event.delta.type === "text_delta") {
        process.stdout.write(event.delta.text);
      }
    }
  }
}

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.

Il campo parent_tool_use_id è sempre None in Python e null in TypeScript. Gli eventi di streaming vengono emessi solo per la sessione principale; i delta a livello di token dai subagent non vengono inoltrati. Per attribuire l'output a un subagent, utilizza messaggi completi, che contengono parent_tool_use_id. Vedi Rilevare l'invocazione di subagent.

Claude Code imposta user_message_uuid sul primo evento di streaming non-ping del turno, e di nuovo quando il messaggio a cui il turno sta rispondendo cambia, secondo le condizioni in user_message_uuid. Il StreamEvent Python non espone questo campo.

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

Flusso dei messaggi

Claude Code emette un AssistantMessage quando ogni blocco di contenuto non vuoto si completa, quindi una risposta con un blocco di testo e una chiamata a uno strumento produce due oggetti AssistantMessage. Ognuno contiene solo il proprio blocco di contenuto, e entrambi condividono lo stesso ID messaggio, che leggete come message.message.id in TypeScript e message.message_id in Python. Con i messaggi parziali abilitati, ogni AssistantMessage arriva prima dell'evento content_block_stop di quel blocco, e ricevete i messaggi in questo ordine:

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

Senza i messaggi parziali abilitati, ricevete tutti i tipi di messaggio tranne StreamEvent. I tipi comuni includono SystemMessage (inizializzazione della sessione), AssistantMessage (blocchi di contenuto completi), 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).

Gestire uno stream interrotto

Se uno stream viene interrotto a metà messaggio, ricevi comunque il message_stop di quel messaggio prima che il turno termini. Tra le cause di uno stream interrotto ci sono un turno interrotto, una connessione caduta e un fallback a una richiesta non in streaming. content_block_stop e message_delta possono mancare prima di quel message_stop:

• content_block_stop: arriva per il blocco che era aperto, a meno che quel blocco non sia una chiamata a uno strumento. Se message_stop arriva mentre il blocco di una chiamata a uno strumento è ancora aperto, considera incompleto l'input di quella chiamata.
• message_delta: arriva quando interrompi il turno e può mancare dopo una connessione caduta o un fallback a una richiesta non in streaming, quindi non attenderlo su un messaggio interrotto.

Prima di Claude Code v2.1.290, uno stream interrotto poteva terminare il turno senza message_stop, quindi una risposta che visualizzi a partire dagli eventi di stream poteva restare mostrata come in corso. L'Agent SDK per TypeScript include Claude Code v2.1.290 o successivo a partire dalla v0.3.290, e l'Agent SDK per Python a partire dalla v0.2.164. Se una risposta resta mostrata come in corso dopo la fine del turno, aggiorna l'SDK.

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

import { query } from "@anthropic-ai/claude-agent-sdk";

// Track the current tool and accumulate its input JSON
let currentTool: string | null = null;
let toolInput = "";

for await (const message of query({
  prompt: "Read the README.md file",
  options: {
    includePartialMessages: true,
    allowedTools: ["Read", "Bash"]
  }
})) {
  if (message.type === "stream_event") {
    const event = message.event;

    if (event.type === "content_block_start") {
      // New tool call is starting
      if (event.content_block.type === "tool_use") {
        currentTool = event.content_block.name;
        toolInput = "";
        console.log(`Starting tool: ${currentTool}`);
      }
    } else if (event.type === "content_block_delta") {
      if (event.delta.type === "input_json_delta") {
        // Accumulate JSON input as it streams in
        const chunk = event.delta.partial_json;
        toolInput += chunk;
        console.log(`  Input chunk: ${chunk}`);
      }
    } else if (event.type === "content_block_stop") {
      // Tool call complete - show final input
      if (currentTool) {
        console.log(`Tool ${currentTool} called with: ${toolInput}`);
        currentTool = null;
      }
    }
  }
}

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

import { query } from "@anthropic-ai/claude-agent-sdk";

// Track whether we're currently in a tool call
let inTool = false;

for await (const message of query({
  prompt: "Find all TODO comments in the codebase",
  options: {
    includePartialMessages: true,
    allowedTools: ["Read", "Bash", "Grep"]
  }
})) {
  if (message.type === "stream_event") {
    const event = message.event;

    if (event.type === "content_block_start") {
      if (event.content_block.type === "tool_use") {
        // Tool call is starting - show status indicator
        process.stdout.write(`\n[Using ${event.content_block.name}...]`);
        inTool = true;
      }
    } else if (event.type === "content_block_delta") {
      // Only stream text when not executing a tool
      if (event.delta.type === "text_delta" && !inTool) {
        process.stdout.write(event.delta.text);
      }
    } else if (event.type === "content_block_stop") {
      if (inTool) {
        // Tool call finished
        console.log(" done");
        inTool = false;
      }
    }
  } else if (message.type === "result") {
    // Agent finished all work
    console.log("\n\n--- Complete ---");
  }
}

Limitazioni note

• Structured output: con i messaggi parziali abilitati, il JSON viene trasmesso come chunk input_json_delta non convalidati di una chiamata a uno strumento, e solo il risultato convalidato raggiunge il ResultMessage.structured_output finale. Riconoscere la chiamata allo strumento StructuredOutput spiega come quella chiamata appare nel tuo stream.

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