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.

[!TIP]

Esta página aborda o streaming de saída (recebimento de tokens em tempo real). Para modos de entrada (como você envia mensagens), consulte Enviar mensagens para agentes. Você também pode transmitir respostas usando o Agent SDK via CLI.

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:

1. Verificar o tipo de cada mensagem para distinguir StreamEvent de outros tipos de mensagem
2. Para StreamEvent, extrair o campo event e verificar seu type
3. Procurar por eventos content_block_delta onde delta.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 de claude_agent_sdk.types)
• TypeScript: SDKPartialAssistantMessage com type: 'stream_event'

Ambos contêm eventos brutos da Claude API, não texto acumulado. Você precisa extrair e acumular deltas de texto você mesmo.

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:

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

Lidar com um stream interrompido

Se um stream for interrompido no meio de uma mensagem, você ainda recebe o message_stop dessa mensagem antes que o turno termine. As causas de um stream interrompido incluem um turno interrompido, uma conexão que caiu e um fallback para uma requisição sem streaming. content_block_stop e message_delta podem estar ausentes antes desse message_stop:

• content_block_stop: chega para o bloco que estava aberto, a menos que esse bloco seja uma chamada de ferramenta. Se message_stop chegar enquanto o bloco de uma chamada de ferramenta ainda estiver aberto, trate a entrada dessa chamada como incompleta.
• message_delta: chega quando você interrompe o turno e pode estar ausente após uma conexão que caiu ou um fallback para uma requisição sem streaming, portanto não espere por ele em uma mensagem interrompida.

Antes do Claude Code v2.1.290, um stream interrompido podia encerrar o turno sem message_stop, de modo que uma resposta que você renderiza a partir de eventos de stream podia continuar sendo exibida como em andamento. O Agent SDK para TypeScript inclui o Claude Code v2.1.290 ou posterior a partir da v0.3.290, e o Agent SDK para Python a partir da v0.2.164. Se uma resposta continuar sendo exibida como em andamento após o término do turno, atualize o SDK.

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ça
• content_block_delta com input_json_delta: chunks de entrada chegam
• content_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: com mensagens parciais ativadas, o JSON é transmitido como chunks input_json_delta não validados de uma chamada de ferramenta, e apenas o resultado validado chega ao ResultMessage.structured_output final. Reconhecer a chamada de ferramenta StructuredOutput explica como essa chamada aparece no seu stream.

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