Understanding the two input modes for Claude Agent SDK and when to use each
Overview
The Claude Agent SDK supports two distinct input modes for interacting with agents:
- Streaming Input Mode: a persistent, interactive session
- Single Message Input: one-shot queries that use session state and resuming
Streaming Input Mode (Recommended)
Streaming input mode is the preferred way to use the Claude Agent SDK. It provides full access to the agent's capabilities and enables rich, interactive experiences.
It allows the agent to operate as a long lived process that takes in user input, handles interruptions, surfaces permission requests, and handles session management.
How It Works
sequenceDiagram
participant App as Your Application
participant Agent as Claude Agent
participant Tools as Tools/Hooks
participant FS as Environment/<br/>File System
App->>Agent: Initialize with AsyncGenerator
activate Agent
App->>Agent: Yield Message 1
Agent->>Tools: Execute tools
Tools->>FS: Read files
FS-->>Tools: File contents
Tools->>FS: Write/Edit files
FS-->>Tools: Success/Error
Agent-->>App: Stream partial response
Agent-->>App: Stream more content...
Agent->>App: Complete Message 1
App->>Agent: Yield Message 2 + Image
Agent->>Tools: Process image & execute
Tools->>FS: Access filesystem
FS-->>Tools: Operation results
Agent-->>App: Stream response 2
App->>Agent: Queue Message 3
App->>Agent: Interrupt/Cancel
Agent->>App: Handle interruption
Note over App,Agent: Session stays alive
Note over Tools,FS: Persistent file system<br/>state maintained
deactivate Agent
Benefits
In streaming input mode, you work in a persistent session with these capabilities:
- Image uploads: attach images directly to messages for visual analysis and understanding
- Queued messages: send multiple messages that process sequentially, with ability to interrupt
- Tool integration: full access to all tools and custom MCP servers during the session
- Real-time feedback: see responses as they're generated, not just final results
- Context persistence: maintain conversation context across multiple turns naturally
Implementation Example
These examples read an image named diagram.png from the working directory. Create one there first, or change the filename to point at your own image.
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())
When you run the example, the TypeScript version prints each response as it completes. The Python version's receive_response() loop ends at the first result message, so it prints the security analysis; to read both responses, use one query() and receive_response() pair per message as shown in the Python reference's example of continuing a conversation.
In the TypeScript SDK, if your message generator throws, for example when a file it reads is missing, the stream ends with an error that reads Claude Code process aborted by user instead of the original error, so check the code inside your generator first when you see that message. The error may also be preceded by a long minified line of bundled SDK source, so read to the end of the output for the error text.
In the Python SDK, a generator exception is logged at debug level and the session stalls without raising, so if a streaming session hangs with no output, enable debug logging and check your generator.
Single Message Input
Single message input is simpler but more limited.
When to Use Single Message Input
Use single message input when:
- You need a one-shot response
- You do not need image attachments or mid-session control methods
- You need to operate in a stateless environment, such as a lambda function
Limitations
Single message input mode does not support:
- Direct image attachments in messages
- Dynamic message queueing
- Real-time interruption
- Natural multi-turn conversations
If a query ends with an error result, such as error_max_turns, a single message query() call raises an error that includes the failure text after yielding the final result message, so wrap the loop in a try block if your code needs to continue. See Handle the result for the result subtypes.
Implementation Example
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 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)
# The SDK raises a plain Exception for error results, so match Exception here
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())
When you run the example, each query prints its final result text: first the authentication explanation, then the authorization explanation.