Streaming Responses Basics

9 ejemplos para empezar con Streaming Responses: 6 básicos y 3 intermedios.

Prerrequisitos

• Instala el SDK: pip install anthropic.
• Configura tu clave API en el entorno: export ANTHROPIC_API_KEY=sk-ant-....
• Todos los ejemplos usan client = anthropic.Anthropic(), que lee la clave del entorno automáticamente.

Ejemplos Básicos

1. Abrir un Stream e Imprimir Texto

La llamada de streaming más simple posible: abre un stream e imprime cada fragmento de texto a medida que llega.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=512,
    messages=[{"role": "user", "content": "Escribe un haiku sobre ríos."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()

• client.messages.stream(...) devuelve un gestor de contexto, no un objeto de respuesta.
• stream.text_stream es un iterador de conveniencia que produce solo los fragmentos de texto, omitiendo otros tipos de eventos.
• flush=True es importante aquí; sin él, Python puede almacenar en búfer la salida y frustrar el propósito del streaming.

Relacionado: Cómo Server-Sent Events Potencian las Respuestas en Streaming de Claude - lo que sucede debajo de este bucle.

2. Iterar Eventos Crudos en Lugar de Texto

Desciende de text_stream al iterador de eventos crudos para ver cada tipo de evento a medida que llega.

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

• Iterar stream directamente (en lugar de stream.text_stream) produce cada evento SSE: message_start, content_block_start, content_block_delta, content_block_stop, message_delta, message_stop.
• Este es el nivel que necesitas siempre que te importe algo más que texto plano: llamadas a herramientas, pensamiento o totales de uso.
• stream.text_stream se basa en este mismo iterador, filtrando los eventos text_delta.

3. Obtener el Mensaje Final Ensamblado

Después de hacer streaming, recupera el objeto Message completo exactamente como lo devolvería una llamada sin streaming.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=256,
    messages=[{"role": "user", "content": "Resume la fotosíntesis en una oración."}],
) 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() debe llamarse después de que el stream se haya consumido por completo (dentro o después de que el bloque with termine de iterar).
• Devuelve la misma forma Message que obtendrías de client.messages.create(...) sin streaming: útil cuando deseas renderizado en vivo y el objeto final para el registro.
• final_message.usage te da los recuentos de tokens que solo se finalizan una vez que la generación se detiene.

4. Comprobar el Tipo de Evento Antes de Actuar

Filtra explícitamente por event.type en lugar de depender de text_stream, que es el patrón que extenderás para el uso de herramientas y el pensamiento.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=256,
    messages=[{"role": "user", "content": "¿Cuál es el punto de ebullición del agua a nivel del 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 es un sobre genérico; siempre comprueba también event.delta.type, ya que puede ser text_delta, input_json_delta o thinking_delta.
• message_stop es un lugar fiable para ejecutar lógica de limpieza de "stream finalizado" (cerrar un indicador de progreso, vaciar un búfer).
• Esta forma explícita es a la que recurres una vez que una respuesta puede contener más que texto plano.

5. Streaming con un Prompt del Sistema y Múltiples Mensajes

El streaming funciona con la misma forma de solicitud que una llamada normal: incluye prompts del sistema e historial de múltiples turnos.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=300,
    system="Eres un escritor técnico conciso. Responde en dos oraciones o menos.",
    messages=[
        {"role": "user", "content": "¿Qué es un bloque de contenido?"},
        {"role": "assistant", "content": "Es una unidad del contenido de un mensaje, como texto o una llamada a herramienta."},
        {"role": "user", "content": "¿Y qué identifica a qué bloque pertenece un delta?"},
    ],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()

• system y messages se pasan exactamente como en una llamada client.messages.create(...) sin streaming: el streaming cambia la entrega, no la forma de la solicitud.
• El historial de múltiples turnos son simplemente entradas previas de user/assistant en la lista messages, como siempre.
• Mantén max_tokens razonable para una demostración de streaming; un límite más pequeño significa una salida en vivo más corta y fácil de leer.

6. Manejar un Error Básico de Streaming

Envuelve el stream en un try/except para que un error de API a mitad de stream no bloquee silenciosamente todo el proceso.

import anthropic
 
client = anthropic.Anthropic()
 
try:
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=200,
        messages=[{"role": "user", "content": "Explica brevemente la limitación de tasa."}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()
except anthropic.APIStatusError as e:
    print(f"\n[stream fallido: {e.status_code} - {e.message}]")

• anthropic.APIStatusError (y sus subclases como RateLimitError) pueden aparecer en cualquier momento durante la iteración, no solo cuando se envía la solicitud inicialmente.
• Envolver todo el bloque with, no solo la llamada inicial, es lo que captura los errores que se generan a mitad de stream.
• Esta es la red de seguridad mínima; consulta Mejores Prácticas de Streaming para estrategias de reconexión y reintento.

Ejemplos Intermedios

7. Seguir el Progreso del Streaming con un Bucle Estilo Callback

Combina text_stream con un contador de caracteres en ejecución para impulsar un indicador de progreso simple.

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 recibidos", end="", flush=True)
    print()
    return "".join(chunks)
 
result = stream_with_progress("Describe el ciclo del agua en un párrafo corto.")
print(result)

• Acumular fragmentos en una lista y unirlos al final es más económico que la concatenación repetida de cadenas para respuestas más largas.
• El retorno de carro \r permite que el contador de progreso se actualice en el mismo lugar en una terminal.
• Este patrón (acumular, informar progreso, devolver la cadena final) es la base para conectar un stream a un callback de interfaz de usuario en lugar de print.

8. Extraer Totales de Uso de message_delta

Lee el uso de tokens que solo está disponible una vez que el modelo termina de generar.

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": "Enumera tres beneficios de las 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 salida hasta ahora: {usage.output_tokens}]")

• Los eventos message_delta contienen stop_reason y usage acumulativo, campos que no se conocen hasta que la generación está terminando.
• Los recuentos de tokens de entrada llegan antes (en message_start); los recuentos de tokens de salida se finalizan en message_delta.
• Registrar el uso aquí, en lugar de solo después de get_final_message(), te permite rastrear el costo en tiempo casi real para streams de larga duración.

9. Hacer Streaming de Dos Solicitudes Concurrentemente con asyncio

Usa el cliente asíncrono para ejecutar dos streams independientes al mismo tiempo en lugar de uno tras otro.

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("Da un dato curioso de una línea sobre Marte.", "marte"),
        stream_one("Da un dato curioso de una línea sobre Venus.", "venus"),
    )
 
asyncio.run(main())

• anthropic.AsyncAnthropic() refleja la API del cliente síncrono, pero cada método de stream se espera con async with / async for.
• asyncio.gather ejecuta ambos streams concurrentemente a través de conexiones separadas, lo que es más rápido que las llamadas síncronas secuenciales cuando tienes múltiples prompts independientes.
• La salida intercalada de dos streams etiquetados es una vista previa de las preocupaciones de almacenamiento en búfer que un backend de chat multiusuario real tiene que resolver.

Relacionado: Elección entre Anthropic y AsyncAnthropic en Código de Producción - cuándo usar el cliente asíncrono.

Versiones de la Pila: Escrito contra la línea de modelos de Claude actual a partir de ~junio de 2026: Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (el predeterminado) y Claude Haiku 4.5, y el SDK oficial de Python anthropic (última versión 0.x). Los nombres de los modelos, las versiones del SDK y los precios cambian rápidamente; verifica los detalles actuales en platform.claude.com/docs antes de confiar en ellos.