Claude Agent SDK

from claude_agent_sdk import query, ClaudeSDKClient, ClaudeAgentOptions

from claude_agent_sdk import (
    # Main functions
    query,
    __version__,

    # Client
    ClaudeSDKClient,
    Transport,

    # Configuration
    ClaudeAgentOptions,

    # Message types
    UserMessage,
    AssistantMessage,
    SystemMessage,
    ResultMessage,
    StreamEvent,
    Message,

    # Content blocks
    TextBlock,
    ThinkingBlock,
    ToolUseBlock,
    ToolResultBlock,
    ContentBlock,

    # MCP server support
    create_sdk_mcp_server,
    tool,
    SdkMcpTool,
    McpServerConfig,
    McpSdkServerConfig,

    # Hook system
    HookCallback,
    HookContext,
    HookInput,
    HookJSONOutput,
    HookMatcher,

    # Permission system
    PermissionMode,
    CanUseTool,
    ToolPermissionContext,
    PermissionResult,
    PermissionResultAllow,
    PermissionResultDeny,
    PermissionUpdate,

    # Agent system
    AgentDefinition,
    SettingSource,

    # Plugin support
    SdkPluginConfig,

    # Error types
    ClaudeSDKError,
    CLIConnectionError,
    CLINotFoundError,
    ProcessError,
    CLIJSONDecodeError,
)

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",  # Auto-accept file edits
        cwd="/path/to/project"
    )

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

anyio.run(main)

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

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

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

        Args:
            options: Configuration options
            transport: Custom transport implementation
        """

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

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

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

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

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

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

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

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

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

@dataclass
class UserMessage:
    """Message from user to Claude."""
    content: str | list[ContentBlock]
    parent_tool_use_id: str | None = None

@dataclass
class AssistantMessage:
    """Message from Claude to user."""
    content: list[ContentBlock]
    model: str
    parent_tool_use_id: str | None = None

@dataclass
class SystemMessage:
    """System-level message."""
    subtype: str
    data: dict[str, Any]

@dataclass
class ResultMessage:
    """Final result of conversation turn."""
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None
    structured_output: Any = None

@dataclass
class StreamEvent:
    """Raw stream event from Anthropic API."""
    uuid: str
    session_id: str
    event: dict[str, Any]
    parent_tool_use_id: str | None = None

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent

@dataclass
class TextBlock:
    """Plain text content."""
    text: str

@dataclass
class ThinkingBlock:
    """Extended thinking content."""
    thinking: str
    signature: str

@dataclass
class ToolUseBlock:
    """Tool invocation request."""
    id: str
    name: str
    input: dict[str, Any]

@dataclass
class ToolResultBlock:
    """Tool execution result."""
    tool_use_id: str
    content: str | list[dict[str, Any]] | None = None
    is_error: bool | None = None

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

@dataclass
class ClaudeAgentOptions:
    """Configuration options for Claude SDK."""

    # Tool configuration
    allowed_tools: list[str] = field(default_factory=list)
    disallowed_tools: list[str] = field(default_factory=list)

    # Prompt configuration
    system_prompt: str | SystemPromptPreset | None = None

    # MCP server configuration
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)

    # Permission configuration
    permission_mode: PermissionMode | None = None
    permission_prompt_tool_name: str | None = None
    can_use_tool: CanUseTool | None = None

    # Session configuration
    continue_conversation: bool = False
    resume: str | None = None
    fork_session: bool = False

    # Budget and limits
    max_turns: int | None = None
    max_budget_usd: float | None = None
    max_thinking_tokens: int | None = None

    # Model configuration
    model: str | None = None
    fallback_model: str | None = None

    # Working directory and CLI
    cwd: str | Path | None = None
    cli_path: str | Path | None = None
    add_dirs: list[str | Path] = field(default_factory=list)

    # Settings and sources
    settings: str | None = None
    setting_sources: list[SettingSource] | None = None

    # Environment
    env: dict[str, str] = field(default_factory=dict)
    extra_args: dict[str, str | None] = field(default_factory=dict)

    # Callbacks
    stderr: Callable[[str], None] | None = None
    debug_stderr: Any = sys.stderr  # Deprecated: Use stderr callback instead

    # Hooks and plugins
    hooks: dict[HookEvent, list[HookMatcher]] | None = None
    agents: dict[str, AgentDefinition] | None = None
    plugins: list[SdkPluginConfig] = field(default_factory=list)

    # Advanced features
    include_partial_messages: bool = False
    output_format: dict[str, Any] | None = None
    max_buffer_size: int | None = None
    user: str | None = None

def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any]
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]:
    """
    Decorator for creating custom tools.

    Args:
        name: Tool name
        description: Tool description for Claude
        input_schema: Input schema as type or dict

    Returns:
        Decorator that wraps handler function
    """

def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig:
    """
    Create an in-process MCP server with custom tools.

    Args:
        name: Server name
        version: Server version
        tools: List of tool definitions

    Returns:
        Server configuration for use in ClaudeAgentOptions.mcp_servers
    """

@dataclass
class SdkMcpTool(Generic[T]):
    """Custom tool definition."""
    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[[T], Awaitable[dict[str, Any]]]

HookEvent = Literal[
    "PreToolUse",
    "PostToolUse",
    "UserPromptSubmit",
    "Stop",
    "SubagentStop",
    "PreCompact"
]

@dataclass
class HookMatcher:
    """Hook configuration with pattern matching."""
    matcher: str | None = None  # Tool name pattern (e.g., "Bash", "Write|Edit")
    hooks: list[HookCallback] = field(default_factory=list)

HookCallback = Callable[
    [HookInput, str | None, HookContext],
    Awaitable[HookJSONOutput],
]

class HookContext(TypedDict):
    """Context provided to hooks."""
    signal: Any | None

PermissionMode = Literal["default", "acceptEdits", "plan", "bypassPermissions"]

CanUseTool = Callable[
    [str, dict[str, Any], ToolPermissionContext],
    Awaitable[PermissionResult]
]

@dataclass
class ToolPermissionContext:
    """Context for permission callbacks."""
    signal: Any | None = None
    suggestions: list[PermissionUpdate] = field(default_factory=list)

@dataclass
class PermissionResultAllow:
    """Allow permission decision."""
    behavior: Literal["allow"] = "allow"
    updated_input: dict[str, Any] | None = None
    updated_permissions: list[PermissionUpdate] | None = None

@dataclass
class PermissionResultDeny:
    """Deny permission decision."""
    behavior: Literal["deny"] = "deny"
    message: str = ""
    interrupt: bool = False

PermissionResult = PermissionResultAllow | PermissionResultDeny

@dataclass
class PermissionUpdate:
    """Permission configuration update."""
    type: Literal[
        "addRules", "replaceRules", "removeRules",
        "setMode", "addDirectories", "removeDirectories"
    ]
    rules: list[PermissionRuleValue] | None = None
    behavior: PermissionBehavior | None = None
    mode: PermissionMode | None = None
    directories: list[str] | None = None
    destination: PermissionUpdateDestination | None = None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for CLI."""

