0
# Interactive Client
1

2
The ClaudeSDKClient class provides full control over conversation flow with support for bidirectional communication, streaming, interrupts, and dynamic message sending. It's ideal for chat interfaces, interactive debugging, and multi-turn conversations.
3

4
## Capabilities
5

6
### ClaudeSDKClient Class
7

8
Client for bidirectional, interactive conversations with Claude Code. Maintains conversation state and allows sending messages at any time based on responses.
9

10
```python { .api }
11
class ClaudeSDKClient:
12
    """
13
    Client for bidirectional, interactive conversations with Claude Code.
14

15
    This client provides full control over the conversation flow with support
16
    for streaming, interrupts, and dynamic message sending. For simple one-shot
17
    queries, consider using the query() function instead.
18

19
    Key features:
20
    - Bidirectional: Send and receive messages at any time
21
    - Stateful: Maintains conversation context across messages
22
    - Interactive: Send follow-ups based on responses
23
    - Control flow: Support for interrupts and session management
24

25
    When to use ClaudeSDKClient:
26
    - Building chat interfaces or conversational UIs
27
    - Interactive debugging or exploration sessions
28
    - Multi-turn conversations with context
29
    - When you need to react to Claude's responses
30
    - Real-time applications with user input
31
    - When you need interrupt capabilities
32

33
    When to use query() instead:
34
    - Simple one-off questions
35
    - Batch processing of prompts
36
    - Fire-and-forget automation scripts
37
    - When all inputs are known upfront
38
    - Stateless operations
39

40
    Caveat: As of v0.0.20, you cannot use a ClaudeSDKClient instance across
41
    different async runtime contexts (e.g., different trio nurseries or asyncio
42
    task groups). The client internally maintains a persistent anyio task group
43
    for reading messages that remains active from connect() until disconnect().
44
    This means you must complete all operations with the client within the same
45
    async context where it was connected.
46
    """
47

48
    def __init__(
49
        self,
50
        options: ClaudeAgentOptions | None = None,
51
        transport: Transport | None = None,
52
    ):
53
        """
54
        Initialize Claude SDK client.
55

56
        Args:
57
            options: Optional configuration (ClaudeAgentOptions instance)
58
            transport: Optional custom transport implementation
59
        """
60

61
    async def connect(
62
        self, prompt: str | AsyncIterable[dict[str, Any]] | None = None
63
    ) -> None:
64
        """
65
        Connect to Claude with a prompt or message stream.
66

67
        Args:
68
            prompt: Optional initial prompt (string or AsyncIterable of message dicts)
69
                   If None, connects without sending initial messages (for interactive use)
70
        """
71

72
    async def receive_messages(self) -> AsyncIterator[Message]:
73
        """
74
        Receive all messages from Claude.
75

76
        Yields:
77
            Message objects (UserMessage, AssistantMessage, SystemMessage,
78
            ResultMessage, or StreamEvent)
79
        """
80

81
    async def query(
82
        self, prompt: str | AsyncIterable[dict[str, Any]], session_id: str = "default"
83
    ) -> None:
84
        """
85
        Send a new request in streaming mode.
86

87
        Args:
88
            prompt: Either a string message or an async iterable of message dictionaries
89
            session_id: Session identifier for the conversation (default: "default")
90
        """
91

92
    async def interrupt(self) -> None:
93
        """
94
        Send interrupt signal (only works with streaming mode).
95

96
        Interrupts the current Claude operation, stopping generation and
97
        tool execution.
98
        """
99

100
    async def set_permission_mode(self, mode: str) -> None:
101
        """
102
        Change permission mode during conversation (only works with streaming mode).
103

104
        Args:
105
            mode: The permission mode to set. Valid options:
106
                - 'default': CLI prompts for dangerous tools
107
                - 'acceptEdits': Auto-accept file edits
108
                - 'bypassPermissions': Allow all tools (use with caution)
109
        """
110

111
    async def set_model(self, model: str | None = None) -> None:
112
        """
113
        Change the AI model during conversation (only works with streaming mode).
114

115
        Args:
116
            model: The model to use, or None to use default. Examples:
117
                - 'claude-sonnet-4-5'
118
                - 'claude-opus-4-1-20250805'
119
                - 'claude-opus-4-20250514'
120
        """
121

122
    async def get_server_info(self) -> dict[str, Any] | None:
123
        """
124
        Get server initialization info including available commands and output styles.
125

126
        Returns initialization information from the Claude Code server including:
127
        - Available commands (slash commands, system commands, etc.)
128
        - Current and available output styles
129
        - Server capabilities
130

131
        Returns:
132
            Dictionary with server info, or None if not in streaming mode
133
        """
134

135
    async def receive_response(self) -> AsyncIterator[Message]:
136
        """
137
        Receive messages from Claude until and including a ResultMessage.
138

139
        This async iterator yields all messages in sequence and automatically terminates
140
        after yielding a ResultMessage (which indicates the response is complete).
141
        It's a convenience method over receive_messages() for single-response workflows.
142

143
        Stopping Behavior:
144
        - Yields each message as it's received
145
        - Terminates immediately after yielding a ResultMessage
146
        - The ResultMessage IS included in the yielded messages
147
        - If no ResultMessage is received, the iterator continues indefinitely
148

149
        Yields:
150
            Message objects (each message received until ResultMessage)
151

152
        Note:
153
            To collect all messages: messages = [msg async for msg in client.receive_response()]
154
            The final message in the list will always be a ResultMessage.
155
        """
156

157
    async def disconnect(self) -> None:
158
        """
159
        Disconnect from Claude.
160

161
        Closes the connection and cleans up resources.
162
        """
163

164
    async def __aenter__(self) -> "ClaudeSDKClient":
165
        """
166
        Enter async context - automatically connects with empty stream for interactive use.
167

168
        Returns:
169
            Self (the ClaudeSDKClient instance)
170
        """
171

172
    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool:
173
        """
174
        Exit async context - always disconnects.
175

176
        Returns:
177
            False (does not suppress exceptions)
178
        """
179
```
180

