Streaming Responses with the Python SDK

Streaming permite que seu aplicativo renderize a resposta de Claude à medida que ela é gerada, em vez de esperar pela resposta completa.

O pacote anthropic expõe isso através de client.messages.stream(), um gerenciador de contexto construído especificamente para consumir uma resposta incrementalmente.

Resumo

Uma chamada não-streaming para messages.create() retorna um objeto Message completo após o modelo ter terminado a geração.

Para respostas longas, isso significa que seu aplicativo fica ocioso, não mostrando nada, até que o último token esteja pronto.

messages.stream() retorna um gerenciador de contexto em vez disso, expondo eventos (e um conveniente iterador text_stream) à medida que chegam do servidor.

A mensagem final, totalmente montada, ainda está disponível assim que o stream termina, então o streaming adiciona uma maneira de observar a resposta incrementalmente sem perder nada do que você obteria de uma chamada normal.

O streaming também é a maneira recomendada de solicitar grandes saídas, pois requisições não-streaming correm o risco de atingir timeouts HTTP do lado do cliente assim que max_tokens sobe para dezenas de milhares.

Receita

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Escreva um poema curto sobre o oceano."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
 
    final_message = stream.get_final_message()
 
print()
print(f"\nStop reason: {final_message.stop_reason}")

Quando usar isso:

• Interfaces de chat onde você quer mostrar texto à medida que é gerado, combinando com a expectativa dos usuários sobre como uma UI conversacional deve se sentir.
• Qualquer requisição com max_tokens acima de aproximadamente 16.000, onde uma chamada não-streaming arrisca um timeout do lado do cliente.
• Geração de longa forma (artigos, relatórios, arquivos de código grandes) onde a saída parcial ainda é útil para o usuário antes que a resposta completa seja concluída.
• Ferramentas CLI que imprimem a saída progressivamente em vez de pausar silenciosamente.

Exemplo de Trabalho

import anthropic
 
client = anthropic.Anthropic()
 
 
def stream_answer(question: str) -> str:
    """Faz streaming de uma resposta para stdout e retorna o texto completo quando terminar."""
    full_text = ""
 
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=4096,
        system="Você é um escritor técnico conciso.",
        messages=[{"role": "user", "content": question}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
            full_text += text
 
        final_message = stream.get_final_message()
 
    print()  # nova linha após a saída transmitida
    print(f"[stop_reason={final_message.stop_reason}, "
          f"output_tokens={final_message.usage.output_tokens}]")
 
    return full_text
 
 
if __name__ == "__main__":
    stream_answer("Explique o que é uma janela de contexto, em três frases.")

O que isso demonstra:

• client.messages.stream(...) é usado como um gerenciador de contexto (with ... as stream:), que lida com a abertura e fechamento da conexão subjacente.
• stream.text_stream produz blocos de texto puro à medida que chegam - sem necessidade de análise manual de eventos para o caso comum.
• stream.get_final_message() (chamado após o loop for terminar) retorna o mesmo objeto Message completo que uma chamada não-streaming retornaria, incluindo stop_reason e usage.
• Acumular blocos em full_text permite que você imprima progressivamente e ainda retorne a string completa para o chamador.

Mergulho Profundo

Como Funciona

• Por baixo dos panos, messages.stream() abre uma conexão HTTP e lê uma série de eventos enviados pelo servidor (message_start, content_block_delta, message_delta, message_stop, e outros) à medida que chegam.
• text_stream é um iterador de conveniência construído sobre esses eventos brutos - ele filtra por deltas de texto e produz apenas o conteúdo da string, para que você não precise analisar os tipos de evento por si mesmo para uso básico.
• O objeto stream armazena em buffer tudo o que viu, o que permite que get_final_message() retorne uma Message completa e tipada após o término do loop, com todos os mesmos campos (content, stop_reason, usage) de uma resposta não-streaming.
• O uso de tokens e a cobrança não são afetados pelo streaming - você paga pelos mesmos tokens de saída, quer os receba como um bloco ou como muitos pequenos blocos.
• Se a conexão cair no meio do stream, a exceção surge da própria iteração (dentro do bloco with), então envolva as chamadas de streaming no mesmo tratamento de erro repetível que você usaria para qualquer outra chamada de API.

Tipos de Evento que Você Pode Acessar Diretamente

Para qualquer coisa além de texto puro, itere o próprio objeto stream em vez de text_stream:

Notas do Python

# Inspecionando eventos brutos em vez de usar text_stream - útil quando você
# precisa de conteúdo de pensamento ou blocos tool_use à medida que eles são transmitidos, não apenas texto.
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Explique recursão."}],
) 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":
            # carrega atualizações incrementais de uso e stop_reason
            pass

Streaming Assíncrono

import asyncio
import anthropic
 
client = anthropic.AsyncAnthropic()
 
 
async def stream_answer(question: str) -> None:
    async with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": question}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
 
    print()
 
asyncio.run(stream_answer("O que é uma corrotina, em duas frases?"))

O streaming AsyncAnthropic() usa async with e async for em vez de with e for - todo o resto sobre a API é o mesmo.

Armadilhas

• Usar with em vez de async with no cliente assíncrono. AsyncAnthropic().messages.stream(...) retorna um gerenciador de contexto assíncrono; abri-lo com with simples levanta um TypeError. Correção: use async with e async for sempre que estiver usando AsyncAnthropic().
• Esquecer flush=True ao imprimir texto transmitido. Sem isso, o Python pode armazenar em buffer o stdout e a saída aparece em rajadas em vez de suavemente. Correção: passe flush=True para print(), ou escreva diretamente para sys.stdout e descarregue explicitamente.
• Ler stream.text_stream duas vezes. O iterador é consumido uma vez; iterá-lo novamente após o loop terminar não produz nada. Correção: acumule o texto em uma variável durante a primeira passagem se precisar dele novamente depois.
• Chamar get_final_message() antes que o stream tenha sido totalmente consumido. Se você sair do loop for cedo, a mensagem final pode estar incompleta ou a chamada pode bloquear esperando pelo resto do stream. Correção: deixe a iteração terminar naturalmente, ou feche explicitamente o stream primeiro se pretender abandoná-lo cedo.
• Assumir que o streaming evita o problema de truncamento de max_tokens. O streaming muda como você recebe a saída, não quanto dela o modelo pode gerar; stop_reason == "max_tokens" ainda pode acontecer. Correção: dimensione max_tokens para a tarefa, independentemente de você transmitir ou não.
• Não lidar com erros de rede durante a iteração. Uma conexão interrompida surge como uma exceção levantada de dentro do loop for text in stream.text_stream:, não antes de começar. Correção: envolva o bloco with no mesmo tratamento de exceção tipada que você usa para chamadas não-streaming.

Alternativas

FAQs

Relacionados

• O Modelo Mental do SDK Python da Anthropic - como o streaming se encaixa com clientes síncronos/assíncronos
• Noções Básicas do SDK Python - sua primeira requisição não-streaming, para comparação
• Lidando com Streaming de Tool Use em Fluxos de Trabalho Python Assíncronos - streaming combinado com tool_use
• Referência de Configuração de Retentativa e Timeout do SDK Python - comportamento de timeout durante streams longos

Versões da Stack: Escrito contra a linha de modelos Claude atual em ~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 (último lançamento 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.