Interactive Client

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

    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: ClaudeCodeOptions | None = None):
        """
        Initialize Claude SDK client.

        Args:
            options: Optional configuration (defaults to ClaudeCodeOptions() if None)
        """

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. Can be string, async iterable of messages, or None
                for empty connection that stays open for interactive use
    """

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

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

    Yields:
        Message: All messages in the conversation stream
    """

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

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).

    Yields:
        Message: Each message received (UserMessage, AssistantMessage, SystemMessage, ResultMessage)
    """

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

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

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

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

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

from claude_code_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def main():
    async with ClaudeSDKClient() as client:
        # Send initial query
        await client.query("Hello, how are you?")

        # Receive and process response
        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}")

        # Send follow-up based on response
        await client.query("Can you help me write a Python function?")

        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)

from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def main():
    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits",
        system_prompt="You are a helpful coding assistant",
        cwd="/path/to/project"
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Create a Python web server")

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

anyio.run(main)

from claude_code_sdk import (
    ClaudeSDKClient, ClaudeCodeOptions,
    tool, create_sdk_mcp_server
)

@tool("greet", "Greet a user", {"name": str})
async def greet_user(args):
    return {
        "content": [
            {"type": "text", "text": f"Hello, {args['name']}!"}
        ]
    }

async def main():
    # Create SDK MCP server
    server = create_sdk_mcp_server(
        name="my-tools",
        version="1.0.0",
        tools=[greet_user]
    )

    options = ClaudeCodeOptions(
        mcp_servers={"tools": server},
        allowed_tools=["mcp__tools__greet"]
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Greet Alice")

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

anyio.run(main)

from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions, HookMatcher

async def check_bash_command(input_data, tool_use_id, context):
    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]

    if tool_name != "Bash":
        return {}

    command = tool_input.get("command", "")
    forbidden_patterns = ["rm -rf", "format"]

    for pattern in forbidden_patterns:
        if pattern in command:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"Command contains dangerous pattern: {pattern}",
                }
            }
    return {}

async def main():
    options = ClaudeCodeOptions(
        allowed_tools=["Bash"],
        hooks={
            "PreToolUse": [
                HookMatcher(matcher="Bash", hooks=[check_bash_command]),
            ],
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Run the bash command: echo 'Hello World!'")

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

anyio.run(main)

import asyncio
from claude_code_sdk import ClaudeSDKClient

async def main():
    async with ClaudeSDKClient() as client:
        await client.query("Write a very long story about space exploration")

        # Start receiving messages in background
        async def receive_messages():
            async for msg in client.receive_messages():
                print(msg)

        receive_task = asyncio.create_task(receive_messages())

        # Wait a bit, then interrupt
        await asyncio.sleep(5)
        await client.interrupt()

        await receive_task

anyio.run(main)

from claude_code_sdk import ClaudeSDKClient

async def main():
    async with ClaudeSDKClient() as client:
        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')}")

anyio.run(main)

from claude_code_sdk import ClaudeSDKClient

async def main():
    client = ClaudeSDKClient()

    try:
        await client.connect("Hello Claude")

        async for msg in client.receive_messages():
            print(msg)
            # Break after first complete response
            if hasattr(msg, 'subtype') and msg.subtype == "result":
                break

    finally:
        await client.disconnect()

anyio.run(main)