8 de mar. de 2026

Respostas de Agentes em Streaming: Saída em Tempo Real para Fluxos Multi-Etapas

Respostas de Agentes em Streaming: Saída em Tempo Real para Fluxos Multi-Etapas

Seu agente leva 20 segundos para pesquisar uma pergunta, chamar três ferramentas e sintetizar uma resposta. O usuário clica em “Perguntar” e fica olhando para um spinner por 20 segundos. Ele não tem certeza se está funcionando. Considera atualizar a página. Pergunta a si mesmo se deve recomeçar.

Agora imagine o seguinte: o usuário clica em “Perguntar” e, em 500 milissegundos, vê as primeiras palavras do agente aparecerem. Ele acompanha a busca por informações — ”🔍 Pesquisando banco de dados de pedidos…” — e vê os resultados chegando em tempo real. Lê a resposta enquanto ela é escrita, frase por frase. Os mesmos 20 segundos. Uma experiência completamente diferente.

O streaming não é um diferencial para agentes voltados ao usuário — é um requisito de UX. O tempo até o primeiro token é a métrica de latência mais importante em interfaces de agentes. Usuários que veem progresso têm paciência. Usuários que não veem nada desistem. Estudo após estudo sobre desempenho na web mostra que a latência percebida importa mais do que a latência real, e o streaming é a ferramenta mais poderosa que você tem para fechar a lacuna entre as duas.

Neste artigo, você aprenderá a implementar streaming em tempo real para agentes de IA com múltiplas etapas — desde a entrega token a token até a transparência de chamadas de ferramentas, atualizações de status, divulgação progressiva, recuperação de erros e escolhas de camada de transporte.

Seção 1: Fundamentos da API de Streaming do Claude

Antes de fazer streaming de um agente, você precisa fazer streaming de uma única resposta do Claude. Vamos começar pelos fundamentos.

Streaming vs. Sem Streaming

Uma chamada de API sem streaming bloqueia até que toda a resposta seja gerada e, em seguida, a retorna de uma vez. Uma chamada com streaming retorna uma sequência de eventos à medida que a resposta é produzida, começando com o primeiro token.

A diferença no código é mínima. A diferença na experiência do usuário é enorme.

Tipos de Eventos

A API de streaming do Claude emite uma sequência estruturada de eventos server-sent:

message_start — Contém o objeto Message inicial com metadados (modelo, papel, uso).
content_block_start — Sinaliza o início de um bloco de conteúdo (texto ou tool_use).
content_block_delta — Contém conteúdo incremental: fragmentos de texto ou JSON parcial de entrada de ferramenta.
content_block_stop — Sinaliza o fim de um bloco de conteúdo.
message_delta — Atualizações finais da mensagem (motivo de parada, uso final).
message_stop — O stream está completo.

Para respostas em texto, você receberá muitos eventos content_block_delta, cada um carregando um pequeno trecho de texto (geralmente alguns tokens).

Implementação Básica de Streaming

Aqui está um exemplo de streaming síncrono usando o SDK Python da Anthropic:

import anthropic

client = anthropic.Anthropic()

def stream_basic_response(user_message: str):
    """Stream a basic Claude response token by token."""
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()  # Newline after stream completes

stream_basic_response("Explain quantum entanglement in simple terms.")

E a versão assíncrona, que é o que você usará em servidores web em produção:

import anthropic
import asyncio

async_client = anthropic.AsyncAnthropic()

async def stream_basic_response_async(user_message: str):
    """Async streaming with the Anthropic SDK."""
    async with async_client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
    print()

asyncio.run(stream_basic_response_async("Explain quantum entanglement in simple terms."))

Lidando com Tokens Parciais e Limites UTF-8

O SDK cuida da decodificação UTF-8 para você, mas se estiver trabalhando com o stream HTTP bruto, esteja ciente de que caracteres multi-byte podem ser divididos entre chunks. Sempre armazene bytes brutos em buffer e decodifique apenas quando tiver sequências UTF-8 completas. O iterador text_stream do SDK anthropic lida com isso automaticamente — mais um motivo para usá-lo em vez de analisar o stream SSE bruto você mesmo.

Seção 2: Streaming de Chamadas de Ferramentas

O streaming básico de texto é o mínimo esperado. O verdadeiro desafio — e o verdadeiro valor — surge quando seu agente usa ferramentas. Um agente com múltiplas etapas que chama três ferramentas em sequência pode parecer travado durante a execução das ferramentas, a menos que você mostre o que está acontecendo.

Como as Chamadas de Ferramentas Aparecem no Stream

