Zum Hauptinhalt springen
Standardmäßig liefert das Agent SDK vollständige AssistantMessage-Objekte, nachdem Claude die Generierung jeder Antwort abgeschlossen hat. Um inkrementelle Aktualisierungen zu erhalten, während Text und Tool-Aufrufe generiert werden, aktivieren Sie das Streaming von Teillmeldungen, indem Sie include_partial_messages (Python) oder includePartialMessages (TypeScript) in Ihren Optionen auf true setzen.
Diese Seite behandelt das Ausgabe-Streaming (Empfangen von Token in Echtzeit). Informationen zu Eingabemodi (wie Sie Nachrichten senden), finden Sie unter Nachrichten an Agenten senden. Sie können auch Antworten mit dem Agent SDK über die CLI streamen.

​
Ausgabe-Streaming aktivieren

Um Streaming zu aktivieren, setzen Sie include_partial_messages (Python) oder includePartialMessages (TypeScript) in Ihren Optionen auf true. Dies führt dazu, dass das SDK StreamEvent-Nachrichten mit rohen API-Ereignissen liefert, die ankommen, zusätzlich zu den üblichen AssistantMessage- und ResultMessage-Objekten. Ihr Code muss dann:
  1. Den Typ jeder Nachricht überprüfen, um StreamEvent von anderen Nachrichtentypen zu unterscheiden
  2. Für StreamEvent das Feld event extrahieren und seinen type überprüfen
  3. Nach content_block_delta-Ereignissen suchen, bei denen delta.type gleich text_delta ist, die die tatsächlichen Text-Chunks enthalten
Das folgende Beispiel aktiviert Streaming und gibt Text-Chunks aus, während sie ankommen. Beachten Sie die verschachtelten Typüberprüfungen: zuerst für StreamEvent, dann für content_block_delta, dann für 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())

​
StreamEvent-Referenz

Wenn Teillmeldungen aktiviert sind, erhalten Sie rohe Claude-API-Streaming-Ereignisse, die in einem Objekt verpackt sind. Der Typ hat in jedem SDK unterschiedliche Namen:
  • Python: StreamEvent (importieren aus claude_agent_sdk.types)
  • TypeScript: SDKPartialAssistantMessage mit type: 'stream_event'
Beide enthalten rohe Claude-API-Ereignisse, nicht angesammelte Text. Sie müssen Text-Deltas selbst extrahieren und ansammeln. Hier ist die Struktur jedes Typs: 
@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
Das Feld event enthält das rohe Streaming-Ereignis aus der Claude API. Häufige Ereignistypen sind:
EreignistypBeschreibung
message_startBeginn einer neuen Nachricht
content_block_startBeginn eines neuen Inhaltsblocks (Text oder Tool-Verwendung)
content_block_deltaInkrementelle Aktualisierung des Inhalts
content_block_stopEnde eines Inhaltsblocks
message_deltaAktualisierungen auf Nachrichtenebene (Stoppgrund, Nutzung)
message_stopEnde der Nachricht

​
Nachrichtenfluss

Mit aktivierten Teillmeldungen erhalten Sie Nachrichten in dieser Reihenfolge: 
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
Ohne aktivierte Teillmeldungen (include_partial_messages in Python, includePartialMessages in TypeScript) erhalten Sie alle Nachrichtentypen außer StreamEvent. Häufige Typen sind SystemMessage (Sitzungsinitialisierung), AssistantMessage (vollständige Antworten), ResultMessage (Endergebnis) und eine kompakte Grenzmarkierung, die anzeigt, wann der Gesprächsverlauf komprimiert wurde (SDKCompactBoundaryMessage in TypeScript; SystemMessage mit Subtyp "compact_boundary" in Python).

​
Text-Antworten streamen

Um Text anzuzeigen, während er generiert wird, suchen Sie nach content_block_delta-Ereignissen, bei denen delta.type gleich text_delta ist. Diese enthalten die inkrementellen Text-Chunks. Das folgende Beispiel gibt jeden Chunk aus, während er ankommt: 
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())

​
Tool-Aufrufe streamen

Tool-Aufrufe werden auch inkrementell gestreamt. Sie können verfolgen, wann Tools starten, ihre Eingabe erhalten, während sie generiert wird, und sehen, wann sie abgeschlossen sind. Das folgende Beispiel verfolgt das aktuell aufgerufene Tool und sammelt die JSON-Eingabe, während sie gestreamt wird. Es verwendet drei Ereignistypen:
  • content_block_start: Tool beginnt
  • content_block_delta mit input_json_delta: Eingabe-Chunks kommen an
  • content_block_stop: Tool-Aufruf abgeschlossen
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())

​
Streaming-UI erstellen

Dieses Beispiel kombiniert Text- und Tool-Streaming in eine kohärente Benutzeroberfläche. Es verfolgt, ob der Agent gerade ein Tool ausführt (mit einem in_tool-Flag), um Statusanzeigen wie [Using Read...] anzuzeigen, während Tools ausgeführt werden. Text wird normal gestreamt, wenn nicht in einem Tool, und die Tool-Fertigstellung löst eine „done”-Nachricht aus. Dieses Muster ist nützlich für Chat-Schnittstellen, die während mehrstufiger Agent-Aufgaben Fortschritt anzeigen müssen. 
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())

​
Bekannte Einschränkungen

  • Strukturierte Ausgabe: Das JSON-Ergebnis erscheint nur in der finalen ResultMessage.structured_output, nicht als Streaming-Deltas. Weitere Informationen finden Sie unter Strukturierte Ausgaben.

​
Nächste Schritte

Jetzt, da Sie Text und Tool-Aufrufe in Echtzeit streamen können, erkunden Sie diese verwandten Themen: