0
# Core Query Interface
1

2
The core query interface provides two interaction patterns for communicating with Claude: a simple `query()` function for stateless interactions and a `ClaudeSDKClient` class for bidirectional streaming with state management.
3

4
## Capabilities
5

6
### Simple Query Function
7

8
The `query()` function provides the simplest way to interact with Claude for one-shot or unidirectional streaming queries.
9

10
```python { .api }
11
async def query(
12
    *,
13
    prompt: str | AsyncIterable[dict[str, Any]],
14
    options: ClaudeAgentOptions | None = None,
15
    transport: Transport | None = None,
16
) -> AsyncIterator[Message]:
17
    """
18
    One-shot or unidirectional streaming query to Claude.
19

20
    Args:
21
        prompt: User message string or async iterable of message dicts
22
        options: Configuration options (default: ClaudeAgentOptions())
23
        transport: Custom transport implementation (default: subprocess CLI)
24

25
    Returns:
26
        AsyncIterator yielding Message objects
27

28
    Raises:
29
        CLINotFoundError: When Claude Code CLI is not found
30
        CLIConnectionError: When unable to connect to CLI
31
        ProcessError: When CLI process fails
32
    """
33
```
34

35
**Parameters:**
36

37
- `prompt` (str | AsyncIterable[dict[str, Any]]): The prompt to send to Claude. Can be a string for single-shot queries or an AsyncIterable[dict] for streaming mode with continuous interaction. In streaming mode, each dict should have the structure:
38
  ```python
39
  {
40
      "type": "user",
41
      "message": {"role": "user", "content": "..."},
42
      "parent_tool_use_id": None,
43
      "session_id": "..."
44
  }
45
  ```
46

47
- `options` (ClaudeAgentOptions | None): Configuration options. If None, defaults to `ClaudeAgentOptions()`. Use this to configure tools, permissions, working directory, models, and more.
48

49
- `transport` (Transport | None): Custom transport implementation. If provided, this will be used instead of the default subprocess CLI transport. The transport will be automatically configured with the prompt and options.
50

51
**Returns:**
52

53
AsyncIterator[Message] that yields messages from the conversation, including:
54
- `UserMessage`: Messages from the user
55
- `AssistantMessage`: Messages from Claude
56
- `SystemMessage`: System-level messages
57
- `ResultMessage`: Final result with cost and usage information
58
- `StreamEvent`: Raw stream events from the API
59

60
**When to use query():**
61
- Simple one-off questions ("What is 2+2?")
62
- Batch processing of independent prompts
63
- Code generation or analysis tasks
64
- Automated scripts and CI/CD pipelines
65
- When you know all inputs upfront
66
- Stateless operations
67

68
**When to use ClaudeSDKClient instead:**
69
- Interactive conversations with follow-ups
70
- Chat applications or REPL-like interfaces
71
- When you need to send messages based on responses
72
- When you need interrupt capabilities
73
- Long-running sessions with state
74

75
**Usage Example - Simple Query:**
76

77
```python
78
import anyio
79
from claude_agent_sdk import query, AssistantMessage, TextBlock
80

81
async def main():
82
    # Simple question
83
    async for message in query(prompt="What is 2 + 2?"):
84
        if isinstance(message, AssistantMessage):
85
            for block in message.content:
86
                if isinstance(block, TextBlock):
87
                    print(block.text)
88

89
anyio.run(main)
90
```
91

92
**Usage Example - With Options:**
93

94
```python
95
from claude_agent_sdk import query, ClaudeAgentOptions
96

97
async def main():
98
    options = ClaudeAgentOptions(
99
        allowed_tools=["Read", "Write", "Bash"],
100
        permission_mode="acceptEdits",
101
        cwd="/path/to/project",
102
        system_prompt="You are an expert Python developer"
103
    )
104

105
    async for message in query(
106
        prompt="Create a hello.py file that prints Hello World",
107
        options=options
108
    ):
109
        print(message)
110

111
anyio.run(main)
112
```
113

114
**Usage Example - Streaming Mode:**
115

116
```python
117
async def prompts():
118
    yield {"type": "user", "message": {"role": "user", "content": "Hello"}}
119
    yield {"type": "user", "message": {"role": "user", "content": "How are you?"}}
120

121
# All prompts are sent, then all responses received
122
async for message in query(prompt=prompts()):
123
    print(message)
124
```
125

