Transmitir respuestas en tiempo real
Obtener respuestas en tiempo real del Agent SDK mientras el texto y las llamadas de herramientas se transmiten
De forma predeterminada, el Agent SDK produce un AssistantMessage completo para cada bloque de contenido no vacío, como un bloque de texto o una llamada de herramienta, después de que Claude termina de generar ese bloque. Para recibir actualizaciones incrementales mientras se generan texto y llamadas de herramientas, habilite la transmisión de mensajes parciales.
Habilitar la transmisión de salida
Para habilitar la transmisión, establezca include_partial_messages (Python) o includePartialMessages (TypeScript) en true en sus 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.
Su código entonces necesita:
- Verificar el tipo de cada mensaje para distinguir
StreamEventde otros tipos de mensaje - Para
StreamEvent, extraer el campoeventy verificar sutype - Buscar eventos
content_block_deltadondedelta.typeseatext_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. Observe 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())
import { query } from "@anthropic-ai/claude-agent-sdk";
for await (const message of query({
prompt: "List the files in my project",
options: {
includePartialMessages: true,
allowedTools: ["Bash", "Read"]
}
})) {
if (message.type === "stream_event") {
const event = message.event;
if (event.type === "content_block_delta") {
if (event.delta.type === "text_delta") {
process.stdout.write(event.delta.text);
}
}
}
}
Referencia de StreamEvent
Cuando los mensajes parciales están habilitados, recibe eventos de transmisión sin procesar de la API de Claude envueltos en un objeto. El tipo tiene nombres diferentes en cada SDK:
- Python:
StreamEvent(importar desdeclaude_agent_sdk.types) - TypeScript:
SDKPartialAssistantMessagecontype: 'stream_event'
Ambos contienen eventos sin procesar de la API de Claude, no texto acumulado. Debe extraer y acumular los deltas de texto usted 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 # Always None
type SDKPartialAssistantMessage = {
type: "stream_event";
event: BetaRawMessageStreamEvent; // From Anthropic SDK
parent_tool_use_id: string | null;
uuid: UUID;
session_id: string;
ttft_ms?: number; // Time to first token in ms, present only on message_start events
user_message_uuid?: string;
};
El campo parent_tool_use_id siempre es None en Python y null en TypeScript. Los eventos de transmisión se emiten solo para la sesión principal; los deltas a nivel de token de los subagentes no se reenvían. Para atribuir la salida a un subagente, use mensajes completos, que llevan parent_tool_use_id. Consulte Detectar invocación de subagente.
Claude Code establece user_message_uuid en el primer evento de transmisión que no es ping del turno, y nuevamente cuando cambia el mensaje al que el turno está respondiendo, bajo las condiciones en user_message_uuid. El StreamEvent de Python no expone este campo.
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
Claude Code emite un AssistantMessage a medida que cada bloque de contenido no vacío se completa, por lo que una respuesta con un bloque de texto y una llamada de herramienta produce dos objetos AssistantMessage. Cada uno lleva solo su propio bloque de contenido, y ambos comparten el mismo ID de mensaje, que usted lee como message.message.id en TypeScript y message.message_id en Python. Con mensajes parciales habilitados, cada AssistantMessage llega antes del evento content_block_stop de ese bloque, y usted recibe mensajes en este orden:
StreamEvent (message_start)
StreamEvent (content_block_start) - text block
StreamEvent (content_block_delta) - text chunks...
AssistantMessage - complete text block
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_use block
StreamEvent (content_block_delta) - tool input chunks...
AssistantMessage - complete tool_use block
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
... tool executes ...
... more streaming events for next turn ...
ResultMessage - final result
Sin mensajes parciales habilitados, recibe todos los tipos de mensajes excepto StreamEvent. Los tipos comunes incluyen SystemMessage (inicialización de sesión), AssistantMessage (bloques de contenido completos), ResultMessage (resultado final), y un mensaje de límite compacto que indica cuándo se compactó el historial de conversación (SDKCompactBoundaryMessage en TypeScript; SystemMessage con subtipo "compact_boundary" en Python).
Transmitir llamadas de herramientas
Las llamadas de herramientas también se transmiten incrementalmente. Puede 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 comienzacontent_block_deltaconinput_json_delta: llegan fragmentos de entradacontent_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())
import { query } from "@anthropic-ai/claude-agent-sdk";
// Track the current tool and accumulate its input JSON
let currentTool: string | null = null;
let toolInput = "";
for await (const message of query({
prompt: "Read the README.md file",
options: {
includePartialMessages: true,
allowedTools: ["Read", "Bash"]
}
})) {
if (message.type === "stream_event") {
const event = message.event;
if (event.type === "content_block_start") {
// New tool call is starting
if (event.content_block.type === "tool_use") {
currentTool = event.content_block.name;
toolInput = "";
console.log(`Starting tool: ${currentTool}`);
}
} else if (event.type === "content_block_delta") {
if (event.delta.type === "input_json_delta") {
// Accumulate JSON input as it streams in
const chunk = event.delta.partial_json;
toolInput += chunk;
console.log(` Input chunk: ${chunk}`);
}
} else if (event.type === "content_block_stop") {
// Tool call complete - show final input
if (currentTool) {
console.log(`Tool ${currentTool} called with: ${toolInput}`);
currentTool = null;
}
}
}
}
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 agente de varios 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())
import { query } from "@anthropic-ai/claude-agent-sdk";
// Track whether we're currently in a tool call
let inTool = false;
for await (const message of query({
prompt: "Find all TODO comments in the codebase",
options: {
includePartialMessages: true,
allowedTools: ["Read", "Bash", "Grep"]
}
})) {
if (message.type === "stream_event") {
const event = message.event;
if (event.type === "content_block_start") {
if (event.content_block.type === "tool_use") {
// Tool call is starting - show status indicator
process.stdout.write(`\n[Using ${event.content_block.name}...]`);
inTool = true;
}
} else if (event.type === "content_block_delta") {
// Only stream text when not executing a tool
if (event.delta.type === "text_delta" && !inTool) {
process.stdout.write(event.delta.text);
}
} else if (event.type === "content_block_stop") {
if (inTool) {
// Tool call finished
console.log(" done");
inTool = false;
}
}
} else if (message.type === "result") {
// Agent finished all work
console.log("\n\n--- Complete ---");
}
}
Limitaciones conocidas
- Structured output: el resultado JSON aparece solo en el
ResultMessage.structured_outputfinal, no como deltas de transmisión. Consulte structured outputs para obtener detalles.
Próximos pasos
Ahora que puede transmitir texto y llamadas de herramientas en tiempo real, explore estos temas relacionados:
- Interactive vs one-shot queries: elija entre modos de entrada para su caso de uso
- Structured outputs: obtenga respuestas JSON tipificadas del agente
- Permissions: controle qué herramientas puede usar el agente