Respuestas en streaming con el SDK de Python

El streaming permite que tu aplicación renderice la respuesta de Claude a medida que se genera, en lugar de esperar la respuesta completa.

El paquete anthropic expone esto a través de client.messages.stream(), un gestor de contexto construido específicamente para consumir una respuesta de forma incremental.

Resumen

Una llamada no-streaming a messages.create() devuelve un objeto Message completo después de que el modelo ha terminado de generar.

Para respuestas largas, eso significa que tu aplicación permanece inactiva, sin mostrar nada, hasta que el último token está listo.

messages.stream() devuelve un gestor de contexto en su lugar, exponiendo eventos (y un conveniente iterador text_stream) a medida que llegan del servidor.

El mensaje final, completamente ensamblado, todavía está disponible una vez que el stream termina, por lo que el streaming añade una forma de observar la respuesta incrementalmente sin perder nada de lo que obtendrías de una llamada normal.

El streaming es también la forma recomendada de solicitar salidas grandes, porque las solicitudes no-streaming corren el riesgo de alcanzar tiempos de espera HTTP del lado del cliente una vez que max_tokens aumenta a decenas de miles.

Receta

import anthropic
 
client = anthropic.Anthropic()
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Escribe un poema corto sobre el océano."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
 
    final_message = stream.get_final_message()
 
print()
print(f"\nMotivo de parada: {final_message.stop_reason}")

Cuándo usar esto:

• Interfaces de chat donde quieras mostrar texto a medida que se genera, coincidiendo con la forma en que los usuarios esperan que se sienta una UI conversacional.
• Cualquier solicitud con max_tokens superior a aproximadamente 16.000, donde una llamada no-streaming corre el riesgo de un tiempo de espera del lado del cliente.
• Generación de formato largo (artículos, informes, archivos de código grandes) donde la salida parcial sigue siendo útil para el usuario antes de que se complete la respuesta completa.
• Herramientas CLI que imprimen la salida progresivamente en lugar de pausar silenciosamente.

Ejemplo funcional

import anthropic
 
client = anthropic.Anthropic()
 
 
def stream_answer(question: str) -> str:
    """Transmite una respuesta a stdout y devuelve el texto completo una vez finalizado."""
    full_text = ""
 
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=4096,
        system="Eres un 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()  # nueva línea después de la salida transmitida
    print(f"[motivo_parada={final_message.stop_reason}, "
          f"tokens_salida={final_message.usage.output_tokens}]")
 
    return full_text
 
 
if __name__ == "__main__":
    stream_answer("Explica qué es una ventana de contexto, en tres frases.")

Lo que esto demuestra:

• client.messages.stream(...) se utiliza como un gestor de contexto (with ... as stream:), que se encarga de abrir y cerrar la conexión subyacente.
• stream.text_stream produce fragmentos de texto plano a medida que llegan; no se requiere análisis manual de eventos para el caso común.
• stream.get_final_message() (llamado después de que el bucle for termina) devuelve el mismo objeto Message completo que devolvería una llamada no-streaming, incluyendo stop_reason y usage.
• Acumular fragmentos en full_text te permite imprimir progresivamente y aún así devolver la cadena completa al llamador.

Inmersión profunda

Cómo funciona

• Bajo el capó, messages.stream() abre una conexión HTTP y lee una serie de eventos enviados por el servidor (message_start, content_block_delta, message_delta, message_stop, y otros) a medida que llegan.
• text_stream es un iterador de conveniencia construido sobre esos eventos brutos: filtra los deltas de texto y produce solo el contenido de la cadena, para que no analices los tipos de eventos tú mismo para un uso básico.
• El objeto stream almacena en búfer todo lo que ha visto, lo que permite que get_final_message() devuelva un Message completo y tipado después de que el bucle termine, con todos los mismos campos (content, stop_reason, usage) que una respuesta no-streaming.
• El uso de tokens y la facturación no se ven afectados por el streaming; pagas por los mismos tokens de salida, ya sea que los recibas como un bloque o como muchos fragmentos pequeños.
• Si la conexión se interrumpe a mitad de la transmisión, la excepción surge de la iteración misma (dentro del bloque with), por lo que envuelve las llamadas de streaming en el mismo manejo de errores reintentables que usarías para cualquier otra llamada a la API.

Tipos de eventos a los que puedes acceder directamente

Para cualquier cosa que no sea texto plano, itera el objeto stream en sí en lugar de text_stream:

Notas de Python

# Inspeccionando eventos brutos en lugar de usar text_stream - útil cuando
# necesitas contenido de pensamiento o bloques tool_use a medida que se transmiten, no solo texto.
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Explica la recursividad."}],
) 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":
            # transporta actualizaciones incrementales de uso y stop_reason
            pass

Streaming así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("¿Qué es una corrutina, en dos frases?"))

El streaming de AsyncAnthropic() utiliza async with y async for en lugar de with y for; todo lo demás sobre la API es lo mismo.

Trampas

• Usar with en lugar de async with en el cliente asíncrono. AsyncAnthropic().messages.stream(...) devuelve un gestor de contexto asíncrono; abrirlo con with normal genera un TypeError. Solución: usa async with y async for siempre que estés en AsyncAnthropic().
• Olvidar flush=True al imprimir texto transmitido. Sin él, Python puede almacenar en búfer stdout y la salida parece llegar en ráfagas en lugar de suavemente. Solución: pasa flush=True a print(), o escribe directamente en sys.stdout y vacía explícitamente.
• Leer stream.text_stream dos veces. El iterador se consume una vez; iterarlo de nuevo después de que el bucle termine no produce nada. Solución: acumula el texto en una variable durante el primer pase si lo necesitas de nuevo después.
• Llamar a get_final_message() antes de que el stream se haya consumido por completo. Si sales del bucle for antes de tiempo, el mensaje final puede estar incompleto o la llamada puede bloquearse esperando el resto del stream. Solución: deja que la iteración termine de forma natural, o cierra explícitamente el stream primero si tienes la intención de abandonarlo antes.
• Asumir que el streaming evita el problema de truncamiento de max_tokens. El streaming cambia cómo recibes la salida, no cuánto de ella se le permite al modelo generar; stop_reason == "max_tokens" todavía puede ocurrir. Solución: dimensiona max_tokens para la tarea independientemente de si transmites o no.
• No manejar errores de red durante la iteración. Una conexión interrumpida se manifiesta como una excepción lanzada desde dentro del bucle for text in stream.text_stream:, no antes de que comience. Solución: envuelve el bloque with en el mismo manejo de excepciones tipificadas que usarías para llamadas no-streaming.

Alternativas

Preguntas frecuentes

Relacionado

• El modelo mental del SDK de Python de Anthropic - cómo encaja el streaming con los clientes síncronos/asíncronos
• Conceptos básicos del SDK de Python - tu primera solicitud no-streaming, para comparar
• Flujos de trabajo asíncronos de Python con uso de herramientas de streaming - streaming combinado con tool_use
• Referencia de configuración de reintentos y tiempos de espera del SDK de Python - comportamiento de tiempo de espera durante streams largos

Versiones de Stack: 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.