Streaming-Eingabe
Verständnis der zwei Eingabemodi für Claude Agent SDK und wann jeder verwendet wird
Übersicht
Das Claude Agent SDK unterstützt zwei unterschiedliche Eingabemodi für die Interaktion mit Agenten:
- Streaming-Eingabemodus: eine persistente, interaktive Sitzung
- Einzelne Nachricht-Eingabe: One-Shot-Abfragen, die Sitzungszustand und Wiederaufnahme verwenden
Streaming-Eingabemodus (empfohlen)
Der Streaming-Eingabemodus ist die bevorzugte Methode zur Verwendung des Claude Agent SDK. Er bietet vollen Zugriff auf die Fähigkeiten des Agenten und ermöglicht umfassende, interaktive Erfahrungen.
Er ermöglicht es dem Agenten, als langlebiger Prozess zu arbeiten, der Benutzereingaben entgegennimmt, Unterbrechungen verarbeitet, Berechtigungsanfragen anzeigt und die Sitzungsverwaltung übernimmt.
Vorteile
Im Streaming-Eingabemodus arbeiten Sie in einer persistenten Sitzung mit diesen Funktionen:
- Bild-Uploads: Hängen Sie Bilder direkt an Nachrichten an, um sie visuell analysieren und verstehen zu lassen
- Nachrichten in der Warteschlange: Senden Sie mehrere Nachrichten, die nacheinander verarbeitet werden, mit der Möglichkeit zur Unterbrechung
- Tool-Integration: Voller Zugriff auf alle Tools und benutzerdefinierten MCP-Server während der Sitzung
- Echtzeit-Feedback: Sehen Sie Antworten, während sie generiert werden, nicht nur die Endergebnisse
- Kontextpersistenz: Behalten Sie den Konversationskontext über mehrere Turns hinweg auf natürliche Weise bei
Implementierungsbeispiel
Diese Beispiele lesen ein Bild namens diagram.png aus dem Arbeitsverzeichnis. Erstellen Sie dort zuerst eines, oder ändern Sie den Dateinamen so, dass er auf Ihr eigenes Bild verweist.
import { query, type SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
import { readFile } from "fs/promises";
async function* generateMessages(): AsyncGenerator<SDKUserMessage> {
// First message
yield {
type: "user",
message: {
role: "user",
content: "Analyze this codebase for security issues"
},
parent_tool_use_id: null
};
// Wait for conditions or user input
await new Promise((resolve) => setTimeout(resolve, 2000));
// Follow-up with image
yield {
type: "user",
message: {
role: "user",
content: [
{
type: "text",
text: "Review this architecture diagram"
},
{
type: "image",
source: {
type: "base64",
media_type: "image/png",
data: await readFile("diagram.png", "base64")
}
}
]
},
parent_tool_use_id: null
};
}
// Process streaming responses
for await (const message of query({
prompt: generateMessages(),
options: {
maxTurns: 10,
allowedTools: ["Read", "Grep"]
}
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
AssistantMessage,
TextBlock,
)
import asyncio
import base64
async def streaming_analysis():
async def message_generator():
# First message
yield {
"type": "user",
"message": {
"role": "user",
"content": "Analyze this codebase for security issues",
},
}
# Wait for conditions
await asyncio.sleep(2)
# Follow-up with image
with open("diagram.png", "rb") as f:
image_data = base64.b64encode(f.read()).decode()
yield {
"type": "user",
"message": {
"role": "user",
"content": [
{"type": "text", "text": "Review this architecture diagram"},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image_data,
},
},
],
},
}
# Use ClaudeSDKClient for streaming input
options = ClaudeAgentOptions(max_turns=10, allowed_tools=["Read", "Grep"])
async with ClaudeSDKClient(options) as client:
# Send streaming input
await client.query(message_generator())
# Process responses
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)
asyncio.run(streaming_analysis())
Wenn Sie das Beispiel ausführen, gibt die TypeScript-Version jede Antwort aus, sobald sie abgeschlossen ist. Die receive_response()-Schleife der Python-Version endet bei der ersten Ergebnisnachricht, sodass sie die Sicherheitsanalyse ausgibt; um beide Antworten zu lesen, verwenden Sie pro Nachricht ein Paar aus query() und receive_response(), wie im Beispiel der Python-Referenz zum Fortsetzen einer Konversation gezeigt.
Wenn die source eines Bildblocks fehlt oder kein Objekt ist, meldet das SDK keinen Fehler. Claude Code sendet Claude anstelle des Bildes einen Texthinweis, etwa [Image could not be processed: image block has no source object], und die Sitzung wird fortgesetzt.
Einzelne Nachricht-Eingabe
Die Eingabe einer einzelnen Nachricht ist einfacher, aber begrenzter.
Wann sollte die Eingabe einer einzelnen Nachricht verwendet werden
Verwenden Sie die Eingabe einer einzelnen Nachricht, wenn:
- Sie eine One-Shot-Antwort benötigen
- Sie keine Bild-Anhänge oder Mid-Session-Kontrollmethoden benötigen
- Sie in einer zustandslosen Umgebung arbeiten müssen, z. B. in einer Lambda-Funktion
Einschränkungen
Wenn eine Abfrage mit einem Fehler endet, z. B. error_max_turns, löst ein einzelner query()-Aufruf einen Fehler aus, der den Fehlertext nach dem Ausgeben der endgültigen Ergebnisnachricht enthält. Wickeln Sie daher die Schleife in einen Try-Block ein, wenn Ihr Code fortgesetzt werden muss. Siehe Ergebnis verarbeiten für die Ergebnis-Untertypen.
Implementierungsbeispiel
import { query } from "@anthropic-ai/claude-agent-sdk";
// Simple one-shot query
// query() throws after an error result, such as error_max_turns
try {
for await (const message of query({
prompt: "Explain the authentication flow",
options: {
maxTurns: 5,
allowedTools: ["Read", "Grep"]
}
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}
} catch (error) {
console.error(`Query failed: ${error}`);
}
// Continue conversation with session management
try {
for await (const message of query({
prompt: "Now explain the authorization process",
options: {
continue: true,
maxTurns: 5
}
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}
} catch (error) {
console.error(`Query failed: ${error}`);
}
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
import asyncio
async def single_message_example():
# Simple one-shot query using query() function
# query() raises ResultError after an error result, such as error_max_turns
try:
async for message in query(
prompt="Explain the authentication flow",
options=ClaudeAgentOptions(max_turns=5, allowed_tools=["Read", "Grep"]),
):
if isinstance(message, ResultMessage) and message.subtype == "success":
print(message.result)
except Exception as e:
print(f"Query failed: {e}")
# Continue conversation with session management
try:
async for message in query(
prompt="Now explain the authorization process",
options=ClaudeAgentOptions(continue_conversation=True, max_turns=5),
):
if isinstance(message, ResultMessage) and message.subtype == "success":
print(message.result)
except Exception as e:
print(f"Query failed: {e}")
asyncio.run(single_message_example())
Wenn Sie das Beispiel ausführen, gibt jede Abfrage ihren endgültigen Ergebnistext aus: zuerst die Authentifizierungserklärung, dann die Autorisierungserklärung.