Blog de Agentes IA
SISTEMA DE ACCESO INFORMÁTICO

Streaming de Respuestas de Agentes: Salida en Tiempo Real para Flujos de Trabajo Multi-Paso

streaming agent responsesreal-time agentsprogressive agent outputagent UXstreaming tool callsClaude APIserver-sent events

Streaming de Respuestas de Agentes: Salida en Tiempo Real para Flujos de Trabajo Multi-Paso

Tu agente tarda 20 segundos en investigar una pregunta, llamar a tres herramientas y sintetizar una respuesta. El usuario hace clic en “Preguntar” y se queda mirando un spinner durante 20 segundos. No está seguro de si está funcionando. Considera actualizar la página. Se pregunta si debería empezar de nuevo.

Ahora imagina esto: el usuario hace clic en “Preguntar” y en menos de 500 milisegundos ve aparecer las primeras palabras del agente. Lo observa buscar información —”🔍 Buscando en la base de datos de pedidos…”— y ve cómo llegan los resultados en tiempo real. Lee la respuesta mientras se está escribiendo, frase a frase. Los mismos 20 segundos. Una experiencia completamente diferente.

El streaming no es un añadido opcional para los agentes orientados al usuario —es un requisito de UX. El tiempo hasta el primer token es la métrica de latencia más importante en las interfaces de agentes. Los usuarios que ven progreso son pacientes. Los usuarios que no ven nada abandonan. Estudio tras estudio sobre rendimiento web demuestra que la latencia percibida importa más que la latencia real, y el streaming es la herramienta más poderosa que tienes para cerrar la brecha entre ambas.

En este artículo aprenderás cómo implementar streaming en tiempo real para agentes de IA multi-paso —desde la entrega token a token hasta la transparencia en las llamadas a herramientas, actualizaciones de estado, divulgación progresiva, recuperación de errores y elecciones de capa de transporte.

Sección 1: Fundamentos de la API de Streaming de Claude

Antes de poder hacer streaming de un agente, necesitas hacer streaming de una sola respuesta de Claude. Empecemos por los fundamentos.

Streaming vs. Sin Streaming

Una llamada a la API sin streaming se bloquea hasta que se genera la respuesta completa y luego la devuelve toda de una vez. Una llamada con streaming devuelve una secuencia de eventos a medida que se produce la respuesta, comenzando con el primer token.

La diferencia en el código es mínima. La diferencia en la experiencia del usuario es enorme.

Tipos de Eventos

La API de streaming de Claude emite una secuencia estructurada de eventos server-sent:

  1. message_start — Contiene el objeto Message inicial con metadatos (modelo, rol, uso).
  2. content_block_start — Señala el inicio de un bloque de contenido (texto o tool_use).
  3. content_block_delta — Contiene contenido incremental: fragmentos de texto o JSON parcial de entrada de herramienta.
  4. content_block_stop — Señala el fin de un bloque de contenido.
  5. message_delta — Actualizaciones finales del mensaje (razón de parada, uso final).
  6. message_stop — El stream ha terminado.

Para respuestas de texto, recibirás muchos eventos content_block_delta, cada uno con un pequeño fragmento de texto (normalmente unos pocos tokens).

Implementación Básica de Streaming

Aquí tienes un ejemplo de streaming síncrono usando el SDK de Python de Anthropic:

import anthropic


client = anthropic.Anthropic()


def stream_basic_response(user_message: str):
    """Stream a basic Claude response token by token."""
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()  # Newline after stream completes


stream_basic_response("Explain quantum entanglement in simple terms.")

Y la versión asíncrona, que es la que usarás en servidores web en producción:

import anthropic
import asyncio


async_client = anthropic.AsyncAnthropic()


async def stream_basic_response_async(user_message: str):
    """Async streaming with the Anthropic SDK."""
    async with async_client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
    print()


asyncio.run(stream_basic_response_async("Explain quantum entanglement in simple terms."))

Manejo de Tokens Parciales y Límites UTF-8

El SDK gestiona la decodificación UTF-8 por ti, pero si trabajas con el stream HTTP crudo, ten en cuenta que los caracteres multibyte pueden dividirse entre fragmentos. Siempre almacena en búfer los bytes crudos y decodifica solo cuando tengas secuencias UTF-8 completas. El iterador text_stream del SDK de anthropic gestiona esto automáticamente —otra razón para usarlo en lugar de parsear tú mismo el stream SSE crudo.

Sección 2: Streaming de Llamadas a Herramientas

El streaming de texto básico es el mínimo indispensable. El verdadero reto —y el verdadero valor— surge cuando tu agente utiliza herramientas. Un agente multi-paso que llama a tres herramientas en secuencia puede parecer muerto durante la ejecución de las herramientas a menos que muestres lo que está pasando.

