Leitfäden

Antworten in Echtzeit streamen

Erhalten Sie Echtzeit-Antworten vom Agent SDK, während Text und Tool-Aufrufe eingehen

Standardmäßig liefert das Agent SDK vollständige AssistantMessage-Objekte, nachdem Claude die Generierung jeder Antwort abgeschlossen hat. Um inkrementelle Updates zu erhalten, während Text und Tool-Aufrufe generiert werden, aktivieren Sie das Streaming von Teilnachrichten, 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). Für Eingabemodi (wie Sie Nachrichten senden), siehe Nachrichten an Agenten senden. Sie können auch Antworten mit dem Agent SDK über die CLI streamen.

Streaming-Ausgabe 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 liefert, die rohe API-Events enthalten, während sie ankommen, zusätzlich zu den üblichen AssistantMessage und ResultMessage.

Ihr Code muss dann:

Den Typ jeder Nachricht überprüfen, um StreamEvent von anderen Nachrichtentypen zu unterscheiden
Für StreamEvent das Feld event extrahieren und seinen type überprüfen
Nach content_block_delta-Events 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 Teilnachrichten aktiviert sind, erhalten Sie rohe Claude-API-Streaming-Events, die in einem Objekt verpackt sind. Der Typ hat unterschiedliche Namen in jedem SDK:

Python: StreamEvent (importieren aus claude_agent_sdk.types)
TypeScript: SDKPartialAssistantMessage mit type: 'stream_event'

Beide enthalten rohe Claude-API-Events, nicht angesammelten Text. Sie müssen Text-Deltas selbst extrahieren und ansammeln. Hier ist die Struktur jedes Typs:

@dataclass
class StreamEvent:
    uuid: str                      # Eindeutige Kennung für dieses Event
    session_id: str                # Sitzungskennung
    event: dict[str, Any]          # Das rohe Claude-API-Stream-Event
    parent_tool_use_id: str | None # Parent-Tool-ID, falls von einem Subagenten

Das Feld event enthält das rohe Streaming-Event aus der Claude-API. Häufige Event-Typen sind:

Event-Typ	Beschreibung
`message_start`	Anfang einer neuen Nachricht
`content_block_start`	Anfang eines neuen Content-Blocks (Text oder Tool-Nutzung)
`content_block_delta`	Inkrementelle Aktualisierung des Inhalts
`content_block_stop`	Ende eines Content-Blocks
`message_delta`	Aktualisierungen auf Nachrichtenebene (Stoppgrund, Nutzung)
`message_stop`	Ende der Nachricht

Nachrichtenfluss

Mit aktivierten Teilnachrichten 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 - vollständige Nachricht mit allen Inhalten
... Tool wird ausgeführt ...
... weitere Streaming-Events für nächsten Turn ...
ResultMessage - Endergebnis

Ohne aktivierte Teilnachrichten (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 CompactBoundaryMessage (zeigt an, wenn der Gesprächsverlauf komprimiert wurde).

Text-Antworten streamen

Um Text anzuzeigen, während er generiert wird, suchen Sie nach content_block_delta-Events, 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, wenn Tools starten, ihren Input erhalten, während er generiert wird, und sehen, wenn sie abgeschlossen sind. Das folgende Beispiel verfolgt das aktuelle aufgerufene Tool und sammelt den JSON-Input, während er eingegeben wird. Es verwendet drei Event-Typen:

content_block_start: Tool beginnt
content_block_delta mit input_json_delta: Input-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())

Eine Streaming-UI erstellen

Dieses Beispiel kombiniert Text- und Tool-Streaming in eine kohärente UI. Es verfolgt, ob der Agent gerade ein Tool ausführt (mit einem in_tool-Flag), um Statusindikatoren wie [Using Read...] anzuzeigen, während Tools laufen. 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-Interfaces, die den Fortschritt während mehrstufiger Agent-Aufgaben 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

Einige SDK-Funktionen sind mit Streaming nicht kompatibel:

