ClaudeSDKClient

Client for bidirectional, interactive conversations with Claude Code

Overview

The ClaudeSDKClient provides full control over the conversation flow with support for streaming, interrupts, and dynamic message sending. For simple one-shot queries, consider using the query() function instead.

Key Features

• Bidirectional: Send and receive messages at any time
• Stateful: Maintains conversation context across messages
• Interactive: Send follow-ups based on responses
• Control flow: Support for interrupts and session management

When to Use ClaudeSDKClient

• Building chat interfaces or conversational UIs
• Interactive debugging or exploration sessions
• Multi-turn conversations with context
• When you need to react to Claude’s responses
• Real-time applications with user input
• When you need interrupt capabilities

When to Use query() Instead

• Simple one-off questions
• Batch processing of prompts
• Fire-and-forget automation scripts
• When all inputs are known upfront
• Stateless operations

Async Context Limitation: As of v0.0.20, you cannot use a ClaudeSDKClient instance across different async runtime contexts (e.g., different trio nurseries or asyncio task groups). The client internally maintains a persistent anyio task group for reading messages that remains active from connect() until disconnect(). Complete all operations with the client within the same async context where it was connected.

Constructor

def __init__(
    self,
    options: ClaudeAgentOptions | None = None,
    transport: Transport | None = None,
)

Parameters

options · ClaudeAgentOptions | None · default: "None"

Optional configuration (defaults to ClaudeAgentOptions() if None).

Key options:

• permission_mode: Control tool execution
• cwd: Working directory
• system_prompt: Custom system prompt
• mcp_servers: MCP server configurations
• can_use_tool: Callback for tool permission control
• hooks: Event hooks for custom behavior
• enable_file_checkpointing: Enable file state tracking

transport · Transport | None · default: "None"

Optional custom transport implementation. If None, uses default subprocess transport.

Methods

connect()

Connect to Claude with a prompt or message stream.

async def connect(
    self,
    prompt: str | AsyncIterable[dict[str, Any]] | None = None
) -> None

prompt · str | AsyncIterable[dict[str, Any]] | None · default: "None"

Initial prompt or message stream. If None, connects without sending any messages (for interactive use).

Example:

client = ClaudeSDKClient()
await client.connect()  # Connect without initial prompt

query()

Send a new request in streaming mode.

async def query(
    self,
    prompt: str | AsyncIterable[dict[str, Any]],
    session_id: str = "default"
) -> None

prompt · str | AsyncIterable[dict[str, Any]] · required

Either a string message or an async iterable of message dictionaries.

session_id · str · default: "default"

Session identifier for the conversation.

Example:

await client.query("What is the capital of France?")

receive_messages()

Receive all messages from Claude as they arrive.

async def receive_messages(self) -> AsyncIterator[Message]

Returns: AsyncIterator yielding messages indefinitely until disconnect.

Example:

async for message in client.receive_messages():
    if isinstance(message, AssistantMessage):
        print(message.content)

receive_response()

Receive messages until and including a ResultMessage.

async def receive_response(self) -> AsyncIterator[Message]

Returns: AsyncIterator that automatically terminates after yielding a ResultMessage.

Stopping Behavior:

• Yields each message as it’s received
• Terminates immediately after yielding a ResultMessage
• The ResultMessage IS included in the yielded messages
• If no ResultMessage is received, the iterator continues indefinitely

Example:

await client.query("Explain Python decorators")

async for msg in client.receive_response():
    if isinstance(msg, AssistantMessage):
        for block in msg.content:
            if isinstance(block, TextBlock):
                print(f"Claude: {block.text}")
    elif isinstance(msg, ResultMessage):
        print(f"Cost: ${msg.total_cost_usd:.4f}")
        # Iterator terminates after this

interrupt()

Send interrupt signal to stop the current operation.

async def interrupt(self) -> None

Example:

await client.interrupt()

set_permission_mode()

Change permission mode during conversation.

async def set_permission_mode(self, mode: str) -> None

mode · str · required

The permission mode to set:

• 'default': CLI prompts for dangerous tools
• 'acceptEdits': Auto-accept file edits
• 'bypassPermissions': Allow all tools (use with caution)

Example:

# Start with default permissions
await client.query("Help me analyze this codebase")

# Switch to auto-accept edits
await client.set_permission_mode('acceptEdits')
await client.query("Now implement the fix we discussed")

set_model()

Change the AI model during conversation.

async def set_model(self, model: str | None = None) -> None

