Guías

Transmitir respuestas en tiempo real

Obtén respuestas en tiempo real del SDK del Agente mientras el texto y las llamadas de herramientas se transmiten

Por defecto, el SDK del Agente produce objetos AssistantMessage completos después de que Claude termina de generar cada respuesta. Para recibir actualizaciones incrementales mientras se generan texto y llamadas de herramientas, habilita la transmisión de mensajes parciales estableciendo include_partial_messages (Python) o includePartialMessages (TypeScript) en true en tus opciones.

Esta página cubre la transmisión de salida (recibir tokens en tiempo real). Para modos de entrada (cómo envías mensajes), consulta Enviar mensajes a agentes. También puedes transmitir respuestas usando el SDK del Agente a través de la CLI.

Habilitar la salida de transmisión

Para habilitar la transmisión, establece include_partial_messages (Python) o includePartialMessages (TypeScript) en true en tus opciones. Esto hace que el SDK produzca mensajes StreamEvent que contienen eventos de API sin procesar a medida que llegan, además de los AssistantMessage y ResultMessage habituales.

Tu código entonces necesita:

Verificar el tipo de cada mensaje para distinguir StreamEvent de otros tipos de mensaje
Para StreamEvent, extraer el campo event y verificar su type
Buscar eventos content_block_delta donde delta.type sea text_delta, que contienen los fragmentos de texto reales

El ejemplo a continuación habilita la transmisión e imprime fragmentos de texto a medida que llegan. Observa las verificaciones de tipo anidadas: primero para StreamEvent, luego para content_block_delta, luego para text_delta:

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio

async def stream_response():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Bash", "Read"],
    )

    async for message in query(prompt="List the files in my project", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta.get("text", ""), end="", flush=True)

asyncio.run(stream_response())

Referencia de StreamEvent

Cuando los mensajes parciales están habilitados, recibes eventos de transmisión de API de Claude sin procesar envueltos en un objeto. El tipo tiene diferentes nombres en cada SDK:

Python: StreamEvent (importar desde claude_agent_sdk.types)
TypeScript: SDKPartialAssistantMessage con type: 'stream_event'

Ambos contienen eventos de API de Claude sin procesar, no texto acumulado. Necesitas extraer y acumular deltas de texto tú mismo. Aquí está la estructura de cada tipo:

@dataclass
class StreamEvent:
    uuid: str                      # Unique identifier for this event
    session_id: str                # Session identifier
    event: dict[str, Any]          # The raw Claude API stream event
    parent_tool_use_id: str | None # Parent tool ID if from a subagent

El campo event contiene el evento de transmisión sin procesar de la API de Claude. Los tipos de eventos comunes incluyen:

Tipo de Evento	Descripción
`message_start`	Inicio de un nuevo mensaje
`content_block_start`	Inicio de un nuevo bloque de contenido (texto o uso de herramienta)
`content_block_delta`	Actualización incremental del contenido
`content_block_stop`	Fin de un bloque de contenido
`message_delta`	Actualizaciones a nivel de mensaje (razón de parada, uso)
`message_stop`	Fin del mensaje

Flujo de mensajes

Con mensajes parciales habilitados, recibes mensajes en este orden:

StreamEvent (message_start)
StreamEvent (content_block_start) - text block
StreamEvent (content_block_delta) - text chunks...
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_use block
StreamEvent (content_block_delta) - tool input chunks...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage - complete message with all content
... tool executes ...
... more streaming events for next turn ...
ResultMessage - final result

Sin mensajes parciales habilitados (include_partial_messages en Python, includePartialMessages en TypeScript), recibes todos los tipos de mensaje excepto StreamEvent. Los tipos comunes incluyen SystemMessage (inicialización de sesión), AssistantMessage (respuestas completas), ResultMessage (resultado final) y CompactBoundaryMessage (indica cuándo se compactó el historial de conversación).

Transmitir respuestas de texto

Para mostrar texto a medida que se genera, busca eventos content_block_delta donde delta.type sea text_delta. Estos contienen los fragmentos de texto incrementales. El ejemplo a continuación imprime cada fragmento a medida que llega:

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio

