Interactive Client

class ClaudeSDKClient:
    """Interactive client for bidirectional conversations with Claude Code."""
    
    def __init__(
        self,
        options: ClaudeAgentOptions | None = None,
        transport: Transport | None = None,
    ):
        """Initialize client.
        
        Args:
            options: Optional ClaudeAgentOptions configuration. None uses defaults
            transport: Optional custom Transport implementation. None uses bundled CLI
        """

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    client = ClaudeSDKClient()
    # Use client...

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

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

    client = ClaudeSDKClient(options=options)
    # Use client...

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # Client is automatically connected and disconnected
        await client.query("Hello")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

async def connect(
    self,
    prompt: str | AsyncIterable[dict[str, Any]] | None = None
) -> None:
    """Establish connection to Claude Code.
    
    Args:
        prompt: Optional initial prompt to send immediately after connecting.
                Can be string or async iterable of message dictionaries
                
    Raises:
        CLINotFoundError: Claude Code CLI not found or not installed
        CLIConnectionError: Unable to connect to Claude Code
        ProcessError: CLI process failed to start
    """

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    client = ClaudeSDKClient()
    await client.connect()
    # Now ready to send messages
    
anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    client = ClaudeSDKClient()
    await client.connect(prompt="What files are in the current directory?")
    async for msg in client.receive_response():
        print(msg)
    
anyio.run(main)

async def disconnect(self) -> None:
    """Close connection and cleanup resources.
    
    Closes the transport connection and performs cleanup. Safe to call multiple times.
    """

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    client = ClaudeSDKClient()
    await client.connect()
    # ... use client ...
    await client.disconnect()
    
anyio.run(main)

async def query(
    self,
    prompt: str | AsyncIterable[dict[str, Any]],
    session_id: str = "default"
) -> None:
    """Send message in interactive mode.
    
    Args:
        prompt: String message or async iterable of message dictionaries
        session_id: Session identifier for conversation tracking (default: "default")
        
    Note: Call receive_messages() or receive_response() to get Claude's responses
    """

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    async with ClaudeSDKClient() as client:
        await client.query("What is the capital of France?")
        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:
        # First message
        await client.query("List the Python files")
        async for msg in client.receive_response():
            print(msg)
        
        # Follow-up (has context from first message)
        await client.query("Tell me more about main.py")
        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:
        # Session 1
        await client.query("Question for session 1", session_id="session1")
        async for msg in client.receive_response():
            print(f"[Session 1] {msg}")
        
        # Session 2 (independent conversation)
        await client.query("Question for session 2", session_id="session2")
        async for msg in client.receive_response():
            print(f"[Session 2] {msg}")

anyio.run(main)

async def receive_messages(self) -> AsyncIterator[Message]:
    """Receive all messages continuously.
    
    Yields messages until connection is closed. Use for long-running sessions
    or when you need to process messages as they arrive without stopping.
    
    Yields:
        Message: UserMessage, AssistantMessage, SystemMessage, ResultMessage, or StreamEvent
    """

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    async with ClaudeSDKClient() as client:
        await client.query("Start working")
        
        # Process all messages until disconnected
        async for message in client.receive_messages():
            print(message)

anyio.run(main)

async def receive_response(self) -> AsyncIterator[Message]:
    """Receive messages until ResultMessage, then stop.
    
    Yields messages for current query until ResultMessage is encountered,
    then stops. Use this for typical request-response patterns.
    
    Yields:
        Message: Messages for current query, ending with ResultMessage
    """

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

async def main():
    async with ClaudeSDKClient() as client:
        await client.query("Create a hello.py file")
        
        # Process messages until result
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
            elif isinstance(msg, ResultMessage):
                print(f"Completed in {msg.duration_ms}ms")

anyio.run(main)

async def interrupt(self) -> dict[str, Any]:
    """Send interrupt signal to stop current operation.
    
    Stops Claude from continuing the current task. Useful for long-running
    operations that need to be cancelled.
    
    Returns:
        dict[str, Any]: Response dictionary with interrupt acknowledgment
    """

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    async with ClaudeSDKClient() as client:
        # Start a long-running query
        await client.query("Analyze all files in this large repository")
        
        # Interrupt if needed (e.g., based on user input or timeout)
        result = await client.interrupt()
        print(f"Interrupted: {result}")

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    async with ClaudeSDKClient() as client:
        await client.query("Long running task")
        
        try:
            with anyio.fail_after(30):  # 30 second timeout
                async for msg in client.receive_response():
                    print(msg)
        except TimeoutError:
            print("Operation timed out, interrupting...")
            await client.interrupt()

anyio.run(main)

async def set_permission_mode(self, mode: PermissionMode) -> dict[str, Any]:
    """Change permission mode mid-conversation.
    
    Args:
        mode: New permission mode. One of:
              - "default": CLI prompts user for dangerous tools
              - "acceptEdits": Auto-accept file edit operations
              - "plan": Plan mode for design/planning workflows
              - "bypassPermissions": Allow all tools (use with caution!)
    
    Returns:
        dict[str, Any]: Response dictionary confirming mode change
    """

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    # Start with default permissions
    async with ClaudeSDKClient() as client:
        # Review code (read-only operations)
        await client.query("Review this code")
        async for msg in client.receive_response():
            print(msg)

        # Switch to auto-accept edits for implementation
        await client.set_permission_mode("acceptEdits")

        # Now implement fixes (will auto-approve edits)
        await client.query("Now fix the issues you found")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        permission_mode="default"  # Start conservative
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # Phase 1: Analysis (default permissions)
        await client.query("Analyze security issues")
        async for msg in client.receive_response():
            print(msg)
        
        # Phase 2: Planning (plan mode)
        await client.set_permission_mode("plan")
        await client.query("Create a fix plan")
        async for msg in client.receive_response():
            print(msg)
        
        # Phase 3: Implementation (accept edits)
        await client.set_permission_mode("acceptEdits")
        await client.query("Implement the fixes")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

async def set_model(self, model: str) -> dict[str, Any]:
    """Switch AI model during conversation.
    
    Args:
        model: Model identifier. Can be short name ("sonnet", "opus", "haiku")
               or full model ID (e.g., "claude-sonnet-4-5")
    
    Returns:
        dict[str, Any]: Response dictionary confirming model change
    """

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    async with ClaudeSDKClient() as client:
        # Start with default model (usually sonnet)
        await client.query("Simple question")
        async for msg in client.receive_response():
            print(msg)

        # Switch to more powerful model for complex task
        await client.set_model("opus")

        await client.query("Complex reasoning task requiring deep analysis")
        async for msg in client.receive_response():
            print(msg)
        
        # Switch to faster/cheaper model for simple tasks
        await client.set_model("haiku")
        
        await client.query("Quick summary")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

async def rewind_files(self, user_message_id: str) -> dict[str, Any]:
    """Rewind tracked files to previous state.
    
    Args:
        user_message_id: UUID of user message to rewind to. Get from UserMessage.uuid
        
    Returns:
        dict[str, Any]: Response dictionary with rewind results
        
    Requires: enable_file_checkpointing=True in ClaudeAgentOptions
    
    Note: Only files modified during the session are tracked and can be rewound
    """

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    UserMessage
)

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

    async with ClaudeSDKClient(options=options) as client:
        # Make initial changes
        await client.query("Modify config.json")
        message_id = None
        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                message_id = msg.uuid

        # Make more changes
        await client.query("Make more changes to config.json")
        async for msg in client.receive_response():
            print(msg)

        # Rewind files to previous state
        if message_id:
            result = await client.rewind_files(message_id)
            print(f"Rewound files: {result}")
            
            # Files are now back to state after first query

anyio.run(main)

async def get_mcp_status(self) -> dict[str, Any]:
    """Query MCP server initialization status.
    
    Returns:
        dict[str, Any]: Status dictionary with server states showing which
                        MCP servers are initialized, failed, or in progress
    """

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        mcp_servers={
            "calculator": {
                "type": "stdio",
                "command": "calc-server"
            }
        }
    )
    
    async with ClaudeSDKClient(options=options) as client:
        status = await client.get_mcp_status()
        print(f"MCP servers: {status}")

anyio.run(main)

async def get_server_info(self) -> dict[str, Any] | None:
    """Get server initialization information.
    
    Returns:
        dict[str, Any] | None: Server info dictionary with capabilities and version,
                               or None if not available
    """

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"Server capabilities: {info}")

anyio.run(main)

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
    ResultMessage,
    ToolUseBlock
)

async def interactive_session():
    """Complete interactive session example."""
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful coding assistant",
        cwd="/path/to/project",
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="default"  # Start with confirmations
    )

    async with ClaudeSDKClient(options=options) as client:
        # First query
        print("\n=== Query 1: List files ===")
        await client.query("What files are in this directory?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
                    elif isinstance(block, ToolUseBlock):
                        print(f"[Tool: {block.name}]")
            elif isinstance(msg, ResultMessage):
                print(f"Duration: {msg.duration_ms}ms")

        # Follow-up query (has context)
        print("\n=== Query 2: Read file ===")
        await client.query("Read the README.md file")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

        # Change permissions for implementation
        print("\n=== Switching to acceptEdits mode ===")
        await client.set_permission_mode("acceptEdits")
        
        await client.query("Add a new feature to main.py")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

anyio.run(interactive_session)

import anyio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def main():
    async with ClaudeSDKClient() as client:
        # Turn 1
        await client.query("What programming languages are you familiar with?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
        
        # Turn 2 (has context from Turn 1)
        await client.query("Which one would you recommend for web development?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
        
        # Turn 3 (has context from Turn 1 and 2)
        await client.query("Can you show me a hello world example?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def chat_repl():
    """Interactive REPL-style chat."""
    async with ClaudeSDKClient() as client:
        print("Chat started. Type 'exit' to quit.")
        
        while True:
            # Get user input
            user_input = input("\nYou: ").strip()
            if user_input.lower() == 'exit':
                break
            
            # Send to Claude
            await client.query(user_input)
            
            print("Claude: ", end="")
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(block.text, end=" ")
            print()  # New line

anyio.run(chat_repl)

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def progressive_workflow():
    """Multi-phase workflow with permission escalation."""
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        permission_mode="default"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # Phase 1: Analysis (read-only effectively)
        print("Phase 1: Analysis")
        await client.query("Analyze the codebase for issues")
        async for msg in client.receive_response():
            print(msg)
        
        # Phase 2: Planning
        print("\nPhase 2: Planning")
        await client.set_permission_mode("plan")
        await client.query("Create a detailed plan to fix the issues")
        async for msg in client.receive_response():
            print(msg)
        
        # User review happens here...
        print("\n[User reviews plan]")
        
        # Phase 3: Implementation
        print("\nPhase 3: Implementation")
        await client.set_permission_mode("acceptEdits")
        await client.query("Implement the fixes according to the plan")
        async for msg in client.receive_response():
            print(msg)

anyio.run(progressive_workflow)

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    UserMessage,
    ResultMessage
)

async def main():
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits"
    )
    
    checkpoints = []
    
    async with ClaudeSDKClient(options=options) as client:
        # Checkpoint 1: Initial implementation
        await client.query("Implement feature A")
        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                checkpoints.append(("feature_a", msg.uuid))
        
        # Checkpoint 2: Add feature B
        await client.query("Add feature B")
        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                checkpoints.append(("feature_b", msg.uuid))
        
        # Try risky change
        await client.query("Refactor everything")
        error_occurred = False
        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage) and msg.is_error:
                error_occurred = True
        
        # Rewind on error
        if error_occurred and checkpoints:
            print("Error occurred! Rewinding to last checkpoint...")
            name, checkpoint_id = checkpoints[-1]
            await client.rewind_files(checkpoint_id)
            print(f"Rewound to checkpoint: {name}")

anyio.run(main)

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def process_session(client, session_id: str, task: str):
    """Process a single session."""
    await client.query(task, session_id=session_id)
    async for msg in client.receive_response():
        print(f"[{session_id}] {msg}")

async def main():
    """Run parallel sessions."""
    async with ClaudeSDKClient() as client:
        # Start multiple tasks in parallel sessions
        async with anyio.create_task_group() as tg:
            tg.start_soon(process_session, client, "docs", "Document the API")
            tg.start_soon(process_session, client, "tests", "Write unit tests")
            tg.start_soon(process_session, client, "refactor", "Refactor utils")

anyio.run(main)