Passer au contenu principal
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 :
  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())

​
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énementDescription
message_startDébut d’un nouveau message
content_block_startDébut d’un nouveau bloc de contenu (texte ou utilisation d’outil)
content_block_deltaMise à jour incrémentielles du contenu
content_block_stopFin d’un bloc de contenu
message_deltaMises à jour au niveau du message (raison d’arrêt, utilisation)
message_stopFin 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 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 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 affiche chaque fragment à mesure qu’il arrive : 
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())

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

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

​
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 :