181
## Usage Examples
182

183
### Basic Interactive Session
184

185
```python
186
import anyio
187
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock
188

189
async def main():
190
    async with ClaudeSDKClient() as client:
191
        await client.query("What is the capital of France?")
192

193
        async for msg in client.receive_response():
194
            if isinstance(msg, AssistantMessage):
195
                for block in msg.content:
196
                    if isinstance(block, TextBlock):
197
                        print(f"Claude: {block.text}")
198

199
anyio.run(main)
200
```
201

202
### Multi-Turn Conversation
203

204
```python
205
import anyio
206
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage
207

208
async def main():
209
    async with ClaudeSDKClient() as client:
210
        # First question
211
        await client.query("What is Python?")
212
        async for msg in client.receive_response():
213
            if isinstance(msg, AssistantMessage):
214
                for block in msg.content:
215
                    if isinstance(block, TextBlock):
216
                        print(f"Claude: {block.text}")
217

218
        # Follow-up question
219
        await client.query("Can you show me a simple Python example?")
220
        async for msg in client.receive_response():
221
            if isinstance(msg, AssistantMessage):
222
                for block in msg.content:
223
                    if isinstance(block, TextBlock):
224
                        print(f"Claude: {block.text}")
225

226
anyio.run(main)
227
```
228

229
### Using with Options
230

231
```python
232
import anyio
233
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
234

235
async def main():
236
    options = ClaudeAgentOptions(
237
        system_prompt="You are a helpful Python expert",
238
        allowed_tools=["Read", "Bash"],
239
        permission_mode='default',
240
        cwd="/home/user/project"
241
    )
242

243
    async with ClaudeSDKClient(options=options) as client:
244
        await client.query("Analyze the Python files in this directory")
245

246
        async for msg in client.receive_response():
247
            print(msg)
248

249
anyio.run(main)
250
```
251

252
### Interrupting Execution
253

254
```python
255
import anyio
256
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage
257

258
async def main():
259
    async with ClaudeSDKClient() as client:
260
        await client.query("Generate 1000 lines of code")
261

262
        # Interrupt after a short delay
263
        async def interrupt_after_delay():
264
            await anyio.sleep(2)
265
            await client.interrupt()
266

267
        async with anyio.create_task_group() as tg:
268
            tg.start_soon(interrupt_after_delay)
269

270
            async for msg in client.receive_messages():
271
                if isinstance(msg, AssistantMessage):
272
                    print("Received message before interrupt")
273

274
anyio.run(main)
275
```
276

277
### Dynamic Permission Mode
278

279
```python
280
import anyio
281
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
282

283
async def main():
284
    options = ClaudeAgentOptions(
285
        allowed_tools=["Read", "Write", "Edit"],
286
        permission_mode='default'  # Start with default (ask for permissions)
287
    )
288

289
    async with ClaudeSDKClient(options=options) as client:
290
        # Review mode - ask for permissions
291
        await client.query("Help me analyze this codebase")
292
        async for msg in client.receive_response():
293
            print(msg)
294

295
        # Implementation mode - auto-accept edits
296
        await client.set_permission_mode('acceptEdits')
297
        await client.query("Now implement the fix we discussed")
298
        async for msg in client.receive_response():
299
            print(msg)
300

301
anyio.run(main)
302
```
303

