Tutorial de Streaming con la API de Claude 2026: Respuestas en tiempo real en Python y C#

Cuando llamas a la API de Claude sin streaming, la respuesta no vuelve hasta que el modelo termina de generar el mensaje completo. Para completions cortas eso está bien. Pero para cualquier cosa que supere unos pocos cientos de tokens — una revisión de código detallada, una explicación larga, un análisis multi-paso — los usuarios se quedan mirando una pantalla en blanco durante varios segundos y luego reciben un bloque de texto de golpe. Esa es una mala experiencia, y es completamente evitable.

El streaming resuelve esto enviando la respuesta de forma incremental, token por token, a medida que el modelo la genera. El usuario ve el texto aparecer inmediatamente, lo que hace que una respuesta de 10 segundos se sienta interactiva en lugar de congelada. Esta guía recorre exactamente cómo implementar streaming con la API de Claude en Python y C# (.NET), incluyendo manejo de eventos SSE sin procesar, un endpoint web práctico y cómo manejar errores correctamente. Todos los ejemplos usan los SDKs oficiales de Anthropic y tienen como referencia la documentación de streaming de Anthropic.

Por qué el Streaming Cambia la Experiencia del Usuario

La mejora de rendimiento percibido gracias al streaming es significativa. Según estudios de UX, los usuarios toleran esperas en blanco de alrededor de un segundo antes de empezar a sentir fricción. Una respuesta de Claude que tarda 8 segundos en completarse se siente instantánea cuando ves llegar el primer token en 200ms — aunque el tiempo total sea idéntico.

Más allá de la percepción, el streaming habilita casos de uso específicos que simplemente no funcionan con respuestas bloqueantes. Si estás construyendo una herramienta de revisión de código que procesa un diff de 2.000 líneas, quieres que el desarrollador empiece a leer los primeros hallazgos mientras el modelo sigue analizando el resto. Si estás construyendo un chatbot con contexto largo, quieres que los usuarios vean el inicio de la respuesta en lugar de esperar a que se llene el max_tokens. El streaming también es cómo funcionan la mayoría de las interfaces de chat — si alguna vez has usado Claude.ai o Cursor, el efecto de escritura que ves es SSE por debajo.

Para trabajos por lotes del lado del servidor — clasificación masiva, embeddings, procesamiento asíncrono de miles de documentos — el streaming no añade valor y sí añade complejidad. Para eso están los Message Batches. Usa streaming para funcionalidades interactivas y en tiempo real de cara al usuario.

Cómo Funciona el Streaming de la API de Claude: SSE por Dentro

Cuando estableces "stream": true en una petición a la API de Messages, la API devuelve una respuesta HTTP con Content-Type: text/event-stream y envía una secuencia de server-sent events (SSE). Cada evento tiene un nombre y un payload JSON. Según la documentación oficial de Anthropic, una respuesta de streaming completa sigue esta secuencia de eventos:

Cada content_block_delta para un bloque de texto tiene este aspecto en el formato SSE sin procesar:

event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hello"}}

Los SDKs abstraen todo esto, pero entender la secuencia de eventos importa cuando necesitas manejar llamadas a herramientas (tool use), capturar bloques de razonamiento del extended thinking, o construir routers de eventos personalizados. Los conteos de tokens en message_delta son acumulativos — el output_tokens final en ese evento es contra lo que se factura, según la documentación de Anthropic.

Python: Streaming con el SDK de Anthropic

Instala el SDK de Python de Anthropic con pip install anthropic. El patrón de streaming más simple usa el context manager y la propiedad text_stream, que solo entrega los deltas de texto y maneja todo lo demás internamente:

import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from env

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Review this Python function for edge cases:\n\ndef divide(a, b):\n    return a / b"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()  # newline at end

El iterador stream.text_stream automáticamente omite los eventos message_start, content_block_start, ping y todos los eventos contables, entregando únicamente los fragmentos de texto. Es la opción correcta cuando simplemente quieres imprimir o canalizar la respuesta.

