Core Query Interface

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeAgentOptions | None = None,
    transport: Transport | None = None,
) -> AsyncIterator[Message]:
    """
    One-shot or unidirectional streaming query to Claude.

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

    Returns:
        AsyncIterator yielding Message objects

    Raises:
        CLINotFoundError: When Claude Code CLI is not found
        CLIConnectionError: When unable to connect to CLI
        ProcessError: When CLI process fails
    """

import anyio
from claude_agent_sdk import query, AssistantMessage, TextBlock

async def main():
    # Simple question
    async for message in query(prompt="What is 2 + 2?"):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)

from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits",
        cwd="/path/to/project",
        system_prompt="You are an expert Python developer"
    )

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

anyio.run(main)

async def prompts():
    yield {"type": "user", "message": {"role": "user", "content": "Hello"}}
    yield {"type": "user", "message": {"role": "user", "content": "How are you?"}}

# All prompts are sent, then all responses received
async for message in query(prompt=prompts()):
    print(message)

class ClaudeSDKClient:
    """
    Bidirectional streaming client for interactive conversations.

    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
    """

    def __init__(
        self,
        options: ClaudeAgentOptions | None = None,
        transport: Transport | None = None,
    ):
        """
        Initialize client.

        Args:
            options: Configuration options (default: ClaudeAgentOptions())
            transport: Custom transport implementation
        """

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

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

        Raises:
            CLIConnectionError: When unable to connect
            ValueError: When can_use_tool is used incorrectly
        """

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

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

        Raises:
            CLIConnectionError: When not connected
        """

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

        Yields:
            Message: Each message received from Claude

        Raises:
            CLIConnectionError: When not connected
        """

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

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

        Yields:
            Message: Each message including the final ResultMessage

        Raises:
            CLIConnectionError: When not connected
        """

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

        Only works in streaming mode.

        Raises:
            CLIConnectionError: When not connected
        """

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

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

        Raises:
            CLIConnectionError: When not connected
        """

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

        Args:
            model: Model identifier or None for default

        Raises:
            CLIConnectionError: When not connected
        """

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

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

        Raises:
            CLIConnectionError: When not connected
        """

    async def disconnect(self) -> None:
        """
        Close connection and clean up resources.
        """

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

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

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write"],
        permission_mode="acceptEdits"
    )

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

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

anyio.run(main)

async with ClaudeSDKClient() as client:
    # Start with default permissions
    await client.query("Help me analyze this codebase")
    async for msg in client.receive_response():
        print(msg)

    # Switch to auto-accept edits
    await client.set_permission_mode('acceptEdits')
    await client.query("Now implement the fix we discussed")
    async for msg in client.receive_response():
        print(msg)

async with ClaudeSDKClient() as client:
    await client.query("Generate 1000 test files")

    # Receive some messages
    count = 0
    async for msg in client.receive_messages():
        print(msg)
        count += 1
        if count > 5:
            # Stop the operation
            await client.interrupt()
            break

async with ClaudeSDKClient() as client:
    # Start with default model
    await client.query("Help me understand this problem")
    async for msg in client.receive_response():
        print(msg)

    # Switch to a different model
    await client.set_model('claude-sonnet-4-5')
    await client.query("Now implement the solution")
    async for msg in client.receive_response():
        print(msg)

class Transport:
    """
    Abstract transport interface for custom implementations.

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

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

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

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

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

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

    async def end_input(self) -> None:
        """Signal end of input stream."""

from claude_agent_sdk import Transport, query

class MyCustomTransport(Transport):
    async def connect(self) -> None:
        # Custom connection logic
        pass

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

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

    async def close(self) -> None:
        # Custom cleanup logic
        pass

    def is_ready(self) -> bool:
        return True

    async def end_input(self) -> None:
        pass

# Use custom transport
transport = MyCustomTransport()
async for message in query(prompt="Hello", transport=transport):
    print(message)