Functions

query()

The primary async function for interacting with Claude Code. Returns an async iterator that yields messages as they arrive.
async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeCodeOptions | None = None
) -> AsyncIterator[Message]

Parameters

ParameterTypeDescription
promptstr | AsyncIterable[dict]The input prompt as a string or async iterable for streaming mode
optionsClaudeCodeOptions | NoneOptional configuration object (defaults to ClaudeCodeOptions() if None)

Returns

Returns an AsyncIterator[Message] that yields messages from the conversation.

Example - Simple query

import asyncio
from claude_code_sdk import query

async def main():
    async for message in query(prompt="What is 2+2?"):
        print(message)

asyncio.run(main())

Example - With options


import asyncio
from claude_code_sdk import query, ClaudeCodeOptions

async def main():
    options = ClaudeCodeOptions(
        system_prompt="You are an expert Python developer",
        permission_mode='acceptEdits',
        cwd="/home/user/project"
    )

    async for message in query(
        prompt="Create a Python web server",
        options=options
    ):
        print(message)


asyncio.run(main())

tool()

Decorator for defining MCP tools with type safety.
def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any]
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]

Parameters

ParameterTypeDescription
namestrUnique identifier for the tool
descriptionstrHuman-readable description of what the tool does
input_schematype | dict[str, Any]Schema defining the tool’s input parameters (see below)

Input Schema Options

  1. Simple type mapping (recommended):
    {"text": str, "count": int, "enabled": bool}
    
  2. JSON Schema format (for complex validation):
    {
        "type": "object",
        "properties": {
            "text": {"type": "string"},
            "count": {"type": "integer", "minimum": 0}
        },
        "required": ["text"]
    }
    

Returns

A decorator function that wraps the tool implementation and returns an SdkMcpTool instance.

Example

from claude_code_sdk import tool
from typing import Any

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

create_sdk_mcp_server()

Create an in-process MCP server that runs within your Python application.
def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig

Parameters

ParameterTypeDefaultDescription
namestr-Unique identifier for the server
versionstr"1.0.0"Server version string
toolslist[SdkMcpTool[Any]] | NoneNoneList of tool functions created with @tool decorator

Returns

Returns an McpSdkServerConfig object that can be passed to ClaudeCodeOptions.mcp_servers.

Example

from claude_code_sdk import tool, create_sdk_mcp_server

@tool("add", "Add two numbers", {"a": float, "b": float})
async def add(args):
    return {
        "content": [{
            "type": "text",
            "text": f"Sum: {args['a'] + args['b']}"
        }]
    }

@tool("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply(args):
    return {
        "content": [{
            "type": "text",
            "text": f"Product: {args['a'] * args['b']}"
        }]
    }

calculator = create_sdk_mcp_server(
    name="calculator",
    version="2.0.0",
    tools=[add, multiply]  # Pass decorated functions
)

# Use with Claude
options = ClaudeCodeOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["mcp__calc__add", "mcp__calc__multiply"]
)

Classes

ClaudeSDKClient

Client for bidirectional, interactive conversations with Claude Code. Provides full control over conversation flow with support for streaming, interrupts, and dynamic message sending.
class ClaudeSDKClient:
    def __init__(self, options: ClaudeCodeOptions | None = None)
    async def connect(self, prompt: str | AsyncIterable[dict] | None = None) -> None
    async def query(self, prompt: str | AsyncIterable[dict], session_id: str = "default") -> None
    async def receive_messages(self) -> AsyncIterator[Message]
    async def receive_response(self) -> AsyncIterator[Message]
    async def interrupt(self) -> None
    async def disconnect(self) -> None

Methods

MethodDescription
__init__(options)Initialize the client with optional configuration
connect(prompt)Connect to Claude with an optional initial prompt or message stream
query(prompt, session_id)Send a new request in streaming mode
receive_messages()Receive all messages from Claude as an async iterator
receive_response()Receive messages until and including a ResultMessage
interrupt()Send interrupt signal (only works in streaming mode)
disconnect()Disconnect from Claude