model · str | None · default: "None"

The model to use, or None for default. Examples:

• 'claude-sonnet-4-5'
• 'claude-opus-4-1-20250805'
• 'claude-opus-4-20250514'

Example:

await client.set_model('claude-sonnet-4-5')

rewind_files()

Rewind tracked files to their state at a specific user message.

async def rewind_files(self, user_message_id: str) -> None

user_message_id · str · required

UUID of the user message to rewind to. This should be the uuid field from a UserMessage received during the conversation.

Requirements:

• enable_file_checkpointing=True to track file changes
• extra_args={"replay-user-messages": None} to receive UserMessage objects with uuid

Example:

options = ClaudeAgentOptions(
    enable_file_checkpointing=True,
    extra_args={"replay-user-messages": None},
)
async with ClaudeSDKClient(options) as client:
    await client.query("Make some changes to my files")
    async for msg in client.receive_response():
        if isinstance(msg, UserMessage) and msg.uuid:
            checkpoint_id = msg.uuid  # Save for later

    # Rewind to that point
    await client.rewind_files(checkpoint_id)

get_mcp_status()

Get current MCP server connection status.

async def get_mcp_status(self) -> McpStatusResponse

Returns: McpStatusResponse dictionary with:

• mcpServers: List of server status entries

Each server status includes:

• name: Server name (str)
• status: Connection status ('connected', 'pending', 'failed', 'needs-auth', 'disabled')
• serverInfo: MCP server name/version (when connected)
• error: Error message (when status is 'failed')
• config: Server configuration
• scope: Configuration scope
• tools: List of tools provided (when connected)

Example:

status = await client.get_mcp_status()
for server in status["mcpServers"]:
    print(f"{server['name']}: {server['status']}")
    if server["status"] == "failed":
        print(f"  Error: {server.get('error')}")

reconnect_mcp_server()

Reconnect a disconnected or failed MCP server.

async def reconnect_mcp_server(self, server_name: str) -> None

server_name · str · required

The name of the MCP server to reconnect.

Example:

status = await client.get_mcp_status()
for server in status.get("mcpServers", []):
    if server["status"] == "failed":
        await client.reconnect_mcp_server(server["name"])

toggle_mcp_server()

Enable or disable an MCP server.

async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None

server_name · str · required

The name of the MCP server to toggle.

enabled · bool · required

True to enable the server, False to disable it.

Example:

# Disable a server
await client.toggle_mcp_server("my-server", enabled=False)
await client.query("Do something without my-server tools")

# Re-enable it
await client.toggle_mcp_server("my-server", enabled=True)

stop_task()

Stop a running task.

async def stop_task(self, task_id: str) -> None

task_id · str · required

The task ID from task_notification events.

After this resolves, a task_notification system message with status 'stopped' will be emitted.

Example:

await client.query("Start a long-running task")
# Listen for task_notification to get task_id, then:
await client.stop_task("task-abc123")

get_server_info()

Get server initialization info including available commands and output styles.

async def get_server_info(self) -> dict[str, Any] | None

Returns: Dictionary with server info, or None if not in streaming mode.

Includes:

• Available commands (slash commands, system commands, etc.)
• Current and available output styles
• Server capabilities

Example:

info = await client.get_server_info()
if info:
    print(f"Commands available: {len(info.get('commands', []))}")
    print(f"Output style: {info.get('output_style', 'default')}")

disconnect()

Disconnect from Claude.

async def disconnect(self) -> None

Example:

await client.disconnect()

Context Manager Support

The client supports async context manager protocol for automatic connection/disconnection:

async with ClaudeSDKClient() as client:
    await client.query("Hello!")
    async for msg in client.receive_response():
        print(msg)
# Automatically disconnects when exiting context

Complete Example

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from claude_agent_sdk.types import AssistantMessage, TextBlock, ResultMessage

async def main():
    options = ClaudeAgentOptions(
        cwd="/home/user/project",
        permission_mode="acceptEdits",
        system_prompt="You are a helpful coding assistant"
    )
    
    async with ClaudeSDKClient(options) as client:
        # First query
        await client.query("Analyze the codebase structure")
        
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")
            elif isinstance(msg, ResultMessage):
                print(f"Cost: ${msg.total_cost_usd:.4f}")
        
        # Follow-up query
        await client.query("Now create a test file")
        
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

Related

• query() - For simple one-shot queries
• ClaudeAgentOptions - Configuration options
• Message Types - Available message types
• Transport - Custom transport implementations