Cómo Aparecen las Llamadas a Herramientas en el Stream

Cuando Claude decide usar una herramienta, el stream emite un evento content_block_start con type: "tool_use", seguido de eventos content_block_delta que contienen fragmentos del JSON de entrada de la herramienta. Una vez ensamblada la llamada completa, ejecutas la herramienta, inyectas el resultado y continúas la conversación.

La idea clave: conoces el nombre de la herramienta en cuanto llega content_block_start, incluso antes de que el JSON de entrada esté completo. Esto significa que puedes mostrar al usuario inmediatamente algo como ”🔍 Buscando en la base de datos de pedidos…” sin esperar a la llamada completa.

El Bucle Completo del Agente con Streaming

Aquí tienes un bucle completo de agente con streaming que muestra las invocaciones de herramientas en tiempo real:

import anthropic
import json


client = anthropic.Anthropic()


# Define tools
tools = [
    {
        "name": "search_orders",
        "description": "Search customer orders by order ID or customer email.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Order ID or email"},
            },
            "required": ["query"],
        },
    },
    {
        "name": "get_shipping_status",
        "description": "Get real-time shipping status for an order.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "The order ID"},
            },
            "required": ["order_id"],
        },
    },
]




def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool and return the result as a string."""
    if tool_name == "search_orders":
        # Simulated database lookup
        return json.dumps({
            "order_id": "ORD-12345",
            "customer": "jane@example.com",
            "items": ["Blue Widget x2", "Red Gadget x1"],
            "total": "$47.99",
            "status": "shipped",
        })
    elif tool_name == "get_shipping_status":
        return json.dumps({
            "order_id": tool_input["order_id"],
            "carrier": "FedEx",
            "tracking": "7891011",
            "estimated_delivery": "2026-03-10",
            "current_location": "Memphis, TN",
        })
    return json.dumps({"error": f"Unknown tool: {tool_name}"})




def stream_agent_response(user_message: str):
    """
    Complete streaming agent loop with real-time tool call display.
    """
    messages = [{"role": "user", "content": user_message}]


    while True:
        # Stream the model response
        tool_calls = []
        current_tool = None


        with client.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        ) as stream:
            response = None
            for event in stream:
                # The SDK exposes raw events via iteration
                pass


            # Use the helper to collect text and tool use
            response = stream.get_final_message()


        # Process the response content blocks
        for block in response.content:
            if block.type == "text":
                print(block.text, end="", flush=True)
            elif block.type == "tool_use":
                tool_name = block.name
                tool_input = block.input
                tool_id = block.id


                # Show the user what's happening
                print(f"\n⚙️  Calling tool: {tool_name}({json.dumps(tool_input)})")


                # Execute the tool
                result = execute_tool(tool_name, tool_input)
                print(f"✅  Result received from {tool_name}")


                tool_calls.append({
                    "tool_use_id": tool_id,
                    "tool_name": tool_name,
                    "tool_input": tool_input,
                    "result": result,
                })


        # If the model stopped because it wants to use tools, continue the loop
        if response.stop_reason == "tool_use":
            # Add assistant message with all content blocks
            messages.append({"role": "assistant", "content": response.content})


            # Add tool results
            tool_results = []
            for tc in tool_calls:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": tc["tool_use_id"],
                    "content": tc["result"],
                })
            messages.append({"role": "user", "content": tool_results})


            print("\n--- Continuing agent loop ---")
        else:
            # Model is done (stop_reason == "end_turn")
            print()
            break




# Run the agent
stream_agent_response("Where is my order #ORD-12345? When will it arrive?")

Para un streaming token a token verdadero con detección de llamadas a herramientas, puedes iterar sobre los eventos crudos:

def stream_agent_with_live_tokens(user_message: str):
    """Stream text tokens live while also detecting tool calls."""
    messages = [{"role": "user", "content": user_message}]


    while True:
        collected_content = []
        current_text = ""
        current_tool_name = None
        current_tool_input_json = ""
        current_tool_id = None
        stop_reason = None


        with client.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        ) as stream:
            for event in stream:
                if hasattr(event, 'type'):
                    if event.type == 'content_block_start':
                        if event.content_block.type == 'tool_use':
                            current_tool_name = event.content_block.


---


## Artículos Relacionados


- [Patrones de Uso de Herramientas: Interfaces Agente-Herramienta Confiables](/es/blog/agent-tool-use-patterns/)
- [Patrones Multi-Agente: Orquestadores, Workers y Pipelines](/es/blog/multi-agent-patterns/)
- [Recuperación de Errores en Agentes: 5 Patrones para Fiabilidad en Producción](/es/blog/agent-error-recovery-patterns/)
- [Agentes de Automatización Web: Control del Navegador con Claude y Computer Use](/es/blog/web-automation-agents-browser-control-with-claude-and-computer-use/)