Context Manager Support

The client can be used as an async context manager for automatic connection management:
async with ClaudeSDKClient() as client:
    await client.query("Hello Claude")
    async for message in client.receive_response():
        print(message)
Important: When iterating over messages, avoid using break to exit early as this can cause asyncio cleanup issues. Instead, let the iteration complete naturally or use flags to track when you’ve found what you need.

Example - Interactive conversation

import asyncio
from claude_code_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage

async def main():
    async with ClaudeSDKClient() as client:
        # Send initial message
        await client.query("Let's solve a math problem step by step")
        
        # Receive and process response
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Assistant: {block.text[:100]}...")
            elif isinstance(message, ResultMessage):
                print("Response complete")
        
        # Send follow-up based on response
        await client.query("What's 15% of 80?")

asyncio.run(main())

Types

SdkMcpTool

Definition for an SDK MCP tool created with the @tool decorator.
@dataclass
class SdkMcpTool(Generic[T]):
    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[[T], Awaitable[dict[str, Any]]]
PropertyTypeDescription
namestrUnique identifier for the tool
descriptionstrHuman-readable description
input_schematype[T] | dict[str, Any]Schema for input validation
handlerCallable[[T], Awaitable[dict[str, Any]]]Async function that handles tool execution

ClaudeCodeOptions

Configuration dataclass for Claude Code queries.
@dataclass
class ClaudeCodeOptions:
    allowed_tools: list[str] = field(default_factory=list)
    max_thinking_tokens: int = 8000
    system_prompt: str | None = None
    append_system_prompt: str | None = None
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    permission_mode: PermissionMode | None = None
    continue_conversation: bool = False
    resume: str | None = None
    max_turns: int | None = None
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    permission_prompt_tool_name: str | None = None
    cwd: str | Path | None = None
    settings: str | None = None
    add_dirs: list[str | Path] = field(default_factory=list)
    env: dict[str, str] = field(default_factory=dict)
    extra_args: dict[str, str | None] = field(default_factory=dict)
PropertyTypeDefaultDescription
allowed_toolslist[str][]List of allowed tool names
max_thinking_tokensint8000Maximum tokens for thinking process
system_promptstr | NoneNoneReplace the default system prompt entirely
append_system_promptstr | NoneNoneText to append to the default system prompt
mcp_serversdict[str, McpServerConfig] | str | Path{}MCP server configurations or path to config file
permission_modePermissionMode | NoneNonePermission mode for tool usage
continue_conversationboolFalseContinue the most recent conversation
resumestr | NoneNoneSession ID to resume
max_turnsint | NoneNoneMaximum conversation turns
disallowed_toolslist[str][]List of disallowed tool names
modelstr | NoneNoneClaude model to use
permission_prompt_tool_namestr | NoneNoneMCP tool name for permission prompts
cwdstr | Path | NoneNoneCurrent working directory
settingsstr | NoneNonePath to settings file
add_dirslist[str | Path][]Additional directories Claude can access
extra_argsdict[str, str | None]{}Additional CLI arguments to pass directly to the CLI
can_use_toolCanUseTool | NoneNoneTool permission callback function
hooksdict[HookEvent, list[HookMatcher]] | NoneNoneHook configurations for intercepting events

PermissionMode

Permission modes for controlling tool execution.
PermissionMode = Literal[
    "default",           # Standard permission behavior
    "acceptEdits",       # Auto-accept file edits
    "plan",              # Planning mode - no execution
    "bypassPermissions"  # Bypass all permission checks (use with caution)
]

McpSdkServerConfig

Configuration for SDK MCP servers created with create_sdk_mcp_server().
class McpSdkServerConfig(TypedDict):
    type: Literal["sdk"]
    name: str
    instance: Any  # MCP Server instance