Cuando necesitas el objeto Message completo al final — para registrar el uso, verificar el stop reason, o extraer múltiples bloques de contenido de una respuesta con tool use — usa get_final_message():

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=2048,
    system="You are a code reviewer. Be specific and concise.",
    messages=[{"role": "user", "content": "Review the attached diff for security issues:\n\n{diff}"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    final = stream.get_final_message()
    print(f"\n\n--- Usage: {final.usage.input_tokens} in / {final.usage.output_tokens} out ---")

Para un control más fino — manejar llamadas a herramientas, capturar bloques de razonamiento o construir routers de eventos personalizados — itera sobre el stream sin procesar:

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "What is 7 * 8?"}]
) as stream:
    for event in stream:
        match event.type:
            case "content_block_delta":
                if event.delta.type == "text_delta":
                    print(event.delta.text, end="", flush=True)
            case "message_delta":
                print(f"\nStop reason: {event.delta.stop_reason}")
                print(f"Output tokens: {event.usage.output_tokens}")
            case "error":
                print(f"Stream error: {event.error}")
                break

El Python asíncrono funciona de forma idéntica pero con AsyncAnthropic y async for. Este es el patrón que usarás dentro de un endpoint de FastAPI.

Si estás construyendo aplicaciones LLM serias, el libro AI Engineering by Chip Huyen (~$55) tiene un capítulo exhaustivo sobre construcción de pipelines de inferencia con streaming y patrones de servicio LLM en producción — vale la pena leerlo antes de escalar.

C# y .NET: Streaming con el SDK Oficial de Anthropic

Anthropic publica un SDK oficial de C#. A partir de la versión 10+, el paquete en NuGet es Anthropic (el paquete comunitario anterior tryAGI se ha movido a tryAGI.Anthropic). Instálalo con:

dotnet add package Anthropic

El SDK requiere .NET Standard 2.0 o posterior, por lo que funciona con .NET 6, 8 o 9. Para llamadas sin streaming, usas client.Messages.Create(). Para streaming, cada método de streaming tiene el sufijo Streaming — client.Messages.CreateStreaming() — y devuelve IAsyncEnumerable:

using Anthropic;
using Anthropic.Models.Messages;

AnthropicClient client = new(); // reads ANTHROPIC_API_KEY from env

MessageCreateParams parameters = new()
{
    MaxTokens = 1024,
    Messages = [new UserMessageParam { Content = "What are the main differences between async/await in C# and Python?" }],
    Model = Model.ClaudeOpus5_5,
};

await foreach (var @event in client.Messages.CreateStreaming(parameters))
{
    Console.Write(@event); // prints each SSE event as it arrives
}
Console.WriteLine();

Para extraer texto, filtra a los eventos content_block_delta y verifica el tipo de delta. Según la documentación oficial del SDK, el manejo de errores en streaming usa tipos de excepción específicos. Los errores de rate limit y sobrecarga durante el streaming se manifiestan como AnthropicSseException en lugar del AnthropicRateLimitException estándar (que aplica a errores HTTP sin streaming):

using Anthropic;
using Anthropic.Models.Messages;

AnthropicClient client = new() { MaxRetries = 2 };

MessageCreateParams parameters = new()
{
    MaxTokens = 2048,
    System = "You are a helpful .NET developer assistant.",
    Messages = [new UserMessageParam { Content = "Explain the difference between Task and ValueTask in C#" }],
    Model = Model.ClaudeOpus5_5,
};

try
{
    await foreach (var @event in client.Messages.CreateStreaming(parameters))
    {
        // Print each SSE event chunk as it arrives
        Console.Write(@event);
        await Console.Out.FlushAsync();
    }
}
catch (AnthropicRateLimitException ex)
{
    Console.Error.WriteLine($"Rate limited: retry after {ex.RetryAfter?.TotalSeconds}s");
    throw;
}
catch (AnthropicSseException ex)
{
    Console.Error.WriteLine($"SSE stream error: {ex.Message}");
    throw;
}
catch (AnthropicIOException ex)
{
    Console.Error.WriteLine($"Network error during stream: {ex.Message}");
    throw;
}

