Diffuser les réponses en temps réel

Par défaut, le SDK Agent produit des objets AssistantMessage complets après que Claude ait terminé de générer chaque réponse. 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 en définissant include_partial_messages (Python) ou includePartialMessages (TypeScript) sur true dans vos options.

Cette page couvre la diffusion de sortie (réception des jetons en temps réel). Pour les modes d'entrée (comment vous envoyez les messages), consultez Envoyer des messages aux agents. Vous pouvez également diffuser les réponses en utilisant le SDK Agent via la CLI.

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 :

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

L'exemple ci-dessous active la diffusion et imprime 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 :

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. Voici la structure de chaque type :

@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

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

Type d'événement	Description
`message_start`	Début d'un nouveau message
`content_block_start`	Début d'un nouveau bloc de contenu (texte ou utilisation d'outil)
`content_block_delta`	Mise à jour incrémentielles du contenu
`content_block_stop`	Fin d'un bloc de contenu
`message_delta`	Mises à jour au niveau du message (raison d'arrêt, utilisation)
`message_stop`	Fin du message

Flux de messages

Avec les messages partiels activés, vous recevez les messages dans cet ordre :

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

Sans les messages partiels activés (include_partial_messages en Python, includePartialMessages en TypeScript), vous recevez tous les types de messages sauf StreamEvent. Les types courants incluent SystemMessage (initialisation de session), AssistantMessage (réponses complètes), ResultMessage (résultat final) et CompactBoundaryMessage (indique quand l'historique de conversation a été compacté).

Diffuser les réponses texte

Pour afficher le texte à mesure qu'il est généré, recherchez les événements content_block_delta où delta.type est text_delta. Ceux-ci contiennent les fragments de texte incrémentiels. L'exemple ci-dessous imprime chaque fragment à mesure qu'il arrive :

Diffuser les appels d'outils

Les appels d'outils se diffusent également 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 se diffuse. 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

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'est pas dans un 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.

Limitations connues

Certaines fonctionnalités du SDK sont incompatibles avec la diffusion :

Réflexion étendue : quand vous définissez explicitement max_thinking_tokens (Python) ou maxThinkingTokens (TypeScript), les messages StreamEvent ne sont pas émis. Vous recevrez uniquement les messages complets après chaque tour. Notez que la réflexion est désactivée par défaut dans le SDK, donc la diffusion fonctionne sauf si vous l'activez.
Sortie structurée : le résultat JSON n'apparaît que dans le ResultMessage.structured_output final, pas comme des deltas de diffusion. Consultez sorties structurées pour plus de détails.

Prochaines étapes

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

Requêtes interactives vs 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

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())

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())

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())

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())