McpServerConfig

Union type for MCP server configurations.
McpServerConfig = McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig

McpStdioServerConfig

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]  # Optional for backwards compatibility
    command: str
    args: NotRequired[list[str]]
    env: NotRequired[dict[str, str]]

McpSSEServerConfig

class McpSSEServerConfig(TypedDict):
    type: Literal["sse"]
    url: str
    headers: NotRequired[dict[str, str]]

McpHttpServerConfig

class McpHttpServerConfig(TypedDict):
    type: Literal["http"]
    url: str
    headers: NotRequired[dict[str, str]]

Message Types

Message

Union type of all possible messages.
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage

UserMessage

User input message.
@dataclass
class UserMessage:
    content: str | list[ContentBlock]

AssistantMessage

Assistant response message with content blocks.
@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str

SystemMessage

System message with metadata.
@dataclass
class SystemMessage:
    subtype: str
    data: dict[str, Any]

ResultMessage

Final result message with cost and usage information.
@dataclass
class ResultMessage:
    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

Content Block Types

ContentBlock

Union type of all content blocks.
ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

TextBlock

Text content block.
@dataclass
class TextBlock:
    text: str

ThinkingBlock

Thinking content block (for models with thinking capability).
@dataclass
class ThinkingBlock:
    thinking: str
    signature: str

ToolUseBlock

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

ToolResultBlock

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

Error Types

ClaudeSDKError

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

CLINotFoundError

Raised when Claude Code CLI is not installed or not found.
class CLINotFoundError(CLIConnectionError):
    def __init__(self, message: str = "Claude Code not found", cli_path: str | None = None):
        """
        Args:
            message: Error message (default: "Claude Code not found")
            cli_path: Optional path to the CLI that was not found
        """

CLIConnectionError

Raised when connection to Claude Code fails.
class CLIConnectionError(ClaudeSDKError):
    """Failed to connect to Claude Code."""

ProcessError

Raised when the Claude Code process fails.
class ProcessError(ClaudeSDKError):
    def __init__(self, message: str, exit_code: int | None = None, stderr: str | None = None):
        self.exit_code = exit_code
        self.stderr = stderr

CLIJSONDecodeError

Raised when JSON parsing fails.
class CLIJSONDecodeError(ClaudeSDKError):
    def __init__(self, line: str, original_error: Exception):
        """
        Args:
            line: The line that failed to parse
            original_error: The original JSON decode exception
        """
        self.line = line
        self.original_error = original_error

Hook Types

HookEvent

Supported hook event types. Note that due to setup limitations, the Python SDK does not support SessionStart, SessionEnd, and Notification hooks.
HookEvent = Literal[
    "PreToolUse",      # Called before tool execution
    "PostToolUse",     # Called after tool execution
    "UserPromptSubmit", # Called when user submits a prompt
    "Stop",            # Called when stopping execution
    "SubagentStop",    # Called when a subagent stops
    "PreCompact"       # Called before message compaction
]

HookCallback

Type definition for hook callback functions.
HookCallback = Callable[
    [dict[str, Any], str | None, HookContext],
    Awaitable[dict[str, Any]]
]
Parameters:
  • input_data: Hook-specific input data (see hook documentation)
  • tool_use_id: Optional tool use identifier (for tool-related hooks)
  • context: Hook context with additional information
Returns a dictionary that may contain:
  • decision: "block" to block the action
  • systemMessage: System message to add to the transcript
  • hookSpecificOutput: Hook-specific output data

HookContext

Context information passed to hook callbacks.
@dataclass
class HookContext:
    signal: Any | None = None  # Future: abort signal support

HookMatcher

Configuration for matching hooks to specific events or tools.
@dataclass
class HookMatcher:
    matcher: str | None = None        # Tool name or pattern to match (e.g., "Bash", "Write|Edit")
    hooks: list[HookCallback] = field(default_factory=list)  # List of callbacks to execute