El SDK reintenta automáticamente 2 veces por defecto en errores 429 y 5xx. Para streaming, configura el MaxRetries a nivel de cliente como se muestra — los reintentos aplican a la petición inicial, no a la reconexión a mitad del stream.

Si dedicas tiempo considerable a escribir C# y quieres una buena referencia para patrones de integración LLM, Hands-On Large Language Models (~$55) cubre la construcción de aplicaciones reales en producción de extremo a extremo, con ejemplos que se adaptan bien al pensamiento de arquitectura estilo .NET.

Construyendo un Endpoint de Chat con Streaming en ASP.NET Core

El caso de uso más común para streaming en un backend .NET es un minimal API o un endpoint de controlador que hace proxy del stream SSE de Claude hacia un frontend. La clave está en establecer Content-Type: text/event-stream y deshabilitar el buffering de respuesta para que los fragmentos lleguen al cliente a medida que arriban:

using Anthropic;
using Anthropic.Models.Messages;
using Microsoft.AspNetCore.Mvc;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddSingleton<AnthropicClient>();

var app = builder.Build();

app.MapPost("/api/chat/stream", async (
    [FromBody] ChatRequest request,
    AnthropicClient anthropic,
    HttpContext ctx) =>
{
    ctx.Response.ContentType = "text/event-stream";
    ctx.Response.Headers["Cache-Control"] = "no-cache";
    ctx.Response.Headers["Connection"] = "keep-alive";

    var parameters = new MessageCreateParams
    {
        MaxTokens = 1024,
        Messages = [new UserMessageParam { Content = request.Message }],
        Model = Model.ClaudeOpus5_5,
    };

    await foreach (var @event in anthropic.Messages.CreateStreaming(parameters))
    {
        var data = $"data: {System.Text.Json.JsonSerializer.Serialize(@event)}\n\n";
        await ctx.Response.WriteAsync(data);
        await ctx.Response.Body.FlushAsync();
    }

    await ctx.Response.WriteAsync("data: [DONE]\n\n");
    await ctx.Response.Body.FlushAsync();
});

app.Run();

record ChatRequest(string Message);

En el frontend Angular, consúmelo con la API nativa EventSource o con la Fetch API usando un readable stream reader — ambas funcionan. EventSource te da lógica de reconexión automática; el enfoque de Fetch stream te da más control sobre los headers (útil para autenticación). Para una arquitectura Angular 17+ basada en signals, el enfoque de Fetch stream se integra más limpiamente con signals y toSignal().

Un gotcha específico de .NET: el middleware de buffering de respuesta (como UseResponseBuffering de algunos paquetes antiguos) se tragará tus eventos SSE. Asegúrate de no estar apilando ningún middleware de buffering entre el endpoint y el cliente. Kestrel e IIS Express manejan el chunked transfer encoding correctamente sin configuración adicional.

Manejo de Errores y Reconexión

La API de Claude puede enviar eventos de error a mitad del stream, más comúnmente overloaded_error durante períodos de alta demanda. Según la documentación de Anthropic, esto es equivalente a un HTTP 529 en un contexto sin streaming. Tu manejador del stream necesita detectar estos errores y exponerlos apropiadamente, en lugar de dejar al cliente esperando:

import anthropic
import time

client = anthropic.Anthropic()