126
### Interactive Client
127

128
The `ClaudeSDKClient` class provides bidirectional streaming for interactive conversations with full state management.
129

130
```python { .api }
131
class ClaudeSDKClient:
132
    """
133
    Bidirectional streaming client for interactive conversations.
134

135
    Key features:
136
    - Bidirectional: Send and receive messages at any time
137
    - Stateful: Maintains conversation context across messages
138
    - Interactive: Send follow-ups based on responses
139
    - Control flow: Support for interrupts and session management
140
    """
141

142
    def __init__(
143
        self,
144
        options: ClaudeAgentOptions | None = None,
145
        transport: Transport | None = None,
146
    ):
147
        """
148
        Initialize client.
149

150
        Args:
151
            options: Configuration options (default: ClaudeAgentOptions())
152
            transport: Custom transport implementation
153
        """
154

155
    async def connect(
156
        self, prompt: str | AsyncIterable[dict[str, Any]] | None = None
157
    ) -> None:
158
        """
159
        Establish connection and optionally send initial prompt.
160

161
        Args:
162
            prompt: Initial prompt string, message stream, or None for empty connection
163

164
        Raises:
165
            CLIConnectionError: When unable to connect
166
            ValueError: When can_use_tool is used incorrectly
167
        """
168

169
    async def query(
170
        self, prompt: str | AsyncIterable[dict[str, Any]], session_id: str = "default"
171
    ) -> None:
172
        """
173
        Send a message to Claude in streaming mode.
174

175
        Args:
176
            prompt: String message or async iterable of message dicts
177
            session_id: Session identifier for the conversation
178

179
        Raises:
180
            CLIConnectionError: When not connected
181
        """
182

183
    async def receive_messages(self) -> AsyncIterator[Message]:
184
        """
185
        Receive all messages until connection closes.
186

187
        Yields:
188
            Message: Each message received from Claude
189

190
        Raises:
191
            CLIConnectionError: When not connected
192
        """
193

194
    async def receive_response(self) -> AsyncIterator[Message]:
195
        """
196
        Receive messages until ResultMessage is received.
197

198
        This is a convenience method that automatically terminates after
199
        receiving a ResultMessage, which indicates the response is complete.
200

201
        Yields:
202
            Message: Each message including the final ResultMessage
203

204
        Raises:
205
            CLIConnectionError: When not connected
206
        """
207

208
    async def interrupt(self) -> None:
209
        """
210
        Send interrupt signal to stop current operation.
211

212
        Only works in streaming mode.
213

214
        Raises:
215
            CLIConnectionError: When not connected
216
        """
217

218
    async def set_permission_mode(self, mode: str) -> None:
219
        """
220
        Change permission mode during conversation.
221

222
        Args:
223
            mode: Permission mode ("default", "acceptEdits", "plan", "bypassPermissions")
224

225
        Raises:
226
            CLIConnectionError: When not connected
227
        """
228

229
    async def set_model(self, model: str | None = None) -> None:
230
        """
231
        Switch AI model during conversation.
232

233
        Args:
234
            model: Model identifier or None for default
235

236
        Raises:
237
            CLIConnectionError: When not connected
238
        """
239

240
    async def get_server_info(self) -> dict[str, Any] | None:
241
        """
242
        Get server capabilities and information.
243

244
        Returns:
245
            Server info dict with commands and capabilities, or None
246

247
        Raises:
248
            CLIConnectionError: When not connected
249
        """
250

251
    async def disconnect(self) -> None:
252
        """
253
        Close connection and clean up resources.
254
        """
255

256
    async def __aenter__(self) -> ClaudeSDKClient:
257
        """Context manager entry - automatically connects."""
258

259
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> bool:
260
        """Context manager exit - automatically disconnects."""
261
```
262

263
**Usage Example - Basic Interactive Session:**
264

265
```python
266
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
267

268
async def main():
269
    options = ClaudeAgentOptions(
270
        allowed_tools=["Read", "Write"],
271
        permission_mode="acceptEdits"
272
    )
273

274
    async with ClaudeSDKClient(options=options) as client:
275
        # First query
276
        await client.query("Create a todo.txt file")
277
        async for msg in client.receive_response():
278
            print(msg)
279

280
        # Follow-up query in same session
281
        await client.query("Add 'Buy groceries' to the file")
282
        async for msg in client.receive_response():
283
            print(msg)
284

285
anyio.run(main)
286
```
287