Extended thinking: Wenn Sie explizit max_thinking_tokens (Python) oder maxThinkingTokens (TypeScript) setzen, werden StreamEvent-Nachrichten nicht ausgegeben. Sie erhalten nur vollständige Nachrichten nach jedem Turn. Beachten Sie, dass Thinking standardmäßig im SDK deaktiviert ist, daher funktioniert Streaming, es sei denn, Sie aktivieren es.
Strukturierte Ausgabe: Das JSON-Ergebnis erscheint nur in der finalen ResultMessage.structured_output, nicht als Streaming-Deltas. Siehe strukturierte Ausgaben für Details.

Nächste Schritte

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

Interaktive vs. One-Shot-Abfragen: Wählen Sie zwischen Eingabemodi für Ihren Anwendungsfall
Strukturierte Ausgaben: Erhalten Sie typisierte JSON-Antworten vom Agent
Berechtigungen: Kontrollieren Sie, welche Tools der Agent verwenden kann

Was this page helpful?

Leitfäden

Antworten in Echtzeit streamen

Erhalten Sie Echtzeit-Antworten vom Agent SDK, während Text und Tool-Aufrufe eingehen

Streaming-Ausgabe aktivieren

Ihr Code muss dann:

Den Typ jeder Nachricht überprüfen, um StreamEvent von anderen Nachrichtentypen zu unterscheiden
Für StreamEvent das Feld event extrahieren und seinen type überprüfen
Nach content_block_delta-Events suchen, bei denen delta.type gleich text_delta ist, die die tatsächlichen Text-Chunks enthalten

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 Teilnachrichten aktiviert sind, erhalten Sie rohe Claude-API-Streaming-Events, die in einem Objekt verpackt sind. Der Typ hat unterschiedliche Namen in jedem SDK:

Python: StreamEvent (importieren aus claude_agent_sdk.types)
TypeScript: SDKPartialAssistantMessage mit type: 'stream_event'

Beide enthalten rohe Claude-API-Events, nicht angesammelten Text. Sie müssen Text-Deltas selbst extrahieren und ansammeln. Hier ist die Struktur jedes Typs:

@dataclass
class StreamEvent:
    uuid: str                      # Eindeutige Kennung für dieses Event
    session_id: str                # Sitzungskennung
    event: dict[str, Any]          # Das rohe Claude-API-Stream-Event
    parent_tool_use_id: str | None # Parent-Tool-ID, falls von einem Subagenten

Das Feld event enthält das rohe Streaming-Event aus der Claude-API. Häufige Event-Typen sind:

Event-Typ	Beschreibung
`message_start`	Anfang einer neuen Nachricht
`content_block_start`	Anfang eines neuen Content-Blocks (Text oder Tool-Nutzung)
`content_block_delta`	Inkrementelle Aktualisierung des Inhalts
`content_block_stop`	Ende eines Content-Blocks
`message_delta`	Aktualisierungen auf Nachrichtenebene (Stoppgrund, Nutzung)
`message_stop`	Ende der Nachricht

Nachrichtenfluss

Mit aktivierten Teilnachrichten 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 - vollständige Nachricht mit allen Inhalten
... Tool wird ausgeführt ...
... weitere Streaming-Events für nächsten Turn ...
ResultMessage - Endergebnis

Text-Antworten streamen

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

content_block_start: Tool beginnt
content_block_delta mit input_json_delta: Input-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())

Eine Streaming-UI erstellen

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

Einige SDK-Funktionen sind mit Streaming nicht kompatibel:

Extended thinking: Wenn Sie explizit max_thinking_tokens (Python) oder maxThinkingTokens (TypeScript) setzen, werden StreamEvent-Nachrichten nicht ausgegeben. Sie erhalten nur vollständige Nachrichten nach jedem Turn. Beachten Sie, dass Thinking standardmäßig im SDK deaktiviert ist, daher funktioniert Streaming, es sei denn, Sie aktivieren es.
Strukturierte Ausgabe: Das JSON-Ergebnis erscheint nur in der finalen ResultMessage.structured_output, nicht als Streaming-Deltas. Siehe strukturierte Ausgaben für Details.

Nächste Schritte

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

Interaktive vs. One-Shot-Abfragen: Wählen Sie zwischen Eingabemodi für Ihren Anwendungsfall
Strukturierte Ausgaben: Erhalten Sie typisierte JSON-Antworten vom Agent
Berechtigungen: Kontrollieren Sie, welche Tools der Agent verwenden kann

Was this page helpful?