Transmitir respostas em tempo real
Obtenha respostas em tempo real do Agent SDK conforme o texto e as chamadas de ferramentas são transmitidas
Por padrão, o Agent SDK produz uma AssistantMessage completa para cada bloco de conteúdo não vazio, como um bloco de texto ou uma chamada de ferramenta, após Claude terminar de gerar esse bloco. Para receber atualizações incrementais conforme o texto e as chamadas de ferramentas são geradas, ative o streaming de mensagens parciais.
Ativar streaming de saída
Para ativar o streaming, defina include_partial_messages (Python) ou includePartialMessages (TypeScript) como true nas suas opções. Isso faz com que o SDK produza mensagens StreamEvent contendo eventos brutos da API conforme chegam, além das mensagens AssistantMessage e ResultMessage usuais.
Seu código então precisa:
- Verificar o tipo de cada mensagem para distinguir
StreamEventde outros tipos de mensagem - Para
StreamEvent, extrair o campoevente verificar seutype - Procurar por eventos
content_block_deltaondedelta.typeétext_delta, que contêm os chunks de texto reais
O exemplo abaixo ativa o streaming e imprime chunks de texto conforme chegam. Observe as verificações de tipo aninhadas: primeiro para StreamEvent, depois para content_block_delta, depois para 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);
}
}
}
}
Referência de StreamEvent
Quando mensagens parciais estão ativadas, você recebe eventos brutos de streaming da Claude API encapsulados em um objeto. O tipo tem nomes diferentes em cada SDK:
- Python:
StreamEvent(importar declaude_agent_sdk.types) - TypeScript:
SDKPartialAssistantMessagecomtype: 'stream_event'
Ambos contêm eventos brutos da Claude API, não texto acumulado. Você precisa extrair e acumular deltas de texto você mesmo. Aqui está a estrutura de cada 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 # Always None
type SDKPartialAssistantMessage = {
type: "stream_event";
event: BetaRawMessageStreamEvent; // From Anthropic SDK
parent_tool_use_id: string | null;
uuid: UUID;
session_id: string;
ttft_ms?: number; // Time to first token in ms, present only on message_start events
user_message_uuid?: string;
};
O campo parent_tool_use_id é sempre None em Python e null em TypeScript. Eventos de stream são emitidos apenas para a sessão principal; deltas no nível de token de subagentos não são encaminhados. Para atribuir saída a um subagentos, use mensagens completas, que carregam parent_tool_use_id. Veja Detectar invocação de subagentos.
Claude Code define user_message_uuid no primeiro evento de stream não-ping da rodada, e novamente quando a mensagem que a rodada está respondendo muda, sob as condições em user_message_uuid. O StreamEvent do Python não expõe este campo.
O campo event contém o evento de streaming bruto da Claude API. Os tipos de evento comuns incluem:
| Tipo de Evento | Descrição |
|---|---|
message_start |
Início de uma nova mensagem |
content_block_start |
Início de um novo bloco de conteúdo (texto ou uso de ferramenta) |
content_block_delta |
Atualização incremental do conteúdo |
content_block_stop |
Fim de um bloco de conteúdo |
message_delta |
Atualizações no nível da mensagem (motivo de parada, uso) |
message_stop |
Fim da mensagem |
Fluxo de mensagens
Claude Code emite uma AssistantMessage conforme cada bloco de conteúdo não vazio é concluído, portanto uma resposta com um bloco de texto e uma chamada de ferramenta produz dois objetos AssistantMessage. Cada um carrega apenas seu próprio bloco de conteúdo, e ambos compartilham o mesmo ID de mensagem, que você lê como message.message.id em TypeScript e message.message_id em Python. Com mensagens parciais ativadas, cada AssistantMessage chega antes do evento content_block_stop desse bloco, e você recebe mensagens nesta ordem:
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
Sem mensagens parciais ativadas, você recebe todos os tipos de mensagem, exceto StreamEvent. Os tipos comuns incluem SystemMessage (inicialização de sessão), AssistantMessage (blocos de conteúdo completos), ResultMessage (resultado final) e uma mensagem de limite compacta indicando quando o histórico de conversa foi compactado (SDKCompactBoundaryMessage em TypeScript; SystemMessage com subtipo "compact_boundary" em Python).
Transmitir chamadas de ferramentas
As chamadas de ferramentas também são transmitidas incrementalmente. Você pode rastrear quando as ferramentas começam, receber sua entrada conforme é gerada e ver quando são concluídas. O exemplo abaixo rastreia a ferramenta atual sendo chamada e acumula a entrada JSON conforme é transmitida. Ele usa três tipos de evento:
content_block_start: ferramenta começacontent_block_deltacominput_json_delta: chunks de entrada chegamcontent_block_stop: chamada de ferramenta concluída
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;
}
}
}
}
Construir uma interface de usuário de streaming
Este exemplo combina streaming de texto e ferramentas em uma interface coerente. Ele rastreia se o agente está executando uma ferramenta (usando um sinalizador in_tool) para mostrar indicadores de status como [Using Read...] enquanto as ferramentas são executadas. O texto é transmitido normalmente quando não está em uma ferramenta, e a conclusão da ferramenta dispara uma mensagem "done". Este padrão é útil para interfaces de chat que precisam mostrar progresso durante tarefas de agente em várias etapas.
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 ---");
}
}
Limitações conhecidas
- Structured output: o resultado JSON aparece apenas no
ResultMessage.structured_outputfinal, não como deltas de streaming. Consulte structured outputs para detalhes.
Próximas etapas
Agora que você pode transmitir texto e chamadas de ferramentas em tempo real, explore estes tópicos relacionados:
- Interactive vs one-shot queries: escolha entre modos de entrada para seu caso de uso
- Structured outputs: obtenha respostas JSON digitadas do agente
- Permissions: controle quais ferramentas o agente pode usar