@dataclass
class AgentDefinition:
    """Custom agent configuration."""
    description: str
    prompt: str
    tools: list[str] | None = None
    model: Literal["sonnet", "opus", "haiku", "inherit"] | None = None

SettingSource = Literal["user", "project", "local"]

class McpStdioServerConfig(TypedDict):
    """Subprocess MCP server via stdio."""
    type: NotRequired[Literal["stdio"]]
    command: str
    args: NotRequired[list[str]]
    env: NotRequired[dict[str, str]]

class McpSSEServerConfig(TypedDict):
    """MCP server via Server-Sent Events."""
    type: Literal["sse"]
    url: str
    headers: NotRequired[dict[str, str]]

class McpHttpServerConfig(TypedDict):
    """MCP server via HTTP."""
    type: Literal["http"]
    url: str
    headers: NotRequired[dict[str, str]]

class McpSdkServerConfig(TypedDict):
    """In-process SDK MCP server."""
    type: Literal["sdk"]
    name: str
    instance: McpServer

McpServerConfig = (
    McpStdioServerConfig | McpSSEServerConfig |
    McpHttpServerConfig | McpSdkServerConfig
)

class ClaudeSDKError(Exception):
    """Base exception for all Claude SDK errors."""

class CLIConnectionError(ClaudeSDKError):
    """Raised when unable to connect to Claude Code."""

class CLINotFoundError(CLIConnectionError):
    """Raised when Claude Code is not found or not installed."""

    def __init__(
        self,
        message: str = "Claude Code not found",
        cli_path: str | None = None
    ):
        ...

    cli_path: str | None

class ProcessError(ClaudeSDKError):
    """Raised when the CLI process fails."""

    def __init__(
        self,
        message: str,
        exit_code: int | None = None,
        stderr: str | None = None
    ):
        ...

    exit_code: int | None
    stderr: str | None

class CLIJSONDecodeError(ClaudeSDKError):
    """Raised when unable to decode JSON from CLI output."""

    def __init__(self, line: str, original_error: Exception):
        ...

    line: str
    original_error: Exception

__version__: str  # Current SDK version (e.g., "0.1.8")

class SystemPromptPreset(TypedDict):
    """System prompt preset configuration."""
    type: Literal["preset"]
    preset: Literal["claude_code"]
    append: NotRequired[str]

# Use Claude Code's default system prompt
options = ClaudeAgentOptions(
    system_prompt={"type": "preset", "preset": "claude_code"}
)

# Append additional instructions
options = ClaudeAgentOptions(
    system_prompt={
        "type": "preset",
        "preset": "claude_code",
        "append": "Always explain your reasoning step by step."
    }
)

# Use custom system prompt
options = ClaudeAgentOptions(
    system_prompt="You are a helpful Python expert specializing in async programming."
)

class SdkPluginConfig(TypedDict):
    """Plugin configuration."""
    type: Literal["local"]
    path: str

options = ClaudeAgentOptions(
    plugins=[
        {"type": "local", "path": "/path/to/plugin"}
    ]
)