Guias

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 objetos AssistantMessage completos após Claude terminar de gerar cada resposta. Para receber atualizações incrementais conforme o texto e as chamadas de ferramentas são geradas, ative o streaming de mensagens parciais definindo include_partial_messages (Python) ou includePartialMessages (TypeScript) como true nas suas opções.

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:

Verificar o tipo de cada mensagem para distinguir StreamEvent de outros tipos de mensagem
Para StreamEvent, extrair o campo event e verificar seu type
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())

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. Aqui está a estrutura de cada tipo:

@dataclass
class StreamEvent:
    uuid: str                      # Identificador único para este evento
    session_id: str                # Identificador de sessão
    event: dict[str, Any]          # O evento bruto de stream da Claude API
    parent_tool_use_id: str | None # ID da ferramenta pai se de um suagente

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 de 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

Com mensagens parciais ativadas, você recebe mensagens nesta ordem:

StreamEvent (message_start)
StreamEvent (content_block_start) - bloco de texto
StreamEvent (content_block_delta) - chunks de texto...
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - bloco tool_use
StreamEvent (content_block_delta) - chunks de entrada de ferramenta...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage - mensagem completa com todo o conteúdo
... ferramenta executa ...
... mais eventos de streaming para o próximo turno ...
ResultMessage - resultado final

Sem mensagens parciais ativadas (include_partial_messages em Python, includePartialMessages em TypeScript), você recebe todos os tipos de mensagem exceto StreamEvent. Os tipos comuns incluem SystemMessage (inicialização de sessão), AssistantMessage (respostas completas), ResultMessage (resultado final) e CompactBoundaryMessage (indica quando o histórico de conversa foi compactado).

Transmitir respostas de texto

Para exibir texto conforme é gerado, procure por eventos content_block_delta onde delta.type é text_delta. Estes contêm os chunks de texto incrementais. O exemplo abaixo imprime cada chunk conforme chega:

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

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

Construir uma interface 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())

Limitações conhecidas

Alguns recursos do SDK são incompatíveis com streaming:

Extended thinking: quando você define explicitamente max_thinking_tokens (Python) ou maxThinkingTokens (TypeScript), mensagens StreamEvent não são emitidas. Você receberá apenas mensagens completas após cada turno. Observe que o thinking é desativado por padrão no SDK, então o streaming funciona a menos que você o ative.
Structured output: o resultado JSON aparece apenas no ResultMessage.structured_output final, 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

Was this page helpful?

Guias

Transmitir respostas em tempo real

Obtenha respostas em tempo real do Agent SDK conforme o texto e as chamadas de ferramentas são transmitidas

Ativar streaming de saída

Seu código então precisa:

Verificar o tipo de cada mensagem para distinguir StreamEvent de outros tipos de mensagem
Para StreamEvent, extrair o campo event e verificar seu type
Procurar por eventos content_block_delta onde delta.type é text_delta, que contêm os chunks de texto reais

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

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. Aqui está a estrutura de cada tipo:

@dataclass
class StreamEvent:
    uuid: str                      # Identificador único para este evento
    session_id: str                # Identificador de sessão
    event: dict[str, Any]          # O evento bruto de stream da Claude API
    parent_tool_use_id: str | None # ID da ferramenta pai se de um suagente

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 de 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

Com mensagens parciais ativadas, você recebe mensagens nesta ordem:

StreamEvent (message_start)
StreamEvent (content_block_start) - bloco de texto
StreamEvent (content_block_delta) - chunks de texto...
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - bloco tool_use
StreamEvent (content_block_delta) - chunks de entrada de ferramenta...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage - mensagem completa com todo o conteúdo
... ferramenta executa ...
... mais eventos de streaming para o próximo turno ...
ResultMessage - resultado final

Transmitir respostas de texto

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

Transmitir chamadas de ferramentas

content_block_start: ferramenta começa
content_block_delta com input_json_delta: chunks de entrada chegam
content_block_stop: chamada de ferramenta 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())

Construir uma interface de streaming

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

Limitações conhecidas

Alguns recursos do SDK são incompatíveis com streaming:

Extended thinking: quando você define explicitamente max_thinking_tokens (Python) ou maxThinkingTokens (TypeScript), mensagens StreamEvent não são emitidas. Você receberá apenas mensagens completas após cada turno. Observe que o thinking é desativado por padrão no SDK, então o streaming funciona a menos que você o ative.
Structured output: o resultado JSON aparece apenas no ResultMessage.structured_output final, 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

Was this page helpful?