> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/anthropics/claude-agent-sdk-python/llms.txt
> Use this file to discover all available pages before exploring further.
# 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
```python theme={null}
def __init__(
self,
options: ClaudeAgentOptions | None = None,
transport: Transport | None = None,
)
```
### Parameters
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
Optional custom transport implementation. If None, uses default subprocess transport.
## Methods
### connect()
Connect to Claude with a prompt or message stream.
```python theme={null}
async def connect(
self,
prompt: str | AsyncIterable[dict[str, Any]] | None = None
) -> None
```
Initial prompt or message stream. If None, connects without sending any messages (for interactive use).
**Example:**
```python theme={null}
client = ClaudeSDKClient()
await client.connect() # Connect without initial prompt
```
### query()
Send a new request in streaming mode.
```python theme={null}
async def query(
self,
prompt: str | AsyncIterable[dict[str, Any]],
session_id: str = "default"
) -> None
```
Either a string message or an async iterable of message dictionaries.
Session identifier for the conversation.
**Example:**
```python theme={null}
await client.query("What is the capital of France?")
```
### receive\_messages()
Receive all messages from Claude as they arrive.
```python theme={null}
async def receive_messages(self) -> AsyncIterator[Message]
```
**Returns:** AsyncIterator yielding messages indefinitely until disconnect.
**Example:**
```python theme={null}
async for message in client.receive_messages():
if isinstance(message, AssistantMessage):
print(message.content)
```
### receive\_response()
Receive messages until and including a ResultMessage.
```python theme={null}
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:**
```python theme={null}
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.
```python theme={null}
async def interrupt(self) -> None
```
**Example:**
```python theme={null}
await client.interrupt()
```
### set\_permission\_mode()
Change permission mode during conversation.
```python theme={null}
async def set_permission_mode(self, mode: str) -> None
```
The permission mode to set:
* `'default'`: CLI prompts for dangerous tools
* `'acceptEdits'`: Auto-accept file edits
* `'bypassPermissions'`: Allow all tools (use with caution)
**Example:**
```python theme={null}
# 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.
```python theme={null}
async def set_model(self, model: str | None = None) -> None
```
The model to use, or None for default. Examples:
* `'claude-sonnet-4-5'`
* `'claude-opus-4-1-20250805'`
* `'claude-opus-4-20250514'`
**Example:**
```python theme={null}
await client.set_model('claude-sonnet-4-5')
```
### rewind\_files()
Rewind tracked files to their state at a specific user message.
```python theme={null}
async def rewind_files(self, user_message_id: str) -> None
```
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:**
```python theme={null}
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.
```python theme={null}
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:**
```python theme={null}
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.
```python theme={null}
async def reconnect_mcp_server(self, server_name: str) -> None
```
The name of the MCP server to reconnect.
**Example:**
```python theme={null}
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.
```python theme={null}
async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None
```
The name of the MCP server to toggle.
True to enable the server, False to disable it.
**Example:**
```python theme={null}
# 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.
```python theme={null}
async def stop_task(self, task_id: str) -> None
```
The task ID from `task_notification` events.
After this resolves, a `task_notification` system message with status `'stopped'` will be emitted.
**Example:**
```python theme={null}
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.
```python theme={null}
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:**
```python theme={null}
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.
```python theme={null}
async def disconnect(self) -> None
```
**Example:**
```python theme={null}
await client.disconnect()
```
## Context Manager Support
The client supports async context manager protocol for automatic connection/disconnection:
```python theme={null}
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
```python theme={null}
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()](/api/query) - For simple one-shot queries
* [ClaudeAgentOptions](/api/types/claude-agent-options) - Configuration options
* [Message Types](/api/types/messages) - Available message types
* [Transport](/api/claude-sdk-client) - Custom transport implementations