def stream_with_retry(messages: list, max_retries: int = 2):
    for attempt in range(max_retries + 1):
        try:
            with client.messages.stream(
                model="claude-opus-5-5",
                max_tokens=1024,
                messages=messages,
            ) as stream:
                for event in stream:
                    if event.type == "error":
                        error_type = event.error.get("type", "unknown")
                        if error_type == "overloaded_error" and attempt < max_retries:
                            print(f"\nOverloaded, retrying in 5s (attempt {attempt + 1})...")
                            time.sleep(5 * (attempt + 1))
                            break  # exits the for loop, retries the outer while
                        else:
                            raise RuntimeError(f"Unrecoverable stream error: {error_type}")
                    elif event.type == "content_block_delta" and hasattr(event.delta, "text"):
                        yield event.delta.text
                else:
                    return  # stream completed cleanly
        except anthropic.RateLimitError:
            if attempt < max_retries:
                time.sleep(10 * (attempt + 1))
            else:
                raise

El prompt caching y el streaming funcionan juntos — si tienes un system prompt grande que califica para el prompt caching (en inglés), puedes combinar los bloques cache_control con stream=True. El acierto o fallo de caché se refleja en el objeto usage del evento message_start a través de cache_read_input_tokens y cache_creation_input_tokens, según la documentación de la API.

Cuándo NO Usar Streaming

El streaming añade complejidad — gestión de conexiones SSE, preocupaciones de buffering, estados de error de respuesta parcial y reensamblaje del lado del cliente. Aquí está cuándo usar peticiones bloqueantes estándar en su lugar:

Usa llamadas bloqueantes (sin streaming) cuando estás ejecutando trabajos por lotes del lado del servidor donde el resultado va a una base de datos o cola en lugar de a un usuario. Usa Message Batches (endpoint de API completamente separado) cuando estás procesando cientos de prompts a la vez y puedes esperar hasta 24 horas — obtienes un 50% de descuento en el precio de los tokens a cambio. Usa llamadas bloqueantes cuando necesitas una función síncrona simple que transforma texto y no quieres la complejidad de SSE en tu base de código. Usa streaming para cualquier UI de chat o generación de cara al usuario, cualquier escenario donde el tiempo de respuesta sea notable, y cualquier caso de uso donde el usuario se beneficia de ver la salida parcial.

En cuanto al hardware para ejecutar herramientas potenciadas por Claude localmente durante el desarrollo, una máquina con suficiente RAM para correr tu stack de desarrollo .NET/Python junto con otros servicios importa. Yo ejecuto mi entorno de desarrollo en el Beelink GTi13 Ultra (host Proxmox). Si estás construyendo una máquina de desarrollo dedicada, un Beelink SER5 Mini PC (~$299) es una máquina sólida y de bajo consumo para ejecutar proxies locales de la API de Claude y herramientas de desarrollo. Combínalo con un Crucial 32GB DDR4 SODIMM (~$59) si estás ejecutando Ollama o múltiples servicios de desarrollo junto a él.

Conclusión

El streaming con la API de Claude es directo una vez que entiendes el modelo de eventos SSE. El iterador text_stream del SDK de Python maneja el 90% de los casos de uso con código mínimo. Para C#, el paquete oficial Anthropic de NuGet expone el streaming a través de IAsyncEnumerable mediante CreateStreaming(), que se conecta directamente al patrón minimal API de ASP.NET Core. Las cosas principales que hay que hacer bien son el manejo de errores para los eventos overloaded_error a mitad del stream, deshabilitar el buffering de respuesta en tu web API, y no recurrir al streaming cuando el procesamiento por lotes o las llamadas síncronas simples son la mejor herramienta.

Si eres nuevo en la API de Claude y quieres entender la amplitud completa de funcionalidades antes de construir, el tutorial de tool use (en inglés) cubre las llamadas a funciones — el otro primitive clave de la API que funciona junto al streaming para flujos de trabajo agénticos.

Una nota final sobre la selección de modelo para streaming: la documentación de Anthropic actualmente muestra claude-opus-5-5 como el modelo de referencia para todos los ejemplos de streaming. Para aplicaciones de streaming de alto volumen y sensibles a la latencia, consulta la página de modelos de Anthropic para el modelo recomendado y los precios más recientes — los modelos de nivel inferior de la familia Claude ofrecen menor tiempo hasta el primer token a un costo reducido, lo que importa mucho en la UX de streaming.