async def stream_text():
    options = ClaudeAgentOptions(include_partial_messages=True)

    async for message in query(prompt="Explain how databases work", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    # Print each text chunk as it arrives
                    print(delta.get("text", ""), end="", flush=True)

    print()  # Final newline

asyncio.run(stream_text())

Transmitir llamadas de herramientas

Las llamadas de herramientas también se transmiten incrementalmente. Puedes rastrear cuándo comienzan las herramientas, recibir su entrada a medida que se genera y ver cuándo se completan. El ejemplo a continuación rastrea la herramienta actual que se está llamando y acumula la entrada JSON a medida que se transmite. Utiliza tres tipos de eventos:

content_block_start: la herramienta comienza
content_block_delta con input_json_delta: llegan fragmentos de entrada
content_block_stop: llamada de herramienta completada

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio

async def stream_tool_calls():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash"],
    )

    # Track the current tool and accumulate its input JSON
    current_tool = None
    tool_input = ""

    async for message in query(prompt="Read the README.md file", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                # New tool call is starting
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    current_tool = content_block.get("name")
                    tool_input = ""
                    print(f"Starting tool: {current_tool}")

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "input_json_delta":
                    # Accumulate JSON input as it streams in
                    chunk = delta.get("partial_json", "")
                    tool_input += chunk
                    print(f"  Input chunk: {chunk}")

            elif event_type == "content_block_stop":
                # Tool call complete - show final input
                if current_tool:
                    print(f"Tool {current_tool} called with: {tool_input}")
                    current_tool = None

asyncio.run(stream_tool_calls())

Construir una interfaz de usuario de transmisión

Este ejemplo combina la transmisión de texto y herramientas en una interfaz de usuario coherente. Rastrea si el agente está ejecutando actualmente una herramienta (usando una bandera in_tool) para mostrar indicadores de estado como [Using Read...] mientras se ejecutan las herramientas. El texto se transmite normalmente cuando no está en una herramienta, y la finalización de la herramienta desencadena un mensaje "done". Este patrón es útil para interfaces de chat que necesitan mostrar progreso durante tareas de agentes de múltiples pasos.

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
from claude_agent_sdk.types import StreamEvent
import asyncio
import sys

async def streaming_ui():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash", "Grep"],
    )

    # Track whether we're currently in a tool call
    in_tool = False

    async for message in query(
        prompt="Find all TODO comments in the codebase",
        options=options
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    # Tool call is starting - show status indicator
                    tool_name = content_block.get("name")
                    print(f"\n[Using {tool_name}...]", end="", flush=True)
                    in_tool = True

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                # Only stream text when not executing a tool
                if delta.get("type") == "text_delta" and not in_tool:
                    sys.stdout.write(delta.get("text", ""))
                    sys.stdout.flush()

            elif event_type == "content_block_stop":
                if in_tool:
                    # Tool call finished
                    print(" done", flush=True)
                    in_tool = False

        elif isinstance(message, ResultMessage):
            # Agent finished all work
            print(f"\n\n--- Complete ---")

asyncio.run(streaming_ui())

Limitaciones conocidas

Algunas características del SDK son incompatibles con la transmisión:

Pensamiento extendido: cuando estableces explícitamente max_thinking_tokens (Python) o maxThinkingTokens (TypeScript), los mensajes StreamEvent no se emiten. Solo recibirás mensajes completos después de cada turno. Ten en cuenta que el pensamiento está deshabilitado por defecto en el SDK, por lo que la transmisión funciona a menos que lo habilites.
Salida estructurada: el resultado JSON aparece solo en el ResultMessage.structured_output final, no como deltas de transmisión. Consulta salidas estructuradas para obtener detalles.

Próximos pasos

Ahora que puedes transmitir texto y llamadas de herramientas en tiempo real, explora estos temas relacionados:

Consultas interactivas versus de un solo disparo: elige entre modos de entrada para tu caso de uso
Salidas estructuradas: obtén respuestas JSON tipadas del agente
Permisos: controla qué herramientas puede usar el agente

Was this page helpful?

Guías

Transmitir respuestas en tiempo real

Obtén respuestas en tiempo real del SDK del Agente mientras el texto y las llamadas de herramientas se transmiten

Habilitar la salida de transmisión

Tu código entonces necesita:

Verificar el tipo de cada mensaje para distinguir StreamEvent de otros tipos de mensaje
Para StreamEvent, extraer el campo event y verificar su type
Buscar eventos content_block_delta donde delta.type sea text_delta, que contienen los fragmentos de texto reales

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio

async def stream_response():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Bash", "Read"],
    )

    async for message in query(prompt="List the files in my project", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta.get("text", ""), end="", flush=True)

asyncio.run(stream_response())

Referencia de StreamEvent

Cuando los mensajes parciales están habilitados, recibes eventos de transmisión de API de Claude sin procesar envueltos en un objeto. El tipo tiene diferentes nombres en cada SDK:

Python: StreamEvent (importar desde claude_agent_sdk.types)
TypeScript: SDKPartialAssistantMessage con type: 'stream_event'