Quando o Claude decide usar uma ferramenta, o stream emite um evento content_block_start com type: "tool_use", seguido de eventos content_block_delta contendo fragmentos do JSON de entrada da ferramenta. Assim que a chamada de ferramenta estiver completamente montada, você executa a ferramenta, injeta o resultado e continua a conversa.

O insight principal: você sabe o nome da ferramenta assim que content_block_start chega, mesmo antes que o JSON de entrada esteja completo. Isso significa que você pode imediatamente mostrar ao usuário algo como ”🔍 Pesquisando banco de dados de pedidos…” sem esperar pela chamada de ferramenta completa.

O Loop Completo do Agente com Streaming

Aqui está um loop completo de agente com streaming que exibe as invocações de ferramentas em tempo real:

import anthropic
import json

client = anthropic.Anthropic()

# Define tools
tools = [
    {
        "name": "search_orders",
        "description": "Search customer orders by order ID or customer email.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Order ID or email"},
            },
            "required": ["query"],
        },
    },
    {
        "name": "get_shipping_status",
        "description": "Get real-time shipping status for an order.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "The order ID"},
            },
            "required": ["order_id"],
        },
    },
]


def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool and return the result as a string."""
    if tool_name == "search_orders":
        # Simulated database lookup
        return json.dumps({
            "order_id": "ORD-12345",
            "customer": "jane@example.com",
            "items": ["Blue Widget x2", "Red Gadget x1"],
            "total": "$47.99",
            "status": "shipped",
        })
    elif tool_name == "get_shipping_status":
        return json.dumps({
            "order_id": tool_input["order_id"],
            "carrier": "FedEx",
            "tracking": "7891011",
            "estimated_delivery": "2026-03-10",
            "current_location": "Memphis, TN",
        })
    return json.dumps({"error": f"Unknown tool: {tool_name}"})


def stream_agent_response(user_message: str):
    """
    Complete streaming agent loop with real-time tool call display.
    """
    messages = [{"role": "user", "content": user_message}]

    while True:
        # Stream the model response
        tool_calls = []
        current_tool = None

        with client.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        ) as stream:
            response = None
            for event in stream:
                # The SDK exposes raw events via iteration
                pass

            # Use the helper to collect text and tool use
            response = stream.get_final_message()

        # Process the response content blocks
        for block in response.content:
            if block.type == "text":
                print(block.text, end="", flush=True)
            elif block.type == "tool_use":
                tool_name = block.name
                tool_input = block.input
                tool_id = block.id

                # Show the user what's happening
                print(f"\n⚙️  Calling tool: {tool_name}({json.dumps(tool_input)})")

                # Execute the tool
                result = execute_tool(tool_name, tool_input)
                print(f"✅  Result received from {tool_name}")

                tool_calls.append({
                    "tool_use_id": tool_id,
                    "tool_name": tool_name,
                    "tool_input": tool_input,
                    "result": result,
                })

        # If the model stopped because it wants to use tools, continue the loop
        if response.stop_reason == "tool_use":
            # Add assistant message with all content blocks
            messages.append({"role": "assistant", "content": response.content})

            # Add tool results
            tool_results = []
            for tc in tool_calls:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": tc["tool_use_id"],
                    "content": tc["result"],
                })
            messages.append({"role": "user", "content": tool_results})

            print("\n--- Continuing agent loop ---")
        else:
            # Model is done (stop_reason == "end_turn")
            print()
            break


# Run the agent
stream_agent_response("Where is my order #ORD-12345? When will it arrive?")

Para streaming verdadeiro token a token com detecção de chamadas de ferramentas, você pode iterar sobre os eventos brutos:

def stream_agent_with_live_tokens(user_message: str):
    """Stream text tokens live while also detecting tool calls."""
    messages = [{"role": "user", "content": user_message}]

    while True:
        collected_content = []
        current_text = ""
        current_tool_name = None
        current_tool_input_json = ""
        current_tool_id = None
        stop_reason = None

        with client.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        ) as stream:
            for event in stream:
                if hasattr(event, 'type'):
                    if event.type == 'content_block_start':
                        if event.content_block.type == 'tool_use':
                            current_

---

## Artigos Relacionados

- [Padrões de Uso de Ferramentas: Interfaces Agente-Ferramenta Confiáveis](/pt/blog/agent-tool-use-patterns/)
- [Padrões Multi-Agente: Orquestradores, Workers e Pipelines](/pt/blog/multi-agent-patterns/)
- [Recuperação de Erros em Agentes: 5 Padrões para Confiabilidade em Produção](/pt/blog/agent-error-recovery-patterns/)
- [Agentes de Automação Web: Controle do Navegador com Claude e Computer Use](/pt/blog/web-automation-agents-browser-control-with-claude-and-computer-use/)