304
### Switching Models
305

306
```python
307
import anyio
308
from claude_agent_sdk import ClaudeSDKClient
309

310
async def main():
311
    async with ClaudeSDKClient() as client:
312
        # Start with default model
313
        await client.query("Explain quantum computing briefly")
314
        async for msg in client.receive_response():
315
            print(msg)
316

317
        # Switch to a specific model for detailed implementation
318
        await client.set_model('claude-sonnet-4-5')
319
        await client.query("Now implement a quantum circuit simulator")
320
        async for msg in client.receive_response():
321
            print(msg)
322

323
anyio.run(main)
324
```
325

326
### Getting Server Info
327

328
```python
329
import anyio
330
from claude_agent_sdk import ClaudeSDKClient
331

332
async def main():
333
    async with ClaudeSDKClient() as client:
334
        info = await client.get_server_info()
335
        if info:
336
            print(f"Available commands: {len(info.get('commands', []))}")
337
            print(f"Output style: {info.get('output_style', 'default')}")
338
            print(f"Server capabilities: {info.get('capabilities', {})}")
339

340
anyio.run(main)
341
```
342

343
### Manual Connection Management
344

345
```python
346
import anyio
347
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
348

349
async def main():
350
    options = ClaudeAgentOptions(
351
        allowed_tools=["Read", "Write"]
352
    )
353

354
    client = ClaudeSDKClient(options=options)
355

356
    try:
357
        # Manual connect
358
        await client.connect()
359

360
        await client.query("Hello Claude")
361
        async for msg in client.receive_response():
362
            print(msg)
363

364
    finally:
365
        # Manual disconnect
366
        await client.disconnect()
367

368
anyio.run(main)
369
```
370

371
### Collecting All Messages
372

373
```python
374
import anyio
375
from claude_agent_sdk import ClaudeSDKClient, ResultMessage
376

377
async def main():
378
    async with ClaudeSDKClient() as client:
379
        await client.query("What is 2 + 2?")
380

381
        # Collect all messages into a list
382
        messages = [msg async for msg in client.receive_response()]
383

384
        # Last message is always ResultMessage
385
        result = messages[-1]
386
        if isinstance(result, ResultMessage):
387
            print(f"Total cost: ${result.total_cost_usd:.4f}")
388
            print(f"Number of turns: {result.num_turns}")
389
            print(f"Duration: {result.duration_ms}ms")
390

391
anyio.run(main)
392
```
393

394
### Streaming with AsyncIterable Prompts
395

396
```python
397
import anyio
398
from claude_agent_sdk import ClaudeSDKClient
399

400
async def main():
401
    async def message_stream():
402
        yield {"type": "user", "message": {"role": "user", "content": "First question"}, "session_id": "1"}
403
        await anyio.sleep(1)
404
        yield {"type": "user", "message": {"role": "user", "content": "Follow up"}, "session_id": "1"}
405

406
    async with ClaudeSDKClient() as client:
407
        await client.query(message_stream())
408

409
        async for msg in client.receive_messages():
410
            print(msg)
411

412
anyio.run(main)
413
```
414

415
### Using Custom Transport
416

417
```python
418
import anyio
419
from claude_agent_sdk import ClaudeSDKClient, Transport
420

421
class MyCustomTransport(Transport):
422
    # Implement custom transport logic
423
    async def connect(self) -> None:
424
        pass
425

426
    async def write(self, data: str) -> None:
427
        pass
428

429
    def read_messages(self):
430
        pass
431

432
    async def close(self) -> None:
433
        pass
434

435
    def is_ready(self) -> bool:
436
        return True
437

438
    async def end_input(self) -> None:
439
        pass
440

441
async def main():
442
    transport = MyCustomTransport()
443
    client = ClaudeSDKClient(transport=transport)
444

445
    async with client:
446
        await client.query("Hello")
447
        async for msg in client.receive_response():
448
            print(msg)
449

450
anyio.run(main)
451
```
452

453
### Error Handling
454

455
```python
456
import anyio
457
from claude_agent_sdk import (
458
    ClaudeSDKClient,
459
    ClaudeSDKError,
460
    CLINotFoundError,
461
    CLIConnectionError
462
)
463

464
async def main():
465
    try:
466
        async with ClaudeSDKClient() as client:
467
            await client.query("Hello Claude")
468
            async for msg in client.receive_response():
469
                print(msg)
470
    except CLINotFoundError:
471
        print("Please install Claude Code")
472
    except CLIConnectionError as e:
473
        print(f"Connection error: {e}")
474
    except ClaudeSDKError as e:
475
        print(f"SDK error: {e}")
476

477
anyio.run(main)
478
```
479