Ambos contienen eventos de API de Claude sin procesar, no texto acumulado. Necesitas extraer y acumular deltas de texto tú mismo. Aquí está la estructura de cada tipo:

@dataclass
class StreamEvent:
    uuid: str                      # Unique identifier for this event
    session_id: str                # Session identifier
    event: dict[str, Any]          # The raw Claude API stream event
    parent_tool_use_id: str | None # Parent tool ID if from a subagent

El campo event contiene el evento de transmisión sin procesar de la API de Claude. Los tipos de eventos comunes incluyen:

Tipo de Evento	Descripción
`message_start`	Inicio de un nuevo mensaje
`content_block_start`	Inicio de un nuevo bloque de contenido (texto o uso de herramienta)
`content_block_delta`	Actualización incremental del contenido
`content_block_stop`	Fin de un bloque de contenido
`message_delta`	Actualizaciones a nivel de mensaje (razón de parada, uso)
`message_stop`	Fin del mensaje

Flujo de mensajes

Con mensajes parciales habilitados, recibes mensajes en este orden:

StreamEvent (message_start)
StreamEvent (content_block_start) - text block
StreamEvent (content_block_delta) - text chunks...
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_use block
StreamEvent (content_block_delta) - tool input chunks...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage - complete message with all content
... tool executes ...
... more streaming events for next turn ...
ResultMessage - final result

Transmitir respuestas de texto

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio

async def stream_text():
    options = ClaudeAgentOptions(include_partial_messages=True)

    async for message in query(prompt="Explain how databases work", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    # Print each text chunk as it arrives
                    print(delta.get("text", ""), end="", flush=True)

    print()  # Final newline

asyncio.run(stream_text())

Transmitir llamadas de herramientas

content_block_start: la herramienta comienza
content_block_delta con input_json_delta: llegan fragmentos de entrada
content_block_stop: llamada de herramienta completada

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio

async def stream_tool_calls():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash"],
    )

    # Track the current tool and accumulate its input JSON
    current_tool = None
    tool_input = ""

    async for message in query(prompt="Read the README.md file", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                # New tool call is starting
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    current_tool = content_block.get("name")
                    tool_input = ""
                    print(f"Starting tool: {current_tool}")

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "input_json_delta":
                    # Accumulate JSON input as it streams in
                    chunk = delta.get("partial_json", "")
                    tool_input += chunk
                    print(f"  Input chunk: {chunk}")

            elif event_type == "content_block_stop":
                # Tool call complete - show final input
                if current_tool:
                    print(f"Tool {current_tool} called with: {tool_input}")
                    current_tool = None

asyncio.run(stream_tool_calls())

Construir una interfaz de usuario de transmisión

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
from claude_agent_sdk.types import StreamEvent
import asyncio
import sys

async def streaming_ui():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash", "Grep"],
    )

    # Track whether we're currently in a tool call
    in_tool = False

    async for message in query(
        prompt="Find all TODO comments in the codebase",
        options=options
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    # Tool call is starting - show status indicator
                    tool_name = content_block.get("name")
                    print(f"\n[Using {tool_name}...]", end="", flush=True)
                    in_tool = True

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                # Only stream text when not executing a tool
                if delta.get("type") == "text_delta" and not in_tool:
                    sys.stdout.write(delta.get("text", ""))
                    sys.stdout.flush()

            elif event_type == "content_block_stop":
                if in_tool:
                    # Tool call finished
                    print(" done", flush=True)
                    in_tool = False

        elif isinstance(message, ResultMessage):
            # Agent finished all work
            print(f"\n\n--- Complete ---")

asyncio.run(streaming_ui())

Limitaciones conocidas

Algunas características del SDK son incompatibles con la transmisión:

Pensamiento extendido: cuando estableces explícitamente max_thinking_tokens (Python) o maxThinkingTokens (TypeScript), los mensajes StreamEvent no se emiten. Solo recibirás mensajes completos después de cada turno. Ten en cuenta que el pensamiento está deshabilitado por defecto en el SDK, por lo que la transmisión funciona a menos que lo habilites.
Salida estructurada: el resultado JSON aparece solo en el ResultMessage.structured_output final, no como deltas de transmisión. Consulta salidas estructuradas para obtener detalles.

Próximos pasos

Ahora que puedes transmitir texto y llamadas de herramientas en tiempo real, explora estos temas relacionados:

Consultas interactivas versus de un solo disparo: elige entre modos de entrada para tu caso de uso
Salidas estructuradas: obtén respuestas JSON tipadas del agente
Permisos: controla qué herramientas puede usar el agente

Was this page helpful?