Antworten in Echtzeit streamen

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:

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

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

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  # Always None

type SDKPartialAssistantMessage = {
  type: "stream_event";
  event: BetaRawMessageStreamEvent; // From Anthropic SDK
  parent_tool_use_id: string | null;
  uuid: UUID;
  session_id: string;
  ttft_ms?: number; // Time to first token in ms, present only on message_start events
};

Das Feld parent_tool_use_id ist in Python immer None und in TypeScript immer null. Streaming-Ereignisse werden nur für die Hauptsitzung ausgegeben; Token-Level-Deltas von Subagenten werden nicht weitergeleitet. Um die Ausgabe einem Subagenten zuzuordnen, verwenden Sie vollständige Nachrichten, die parent_tool_use_id enthalten. Siehe Subagenten-Aufruf erkennen. Das Feld event enthält das rohe Streaming-Ereignis aus der Claude API. Häufige Ereignistypen sind:

Ereignistyp	Beschreibung
`message_start`	Beginn einer neuen Nachricht
`content_block_start`	Beginn eines neuen Inhaltsblocks (Text oder Tool-Verwendung)
`content_block_delta`	Inkrementelle Aktualisierung des Inhalts
`content_block_stop`	Ende eines Inhaltsblocks
`message_delta`	Aktualisierungen auf Nachrichtenebene (Stoppgrund, Nutzung)
`message_stop`	Ende 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())

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "Explain how databases work",
  options: { includePartialMessages: true }
})) {
  if (message.type === "stream_event") {
    const event = message.event;
    if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
      process.stdout.write(event.delta.text);
    }
  }
}

console.log(); // Final newline

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

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

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

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

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:

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

Agent SDK

Kernkonzepte

Eingabe und Ausgabe

Mit Tools erweitern

Verhalten anpassen

Kontrolle und Beobachtbarkeit

Bereitstellung

SDK-Referenzen

Antworten in Echtzeit streamen

Ausgabe-Streaming aktivieren

StreamEvent-Referenz

Nachrichtenfluss

Text-Antworten streamen

Tool-Aufrufe streamen

Streaming-UI erstellen

Bekannte Einschränkungen

Nächste Schritte

​Ausgabe-Streaming aktivieren

​StreamEvent-Referenz

​Nachrichtenfluss

​Text-Antworten streamen

​Tool-Aufrufe streamen

​Streaming-UI erstellen

​Bekannte Einschränkungen

​Nächste Schritte

Ausgabe-Streaming aktivieren

StreamEvent-Referenz

Nachrichtenfluss

Text-Antworten streamen

Tool-Aufrufe streamen

Streaming-UI erstellen

Bekannte Einschränkungen

Nächste Schritte