Noções Básicas de Respostas de Streaming

9 exemplos para você começar com Respostas de Streaming - 6 básicos e 3 intermediários.

Pré-requisitos

• Instale o SDK: pip install anthropic.
• Defina sua chave de API no ambiente: export ANTHROPIC_API_KEY=sk-ant-....
• Todos os exemplos usam client = anthropic.Anthropic(), que lê a chave do ambiente automaticamente.

Exemplos Básicos

1. Abra um Stream e Imprima Texto

A chamada de streaming mais simples possível: abra um stream e imprima cada fragmento de texto conforme ele chega.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=512,
    messages=[{"role": "user", "content": "Escreva um haicai sobre rios."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()

• client.messages.stream(...) retorna um gerenciador de contexto, não um objeto de resposta.
• stream.text_stream é um iterador de conveniência que produz apenas os fragmentos de texto, ignorando outros tipos de evento.
• flush=True é importante aqui - sem ele, o Python pode armazenar a saída em buffer e frustrar o propósito do streaming.

Relacionado: Como os Eventos Enviados do Servidor Potencializam as Respostas de Streaming do Claude - o que está acontecendo sob este loop.

2. Itere Eventos Brutos em Vez de Texto

Desça de text_stream para o iterador de eventos brutos para ver cada tipo de evento conforme ele chega.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=256,
    messages=[{"role": "user", "content": "Nomeie três números primos."}],
) as stream:
    for event in stream:
        print(event.type)

• Iterar stream diretamente (em vez de stream.text_stream) produz cada evento SSE: message_start, content_block_start, content_block_delta, content_block_stop, message_delta, message_stop.
• Este é o nível que você precisa sempre que se importar com algo além de texto puro - chamadas de ferramentas, raciocínio ou totais de uso.
• stream.text_stream é construído sobre este mesmo iterador, filtrando eventos text_delta.

3. Obtenha a Mensagem Final Montada

Após o streaming, recupere o objeto Message completo exatamente como uma chamada não-streaming retornaria.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=256,
    messages=[{"role": "user", "content": "Resuma a fotossíntese em uma frase."}],
) as stream:
    for _ in stream.text_stream:
        pass
    final_message = stream.get_final_message()
 
print(final_message.content[0].text)
print(final_message.usage)

• get_final_message() deve ser chamado após o stream ter sido totalmente consumido (dentro ou após o término da iteração do bloco with).
• Ele retorna a mesma forma Message que você obteria de client.messages.create(...) sem streaming - útil quando você deseja renderização ao vivo e o objeto final para registro.
• final_message.usage fornece as contagens de tokens que só se tornam finais quando a geração para.

4. Verifique o Tipo de Evento Antes de Agir

