Interactive Client

class ClaudeSDKClient:
    """
    Client for bidirectional, interactive conversations with Claude Code.

    This client 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

    Caveat: 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().
    This means you must complete all operations with the client within the same
    async context where it was connected.
    """

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

        Args:
            options: Optional configuration (ClaudeAgentOptions instance)
            transport: Optional custom transport implementation
        """

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

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

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

        Yields:
            Message objects (UserMessage, AssistantMessage, SystemMessage,
            ResultMessage, or StreamEvent)
        """

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

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

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

        Interrupts the current Claude operation, stopping generation and
        tool execution.
        """

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

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

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

        Args:
            model: The model to use, or None to use default. Examples:
                - 'claude-sonnet-4-5'
                - 'claude-opus-4-1-20250805'
                - 'claude-opus-4-20250514'
        """

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

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

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

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

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

        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

        Yields:
            Message objects (each message received until ResultMessage)

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

    async def disconnect(self) -> None:
        """
        Disconnect from Claude.

        Closes the connection and cleans up resources.
        """

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

        Returns:
            Self (the ClaudeSDKClient instance)
        """

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

        Returns:
            False (does not suppress exceptions)
        """

import anyio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

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

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

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage

async def main():
    async with ClaudeSDKClient() as client:
        # First question
        await client.query("What is Python?")
        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}")

        # Follow-up question
        await client.query("Can you show me a simple Python example?")
        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}")

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful Python expert",
        allowed_tools=["Read", "Bash"],
        permission_mode='default',
        cwd="/home/user/project"
    )

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

        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage

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

        # Interrupt after a short delay
        async def interrupt_after_delay():
            await anyio.sleep(2)
            await client.interrupt()

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

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

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

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

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

        # Implementation mode - 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)

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    async with ClaudeSDKClient() as client:
        # Start with default model
        await client.query("Explain quantum computing briefly")
        async for msg in client.receive_response():
            print(msg)

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

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient

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

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

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

    client = ClaudeSDKClient(options=options)

    try:
        # Manual connect
        await client.connect()

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

    finally:
        # Manual disconnect
        await client.disconnect()

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, ResultMessage

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

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

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

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient

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

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

        async for msg in client.receive_messages():
            print(msg)

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, Transport

class MyCustomTransport(Transport):
    # Implement custom transport logic
    async def connect(self) -> None:
        pass

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

    def read_messages(self):
        pass

    async def close(self) -> None:
        pass

    def is_ready(self) -> bool:
        return True

    async def end_input(self) -> None:
        pass

async def main():
    transport = MyCustomTransport()
    client = ClaudeSDKClient(transport=transport)

    async with client:
        await client.query("Hello")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeSDKError,
    CLINotFoundError,
    CLIConnectionError
)

async def main():
    try:
        async with ClaudeSDKClient() as client:
            await client.query("Hello Claude")
            async for msg in client.receive_response():
                print(msg)
    except CLINotFoundError:
        print("Please install Claude Code")
    except CLIConnectionError as e:
        print(f"Connection error: {e}")
    except ClaudeSDKError as e:
        print(f"SDK error: {e}")

anyio.run(main)