288
**Usage Example - Dynamic Permission Changes:**
289

290
```python
291
async with ClaudeSDKClient() as client:
292
    # Start with default permissions
293
    await client.query("Help me analyze this codebase")
294
    async for msg in client.receive_response():
295
        print(msg)
296

297
    # Switch to auto-accept edits
298
    await client.set_permission_mode('acceptEdits')
299
    await client.query("Now implement the fix we discussed")
300
    async for msg in client.receive_response():
301
        print(msg)
302
```
303

304
**Usage Example - Interrupt Support:**
305

306
```python
307
async with ClaudeSDKClient() as client:
308
    await client.query("Generate 1000 test files")
309

310
    # Receive some messages
311
    count = 0
312
    async for msg in client.receive_messages():
313
        print(msg)
314
        count += 1
315
        if count > 5:
316
            # Stop the operation
317
            await client.interrupt()
318
            break
319
```
320

321
**Usage Example - Model Switching:**
322

323
```python
324
async with ClaudeSDKClient() as client:
325
    # Start with default model
326
    await client.query("Help me understand this problem")
327
    async for msg in client.receive_response():
328
        print(msg)
329

330
    # Switch to a different model
331
    await client.set_model('claude-sonnet-4-5')
332
    await client.query("Now implement the solution")
333
    async for msg in client.receive_response():
334
        print(msg)
335
```
336

337
**Important Notes:**
338

339
- 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)
340
- The client internally maintains a persistent anyio task group that remains active from `connect()` until `disconnect()`
341
- Complete all operations within the same async context where the client was connected
342
- Always use the context manager (`async with`) or manually call `disconnect()` to clean up resources
343

344
### Transport Interface
345

346
The `Transport` class provides an abstract interface for custom transport implementations.
347

348
```python { .api }
349
class Transport:
350
    """
351
    Abstract transport interface for custom implementations.
352

353
    Note: This is an internal API that may change in future releases.
354
    """
355

356
    async def connect(self) -> None:
357
        """Establish connection to Claude Code."""
358

359
    async def write(self, data: str) -> None:
360
        """Write data to the transport."""
361

362
    def read_messages(self) -> AsyncIterator[dict[str, Any]]:
363
        """Read messages stream from the transport."""
364

365
    async def close(self) -> None:
366
        """Close connection and clean up resources."""
367

368
    def is_ready(self) -> bool:
369
        """Check if transport is ready for communication."""
370

371
    async def end_input(self) -> None:
372
        """Signal end of input stream."""
373
```
374

375
**Usage Example - Custom Transport:**
376

377
```python
378
from claude_agent_sdk import Transport, query
379

380
class MyCustomTransport(Transport):
381
    async def connect(self) -> None:
382
        # Custom connection logic
383
        pass
384

385
    async def write(self, data: str) -> None:
386
        # Custom write logic
387
        pass
388

389
    def read_messages(self) -> AsyncIterator[dict[str, Any]]:
390
        # Custom read logic
391
        pass
392

393
    async def close(self) -> None:
394
        # Custom cleanup logic
395
        pass
396

397
    def is_ready(self) -> bool:
398
        return True
399

400
    async def end_input(self) -> None:
401
        pass
402

403
# Use custom transport
404
transport = MyCustomTransport()
405
async for message in query(prompt="Hello", transport=transport):
406
    print(message)
407
```
408

409
## Comparison: query() vs ClaudeSDKClient
410

411
| Feature | query() | ClaudeSDKClient |
412
|---------|---------|----------------|
413
| Communication | Unidirectional | Bidirectional |
414
| State | Stateless | Stateful |
415
| Use case | One-shot queries | Multi-turn conversations |
416
| Interrupts | No | Yes |
417
| Dynamic config | No | Yes (permissions, model) |
418
| Follow-ups | No | Yes |
419
| Complexity | Simple | Full-featured |
420
| Context manager | No | Yes |
421
422
Choose `query()` for simple, stateless interactions. Choose `ClaudeSDKClient` for interactive, stateful conversations with full control flow.
423