Filtre explicitamente por event.type em vez de depender de text_stream, que é o padrão que você estenderá para uso de ferramentas e raciocínio.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=256,
    messages=[{"role": "user", "content": "Qual é o ponto de ebulição da água ao nível do mar?"}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_stop":
            print("\n[stream completo]")

• content_block_delta é um invólucro genérico - sempre verifique event.delta.type também, pois pode ser text_delta, input_json_delta ou thinking_delta.
• message_stop é um local confiável para executar a lógica de limpeza de "stream finalizado" (fechar um spinner de UI, descarregar um buffer).
• Esta forma explícita é o que você usa quando uma resposta pode conter mais do que texto puro.

5. Faça Streaming com um Prompt do Sistema e Múltiplas Mensagens

O streaming funciona com a mesma forma de solicitação de uma chamada normal - prompts do sistema e histórico de várias voltas incluídos.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=300,
    system="Você é um escritor técnico conciso. Responda em duas frases ou menos.",
    messages=[
        {"role": "user", "content": "O que é um bloco de conteúdo?"},
        {"role": "assistant", "content": "É uma unidade do conteúdo de uma mensagem, como texto ou uma chamada de ferramenta."},
        {"role": "user", "content": "E o que identifica a qual bloco um delta pertence?"},
    ],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()

• system e messages são passados exatamente como em uma chamada client.messages.create(...) não-streaming - o streaming muda a entrega, não a forma da solicitação.
• O histórico de várias voltas são apenas entradas anteriores de user/assistant na lista messages, como sempre.
• Mantenha max_tokens razoável para uma demonstração de streaming - um limite menor significa uma saída ao vivo mais curta e fácil de ler.

6. Lide com um Erro Básico de Streaming

Envolva o stream em um try/except para que um erro de API no meio do stream não trave o processo inteiro silenciosamente.

import anthropic
 
client = anthropic.Anthropic()
 
try:
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=200,
        messages=[{"role": "user", "content": "Explique o rate limiting brevemente."}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()
except anthropic.APIStatusError as e:
    print(f"\n[falha no stream: {e.status_code} - {e.message}]")

• anthropic.APIStatusError (e suas subclasses como RateLimitError) podem surgir em qualquer ponto durante a iteração, não apenas quando a solicitação é enviada inicialmente.
• Envolver todo o bloco with, não apenas a chamada inicial, é o que captura erros levantados no meio do stream.
• Esta é a rede de segurança mínima; veja Melhores Práticas de Streaming para estratégias de reconexão e retentativa.

Exemplos Intermediários

7. Acompanhe o Progresso do Streaming com um Loop Estilo Callback

Combine text_stream com uma contagem de caracteres em execução para acionar um indicador de progresso simples.

import anthropic
 
client = anthropic.Anthropic()
 
def stream_with_progress(prompt: str) -> str:
    chars = 0
    chunks = []
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=500,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            chunks.append(text)
            chars += len(text)
            print(f"\r{chars} caracteres recebidos", end="", flush=True)
    print()
    return "".join(chunks)
 
result = stream_with_progress("Descreva o ciclo da água em um parágrafo curto.")
print(result)

• Acumular pedaços em uma lista e juntá-los no final é mais barato do que concatenações de string repetidas para respostas mais longas.
• O retorno de carro \r permite que o contador de progresso se atualize no local em um terminal.
• Este padrão - acumular, relatar progresso, retornar a string final - é a base para conectar um stream a um callback de UI em vez de print.

8. Extraia Totais de Uso de message_delta

Leia o uso de tokens que só fica disponível quando o modelo termina de gerar.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": "Liste três benefícios de APIs de streaming."}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_delta":
            usage = event.usage
            print(f"\n[tokens de saída até agora: {usage.output_tokens}]")

• Eventos message_delta carregam stop_reason e usage cumulativo - campos que não são conhecidos até que a geração esteja terminando.
• Contagens de tokens de entrada chegam mais cedo (em message_start); contagens de tokens de saída finalizam em message_delta.
• Registrar o uso aqui, em vez de apenas após get_final_message(), permite rastrear o custo em tempo quase real para streams de longa duração.

9. Faça Streaming de Duas Solicitações Concorrentemente com asyncio

Use o cliente assíncrono para executar dois streams independentes ao mesmo tempo, em vez de um após o outro.

import asyncio
import anthropic
 
async_client = anthropic.AsyncAnthropic()
 
async def stream_one(prompt: str, label: str) -> None:
    async with async_client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=200,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        async for text in stream.text_stream:
            print(f"[{label}] {text}", end="", flush=True)
 
async def main() -> None:
    await asyncio.gather(
        stream_one("Dê um fato divertido de uma linha sobre Marte.", "marte"),
        stream_one("Dê um fato divertido de uma linha sobre Vênus.", "venus"),
    )
 
asyncio.run(main())

• anthropic.AsyncAnthropic() espelha a API do cliente síncrono, mas cada método de stream é aguardado com async with / async for.
• asyncio.gather executa ambos os streams concorrentemente em conexões separadas, o que é mais rápido do que chamadas síncronas sequenciais quando você tem vários prompts independentes.
• A saída intercalada de dois streams rotulados é uma prévia das preocupações de buffer que um backend de chat multiusuário real precisa resolver.

Relacionado: Escolhendo Entre Anthropic e AsyncAnthropic em Código de Produção - quando usar o cliente assíncrono.

Versões da Pilha: Escrito contra a linha de modelos Claude atual em aproximadamente junho de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (o padrão) e Claude Haiku 4.5 - e o SDK oficial anthropic para Python (última versão 0.x). Nomes de modelos, versões de SDK e preços mudam rapidamente - verifique os detalhes atuais em platform.claude.com/docs antes de confiar neles.