Hook Usage Example

from claude_code_sdk import query, ClaudeCodeOptions, HookMatcher, HookContext
from typing import Any

async def validate_bash_command(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """Validate and potentially block dangerous bash commands."""
    if input_data['tool_name'] == 'Bash':
        command = input_data['tool_input'].get('command', '')
        if 'rm -rf /' in command:
            return {
                'hookSpecificOutput': {
                    'hookEventName': 'PreToolUse',
                    'permissionDecision': 'deny',
                    'permissionDecisionReason': 'Dangerous command blocked'
                }
            }
    return {}

async def log_tool_use(
    input_data: dict[str, Any],
    tool_use_id: str | None,
    context: HookContext
) -> dict[str, Any]:
    """Log all tool usage for auditing."""
    print(f"Tool used: {input_data.get('tool_name')}")
    return {}

options = ClaudeCodeOptions(
    hooks={
        'PreToolUse': [
            HookMatcher(matcher='Bash', hooks=[validate_bash_command]),
            HookMatcher(hooks=[log_tool_use])  # Applies to all tools
        ],
        'PostToolUse': [
            HookMatcher(hooks=[log_tool_use])
        ]
    }
)

async for message in query(
    prompt="Analyze this codebase",
    options=options
):
    print(message)

Tool Input/Output Types

Documentation of input/output schemas for all built-in Claude Code tools. While the Python SDK doesn’t export these as types, they represent the structure of tool inputs and outputs in messages.

Task

Tool name: Task Input:
{
    "description": str,      # A short (3-5 word) description of the task
    "prompt": str,           # The task for the agent to perform
    "subagent_type": str     # The type of specialized agent to use
}
Output:
{
    "result": str,                    # Final result from the subagent
    "usage": dict | None,             # Token usage statistics
    "total_cost_usd": float | None,  # Total cost in USD
    "duration_ms": int | None         # Execution duration in milliseconds
}

Bash

Tool name: Bash Input:
{
    "command": str,                  # The command to execute
    "timeout": int | None,           # Optional timeout in milliseconds (max 600000)
    "description": str | None,       # Clear, concise description (5-10 words)
    "run_in_background": bool | None # Set to true to run in background
}
Output:
{
    "output": str,              # Combined stdout and stderr output
    "exitCode": int,            # Exit code of the command
    "killed": bool | None,      # Whether command was killed due to timeout
    "shellId": str | None       # Shell ID for background processes
}

Edit

Tool name: Edit Input:
{
    "file_path": str,           # The absolute path to the file to modify
    "old_string": str,          # The text to replace
    "new_string": str,          # The text to replace it with
    "replace_all": bool | None  # Replace all occurrences (default False)
}
Output:
{
    "message": str,      # Confirmation message
    "replacements": int, # Number of replacements made
    "file_path": str     # File path that was edited
}

MultiEdit

Tool name: MultiEdit Input:
{
    "file_path": str,     # The absolute path to the file to modify
    "edits": [            # Array of edit operations
        {
            "old_string": str,          # The text to replace
            "new_string": str,          # The text to replace it with
            "replace_all": bool | None  # Replace all occurrences
        }
    ]
}
Output:
{
    "message": str,       # Success message
    "edits_applied": int, # Total number of edits applied
    "file_path": str      # File path that was edited
}

Read

Tool name: Read Input:
{
    "file_path": str,       # The absolute path to the file to read
    "offset": int | None,   # The line number to start reading from
    "limit": int | None     # The number of lines to read
}
Output (Text files):
{
    "content": str,         # File contents with line numbers
    "total_lines": int,     # Total number of lines in file
    "lines_returned": int   # Lines actually returned
}
Output (Images):
{
    "image": str,       # Base64 encoded image data
    "mime_type": str,   # Image MIME type
    "file_size": int    # File size in bytes
}

Write

Tool name: Write Input:
{
    "file_path": str,  # The absolute path to the file to write
    "content": str     # The content to write to the file
}
Output:
{
    "message": str,        # Success message
    "bytes_written": int,  # Number of bytes written
    "file_path": str       # File path that was written
}

Glob

Tool name: Glob Input:
{
    "pattern": str,       # The glob pattern to match files against
    "path": str | None    # The directory to search in (defaults to cwd)
}
Output:
{
    "matches": list[str],  # Array of matching file paths
    "count": int,          # Number of matches found
    "search_path": str     # Search directory used
}

Grep

Tool name: Grep Input:
{
    "pattern": str,                    # The regular expression pattern
    "path": str | None,                # File or directory to search in
    "glob": str | None,                # Glob pattern to filter files
    "type": str | None,                # File type to search
    "output_mode": str | None,         # "content", "files_with_matches", or "count"
    "-i": bool | None,                 # Case insensitive search
    "-n": bool | None,                 # Show line numbers
    "-B": int | None,                  # Lines to show before each match
    "-A": int | None,                  # Lines to show after each match
    "-C": int | None,                  # Lines to show before and after
    "head_limit": int | None,          # Limit output to first N lines/entries
    "multiline": bool | None           # Enable multiline mode
}
Output (content mode):
{
    "matches": [
        {
            "file": str,
            "line_number": int | None,
            "line": str,
            "before_context": list[str] | None,
            "after_context": list[str] | None
        }
    ],
    "total_matches": int
}
Output (files_with_matches mode):
{
    "files": list[str],  # Files containing matches
    "count": int         # Number of files with matches
}

NotebookEdit

Tool name: NotebookEdit Input:
{
    "notebook_path": str,                     # Absolute path to the Jupyter notebook
    "cell_id": str | None,                    # The ID of the cell to edit
    "new_source": str,                        # The new source for the cell
    "cell_type": "code" | "markdown" | None,  # The type of the cell
    "edit_mode": "replace" | "insert" | "delete" | None  # Edit operation type
}
Output:
{
    "message": str,                              # Success message
    "edit_type": "replaced" | "inserted" | "deleted",  # Type of edit performed
    "cell_id": str | None,                       # Cell ID that was affected
    "total_cells": int                           # Total cells in notebook after edit
}

WebFetch

Tool name: WebFetch Input:
{
    "url": str,     # The URL to fetch content from
    "prompt": str   # The prompt to run on the fetched content
}
Output:
{
    "response": str,           # AI model's response to the prompt
    "url": str,                # URL that was fetched
    "final_url": str | None,   # Final URL after redirects
    "status_code": int | None  # HTTP status code
}

WebSearch

Tool name: WebSearch Input:
{
    "query": str,                        # The search query to use
    "allowed_domains": list[str] | None, # Only include results from these domains
    "blocked_domains": list[str] | None  # Never include results from these domains
}
Output:
{
    "results": [
        {
            "title": str,
            "url": str,
            "snippet": str,
            "metadata": dict | None
        }
    ],
    "total_results": int,
    "query": str
}

TodoWrite

Tool name: TodoWrite Input:
{
    "todos": [
        {
            "content": str,                              # The task description
            "status": "pending" | "in_progress" | "completed",  # Task status
            "activeForm": str                            # Active form of the description
        }
    ]
}
Output:
{
    "message": str,  # Success message
    "stats": {
        "total": int,
        "pending": int,
        "in_progress": int,
        "completed": int
    }
}

BashOutput

Tool name: BashOutput Input:
{
    "bash_id": str,       # The ID of the background shell
    "filter": str | None  # Optional regex to filter output lines
}
Output:
{
    "output": str,                                      # New output since last check
    "status": "running" | "completed" | "failed",       # Current shell status
    "exitCode": int | None                              # Exit code when completed
}

KillBash

Tool name: KillBash Input:
{
    "shell_id": str  # The ID of the background shell to kill
}
Output:
{
    "message": str,  # Success message
    "shell_id": str  # ID of the killed shell
}

ExitPlanMode

Tool name: ExitPlanMode Input:
{
    "plan": str  # The plan to run by the user for approval
}
Output:
{
    "message": str,          # Confirmation message
    "approved": bool | None  # Whether user approved the plan
}

ListMcpResources

Tool name: ListMcpResources Input:
{
    "server": str | None  # Optional server name to filter resources by
}
Output:
{
    "resources": [
        {
            "uri": str,
            "name": str,
            "description": str | None,
            "mimeType": str | None,
            "server": str
        }
    ],
    "total": int
}

ReadMcpResource

Tool name: ReadMcpResource Input:
{
    "server": str,  # The MCP server name
    "uri": str      # The resource URI to read
}
Output:
{
    "contents": [
        {
            "uri": str,
            "mimeType": str | None,
            "text": str | None,
            "blob": str | None
        }
    ],
    "server": str
}

Example Usage

Basic file operations

from claude_code_sdk import query, ClaudeCodeOptions, AssistantMessage, ToolUseBlock
import asyncio

async def create_project():
    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode='acceptEdits',
        cwd="/home/user/project"
    )
    
    async for message in query(
        prompt="Create a Python project structure with setup.py",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    print(f"Using tool: {block.name}")

asyncio.run(create_project())

Error handling

from claude_code_sdk import (
    query,
    CLINotFoundError,
    ProcessError,
    CLIJSONDecodeError
)

try:
    async for message in query(prompt="Hello"):
        print(message)
except CLINotFoundError:
    print("Please install Claude Code: npm install -g @anthropic-ai/claude-code")
except ProcessError as e:
    print(f"Process failed with exit code: {e.exit_code}")
except CLIJSONDecodeError as e:
    print(f"Failed to parse response: {e}")

Streaming mode with client

from claude_code_sdk import ClaudeSDKClient
import asyncio

async def interactive_session():
    async with ClaudeSDKClient() as client:
        # Send initial message
        await client.query("What's the weather like?")
        
        # Process responses
        async for msg in client.receive_response():
            print(msg)
        
        # Send follow-up
        await client.query("Tell me more about that")
        
        # Process follow-up response
        async for msg in client.receive_response():
            print(msg)

asyncio.run(interactive_session())

Using custom tools

from claude_code_sdk import (
    query,
    ClaudeCodeOptions,
    tool,
    create_sdk_mcp_server,
    AssistantMessage,
    TextBlock
)
import asyncio
from typing import Any

# Define custom tools with @tool decorator
@tool("calculate", "Perform mathematical calculations", {"expression": str})
async def calculate(args: dict[str, Any]) -> dict[str, Any]:
    try:
        result = eval(args["expression"], {"__builtins__": {}})
        return {
            "content": [{
                "type": "text",
                "text": f"Result: {result}"
            }]
        }
    except Exception as e:
        return {
            "content": [{
                "type": "text",
                "text": f"Error: {str(e)}"
            }],
            "is_error": True
        }

@tool("get_time", "Get current time", {})
async def get_time(args: dict[str, Any]) -> dict[str, Any]:
    from datetime import datetime
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return {
        "content": [{
            "type": "text",
            "text": f"Current time: {current_time}"
        }]
    }

async def main():
    # Create SDK MCP server with custom tools
    my_server = create_sdk_mcp_server(
        name="utilities",
        version="1.0.0",
        tools=[calculate, get_time]
    )
    
    # Configure options with the server
    options = ClaudeCodeOptions(
        mcp_servers={"utils": my_server},
        allowed_tools=[
            "mcp__utils__calculate",
            "mcp__utils__get_time"
        ]
    )
    
    # Query Claude with custom tools available
    async for message in query(
        prompt="What's 123 * 456 and what time is it?",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

asyncio.run(main())

See also