Diffuser les réponses en temps réel

Recevez les réponses en temps réel du SDK Agent à mesure que le texte et les appels d'outils sont diffusés

Par défaut, le SDK Agent produit un objet AssistantMessage complet pour chaque bloc de contenu non vide, tel qu'un bloc de texte ou un appel d'outil, après que Claude ait terminé de générer ce bloc. Pour recevoir des mises à jour incrémentielles à mesure que le texte et les appels d'outils sont générés, activez la diffusion de messages partiels.

Activer la diffusion de sortie

Pour activer la diffusion, définissez include_partial_messages (Python) ou includePartialMessages (TypeScript) sur true dans vos options. Cela fait que le SDK produit des messages StreamEvent contenant les événements API bruts à mesure qu'ils arrivent, en plus des AssistantMessage et ResultMessage habituels.

Votre code doit alors :

1. Vérifier le type de chaque message pour distinguer StreamEvent des autres types de messages
2. Pour StreamEvent, extraire le champ event et vérifier son type
3. Rechercher les événements content_block_delta où delta.type est text_delta, qui contiennent les fragments de texte réels

L'exemple ci-dessous active la diffusion et affiche les fragments de texte à mesure qu'ils arrivent. Remarquez les vérifications de type imbriquées : d'abord pour StreamEvent, puis pour content_block_delta, puis pour 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);
}
}
}
}

Référence StreamEvent

Lorsque les messages partiels sont activés, vous recevez les événements de diffusion bruts de l'API Claude enveloppés dans un objet. Le type a des noms différents dans chaque SDK :

• Python : StreamEvent (importer depuis claude_agent_sdk.types)
• TypeScript : SDKPartialAssistantMessage avec type: 'stream_event'

Les deux contiennent les événements bruts de l'API Claude, pas le texte accumulé. Vous devez extraire et accumuler les deltas de texte vous-même.

Le champ parent_tool_use_id est toujours None en Python et null en TypeScript. Les événements de diffusion sont émis pour la session principale uniquement ; les deltas au niveau des tokens des sous-agents ne sont pas transmis. Pour attribuer la sortie à un sous-agent, utilisez les messages complets, qui portent parent_tool_use_id. Voir Détecter l'invocation de sous-agent.

Claude Code définit user_message_uuid sur le premier événement de diffusion non-ping du tour, et à nouveau lorsque le message auquel le tour répond change, selon les conditions dans user_message_uuid. Le StreamEvent Python n'expose pas ce champ.

Le champ event contient l'événement de diffusion brut de l'API Claude. Les types d'événements courants incluent :

Flux des messages

Claude Code émet un AssistantMessage à chaque fois qu'un bloc de contenu non vide se termine, donc une réponse avec un bloc de texte et un appel d'outil produit deux objets AssistantMessage. Chacun ne porte que son propre bloc de contenu, et les deux partagent le même ID de message, que vous lisez comme message.message.id en TypeScript et message.message_id en Python. Avec les messages partiels activés, chaque AssistantMessage arrive avant l'événement content_block_stop de ce bloc, et vous recevez les messages dans cet ordre :

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

Sans les messages partiels activés, vous recevez tous les types de messages sauf StreamEvent. Les types courants incluent SystemMessage (initialisation de session), AssistantMessage (blocs de contenu complets), ResultMessage (résultat final), et un message de limite compact indiquant quand l'historique de conversation a été compacté (SDKCompactBoundaryMessage en TypeScript ; SystemMessage avec le sous-type "compact_boundary" en Python).

Diffuser les appels d'outils

Les appels d'outils sont également diffusés de manière incrémentielles. Vous pouvez suivre quand les outils commencent, recevoir leur entrée à mesure qu'elle est générée, et voir quand ils se terminent. L'exemple ci-dessous suit l'outil actuellement appelé et accumule l'entrée JSON à mesure qu'elle est diffusée. Il utilise trois types d'événements :

• content_block_start : l'outil commence
• content_block_delta avec input_json_delta : les fragments d'entrée arrivent
• content_block_stop : l'appel d'outil est complet

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;
}
}
}
}

Construire une interface utilisateur de diffusion

Cet exemple combine la diffusion de texte et d'outils dans une interface utilisateur cohésive. Il suit si l'agent exécute actuellement un outil (en utilisant un drapeau in_tool) pour afficher des indicateurs de statut comme [Using Read...] pendant que les outils s'exécutent. Le texte se diffuse normalement quand il n'y a pas d'outil, et la fin de l'outil déclenche un message « done ». Ce modèle est utile pour les interfaces de chat qui doivent afficher la progression pendant les tâches d'agent multi-étapes.

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 ---");
}
}

Limitations connues

• Sortie structurée : le résultat JSON n'apparaît que dans le ResultMessage.structured_output final, pas comme des deltas de diffusion. Consultez les sorties structurées pour plus de détails.

Étapes suivantes

Maintenant que vous pouvez diffuser le texte et les appels d'outils en temps réel, explorez ces sujets connexes :

• Requêtes interactives ou ponctuelles : choisissez entre les modes d'entrée pour votre cas d'usage
• Sorties structurées : obtenez des réponses JSON typées de l'agent
• Permissions : contrôlez les outils que l'agent peut utiliser