Справочник Agent SDK - Python

Полный справочник API для Python Agent SDK, включая все функции, типы и классы.

Установка

pip install claude-agent-sdk

Выбор между query() и ClaudeSDKClient

Python SDK предоставляет два способа взаимодействия с Claude Code:

Быстрое сравнение

Когда использовать query() (новый сеанс каждый раз)

Лучше всего для:

• Одноразовых вопросов, когда вам не нужна история разговора
• Независимых задач, которые не требуют контекста из предыдущих обменов
• Простых скриптов автоматизации
• Когда вы хотите начать с чистого листа каждый раз

Когда использовать ClaudeSDKClient (непрерывный разговор)

Лучше всего для:

• Продолжения разговоров - Когда вам нужно, чтобы Claude помнил контекст
• Уточняющих вопросов - Построение на основе предыдущих ответов
• Интерактивных приложений - Интерфейсы чата, REPL
• Логики, управляемой ответом - Когда следующее действие зависит от ответа Claude
• Управления сеансом - Явное управление жизненным циклом разговора

Функции

query()

Создает новый сеанс для каждого взаимодействия с Claude Code. Возвращает асинхронный итератор, который выдает сообщения по мере их поступления. Каждый вызов query() начинается с нуля без памяти о предыдущих взаимодействиях.

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeAgentOptions | None = None,
    transport: Transport | None = None
) -> AsyncIterator[Message]

Параметры

Возвращаемое значение

Возвращает AsyncIterator[Message], который выдает сообщения из разговора.

Пример - С параметрами

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions


async def main():
    options = ClaudeAgentOptions(
        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()

Декоратор для определения MCP tools с проверкой типов.

def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any],
    annotations: ToolAnnotations | None = None
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]

Параметры

Варианты схемы ввода

1. Простое сопоставление типов (рекомендуется):

   {"text": str, "count": int, "enabled": bool}
   

2. Формат JSON Schema (для сложной валидации):

   {
       "type": "object",
       "properties": {
           "text": {"type": "string"},
           "count": {"type": "integer", "minimum": 0},
       },
       "required": ["text"],
   }
   

Возвращаемое значение

Функция-декоратор, которая оборачивает реализацию инструмента и возвращает экземпляр SdkMcpTool.

Пример

from claude_agent_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']}!"}]}

ToolAnnotations

Переэкспортировано из mcp.types (также доступно как from claude_agent_sdk import ToolAnnotations). Все поля являются дополнительными подсказками; клиенты не должны полагаться на них для решений безопасности.

from claude_agent_sdk import tool, ToolAnnotations
from typing import Any


@tool(
    "search",
    "Search the web",
    {"query": str},
    annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
)
async def search(args: dict[str, Any]) -> dict[str, Any]:
    return {"content": [{"type": "text", "text": f"Results for: {args['query']}"}]}

create_sdk_mcp_server()

Создайте встроенный MCP server, который работает в вашем приложении Python.

def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig

Параметры

Возвращаемое значение

Возвращает объект McpSdkServerConfig, который можно передать в ClaudeAgentOptions.mcp_servers.

Пример

from claude_agent_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 = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["mcp__calc__add", "mcp__calc__multiply"],
)

list_sessions()

Выводит список прошлых сеансов с метаданными. Фильтруйте по каталогу проекта или выводите сеансы во всех проектах. Синхронно; возвращается немедленно.

def list_sessions(
    directory: str | None = None,
    limit: int | None = None,
    include_worktrees: bool = True
) -> list[SDKSessionInfo]

Параметры

Тип возвращаемого значения: SDKSessionInfo

Пример

Выведите 10 самых последних сеансов для проекта. Результаты отсортированы по last_modified в убывающем порядке, поэтому первый элемент - самый новый. Опустите directory, чтобы искать во всех проектах.

from claude_agent_sdk import list_sessions

for session in list_sessions(directory="/path/to/project", limit=10):
    print(f"{session.summary} ({session.session_id})")

get_session_messages()

Извлекает сообщения из прошлого сеанса. Синхронно; возвращается немедленно.

def get_session_messages(
    session_id: str,
    directory: str | None = None,
    limit: int | None = None,
    offset: int = 0
) -> list[SessionMessage]

Параметры

Тип возвращаемого значения: SessionMessage

Пример

from claude_agent_sdk import list_sessions, get_session_messages

sessions = list_sessions(limit=1)
if sessions:
    messages = get_session_messages(sessions[0].session_id)
    for msg in messages:
        print(f"[{msg.type}] {msg.uuid}")

get_session_info()

Читает метаданные для одного сеанса по ID без сканирования полного каталога проекта. Синхронно; возвращается немедленно.

def get_session_info(
    session_id: str,
    directory: str | None = None,
) -> SDKSessionInfo | None

Параметры

Возвращает SDKSessionInfo или None, если сеанс не найден.

Пример

Найдите метаданные одного сеанса без сканирования каталога проекта. Полезно, когда у вас уже есть ID сеанса из предыдущего запуска.

from claude_agent_sdk import get_session_info

info = get_session_info("550e8400-e29b-41d4-a716-446655440000")
if info:
    print(f"{info.summary} (branch: {info.git_branch}, tag: {info.tag})")

rename_session()

Переименовывает сеанс, добавляя запись с пользовательским названием. Повторные вызовы безопасны; побеждает самое последнее название. Синхронно.

def rename_session(
    session_id: str,
    title: str,
    directory: str | None = None,
) -> None

Параметры

Вызывает ValueError, если session_id не является допустимым UUID или title пуст; FileNotFoundError, если сеанс не найден.

Пример

Переименуйте самый последний сеанс, чтобы его было легче найти позже. Новое название появляется в SDKSessionInfo.custom_title при последующих чтениях.

from claude_agent_sdk import list_sessions, rename_session

sessions = list_sessions(directory="/path/to/project", limit=1)
if sessions:
    rename_session(sessions[0].session_id, "Refactor auth module")

tag_session()

Помечает сеанс. Передайте None для очистки тега. Повторные вызовы безопасны; побеждает самый последний тег. Синхронно.

def tag_session(
    session_id: str,
    tag: str | None,
    directory: str | None = None,
) -> None

Параметры

Вызывает ValueError, если session_id не является допустимым UUID или tag пуст после очистки; FileNotFoundError, если сеанс не найден.

Пример

Пометьте сеанс, затем отфильтруйте по этому тегу при последующем чтении. Передайте None для очистки существующего тега.

from claude_agent_sdk import list_sessions, tag_session

# Tag a session
tag_session("550e8400-e29b-41d4-a716-446655440000", "needs-review")

# Later: find all sessions with that tag
for session in list_sessions(directory="/path/to/project"):
    if session.tag == "needs-review":
        print(session.summary)

Классы

ClaudeSDKClient

Поддерживает сеанс разговора через несколько обменов. Это эквивалент Python того, как функция query() TypeScript SDK работает внутри - она создает объект клиента, который может продолжать разговоры.

Ключевые особенности

• Непрерывность сеанса: Поддерживает контекст разговора через несколько вызовов query()
• Один разговор: Сеанс сохраняет предыдущие сообщения
• Поддержка прерываний: Может остановить выполнение в середине задачи
• Явный жизненный цикл: Вы контролируете, когда сеанс начинается и заканчивается
• Поток, управляемый ответом: Может реагировать на ответы и отправлять дополнения
• Пользовательские инструменты и hooks: Поддерживает пользовательские инструменты (созданные с помощью декоратора @tool) и hooks

class ClaudeSDKClient:
    def __init__(self, options: ClaudeAgentOptions | None = None, transport: Transport | 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 set_permission_mode(self, mode: str) -> None
    async def set_model(self, model: str | None = None) -> None
    async def rewind_files(self, user_message_id: str) -> None
    async def get_mcp_status(self) -> McpStatusResponse
    async def reconnect_mcp_server(self, server_name: str) -> None
    async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None
    async def stop_task(self, task_id: str) -> None
    async def get_server_info(self) -> dict[str, Any] | None
    async def disconnect(self) -> None

Методы

Поддержка менеджера контекста

Клиент можно использовать как асинхронный менеджер контекста для автоматического управления соединением:

async with ClaudeSDKClient() as client:
    await client.query("Hello Claude")
    async for message in client.receive_response():
        print(message)

Важно: При итерации по сообщениям избегайте использования break для раннего выхода, так как это может вызвать проблемы с очисткой asyncio. Вместо этого позвольте итерации завершиться естественным образом или используйте флаги для отслеживания, когда вы нашли то, что вам нужно.

Пример - Продолжение разговора

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


async def main():
    async with ClaudeSDKClient() as client:
        # First question
        await client.query("What's the capital of France?")

        # Process response
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # Follow-up question - the session retains the previous context
        await client.query("What's the population of that city?")

        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # Another follow-up - still in the same conversation
        await client.query("What are some famous landmarks there?")

        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")


asyncio.run(main())

Пример - Потоковый ввод с ClaudeSDKClient

import asyncio
from claude_agent_sdk import ClaudeSDKClient


async def message_stream():
    """Generate messages dynamically."""
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Analyze the following data:"},
    }
    await asyncio.sleep(0.5)
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Temperature: 25°C, Humidity: 60%"},
    }
    await asyncio.sleep(0.5)
    yield {
        "type": "user",
        "message": {"role": "user", "content": "What patterns do you see?"},
    }


async def main():
    async with ClaudeSDKClient() as client:
        # Stream input to Claude
        await client.query(message_stream())

        # Process response
        async for message in client.receive_response():
            print(message)

        # Follow-up in same session
        await client.query("Should we be concerned about these readings?")

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


asyncio.run(main())

Пример - Использование прерываний

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, ResultMessage


async def interruptible_task():
    options = ClaudeAgentOptions(allowed_tools=["Bash"], permission_mode="acceptEdits")

    async with ClaudeSDKClient(options=options) as client:
        # Start a long-running task
        await client.query("Count from 1 to 100 slowly, using the bash sleep command")

        # Let it run for a bit
        await asyncio.sleep(2)

        # Interrupt the task
        await client.interrupt()
        print("Task interrupted!")

        # Drain the interrupted task's messages (including its ResultMessage)
        async for message in client.receive_response():
            if isinstance(message, ResultMessage):
                print(f"Interrupted task finished with subtype={message.subtype!r}")
                # subtype is "error_during_execution" for interrupted tasks

        # Send a new command
        await client.query("Just say hello instead")

        # Now receive the new response
        async for message in client.receive_response():
            if isinstance(message, ResultMessage) and message.subtype == "success":
                print(f"New result: {message.result}")


asyncio.run(interruptible_task())

Пример - Расширенное управление разрешениями

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from claude_agent_sdk.types import (
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext,
)


async def custom_permission_handler(
    tool_name: str, input_data: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    """Custom logic for tool permissions."""

    # Block writes to system directories
    if tool_name == "Write" and input_data.get("file_path", "").startswith("/system/"):
        return PermissionResultDeny(
            message="System directory write not allowed", interrupt=True
        )

    # Redirect sensitive file operations
    if tool_name in ["Write", "Edit"] and "config" in input_data.get("file_path", ""):
        safe_path = f"./sandbox/{input_data['file_path']}"
        return PermissionResultAllow(
            updated_input={**input_data, "file_path": safe_path}
        )

    # Allow everything else
    return PermissionResultAllow(updated_input=input_data)


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

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Update the system config file")

        async for message in client.receive_response():
            # Will use sandbox path instead
            print(message)


asyncio.run(main())

Типы

SdkMcpTool

Определение для SDK MCP tool, созданного с помощью декоратора @tool.

@dataclass
class SdkMcpTool(Generic[T]):
    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[[T], Awaitable[dict[str, Any]]]
    annotations: ToolAnnotations | None = None

Transport

Абстрактный базовый класс для пользовательских реализаций транспорта. Используйте это для связи с процессом Claude через пользовательский канал (например, удаленное соединение вместо локального подпроцесса).

from abc import ABC, abstractmethod
from collections.abc import AsyncIterator
from typing import Any


class Transport(ABC):
    @abstractmethod
    async def connect(self) -> None: ...

    @abstractmethod
    async def write(self, data: str) -> None: ...

    @abstractmethod
    def read_messages(self) -> AsyncIterator[dict[str, Any]]: ...

    @abstractmethod
    async def close(self) -> None: ...

    @abstractmethod
    def is_ready(self) -> bool: ...

    @abstractmethod
    async def end_input(self) -> None: ...

Импорт: from claude_agent_sdk import Transport

ClaudeAgentOptions

Dataclass конфигурации для запросов Claude Code.

@dataclass
class ClaudeAgentOptions:
    tools: list[str] | ToolsPreset | None = None
    allowed_tools: list[str] = field(default_factory=list)
    system_prompt: str | SystemPromptPreset | None = None
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    strict_mcp_config: bool = False
    permission_mode: PermissionMode | None = None
    continue_conversation: bool = False
    resume: str | None = None
    max_turns: int | None = None
    max_budget_usd: float | None = None
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    fallback_model: str | None = None
    betas: list[SdkBeta] = field(default_factory=list)
    output_format: dict[str, Any] | None = None
    permission_prompt_tool_name: str | None = None
    cwd: str | Path | None = None
    cli_path: 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)
    max_buffer_size: int | None = None
    debug_stderr: Any = sys.stderr  # Deprecated
    stderr: Callable[[str], None] | None = None
    can_use_tool: CanUseTool | None = None
    hooks: dict[HookEvent, list[HookMatcher]] | None = None
    user: str | None = None
    include_partial_messages: bool = False
    include_hook_events: bool = False
    fork_session: bool = False
    agents: dict[str, AgentDefinition] | None = None
    setting_sources: list[SettingSource] | None = None
    sandbox: SandboxSettings | None = None
    plugins: list[SdkPluginConfig] = field(default_factory=list)
    max_thinking_tokens: int | None = None  # Deprecated: use thinking instead
    thinking: ThinkingConfig | None = None
    effort: Literal["low", "medium", "high", "xhigh", "max"] | None = None
    enable_file_checkpointing: bool = False
    session_store: SessionStore | None = None
    session_store_flush: SessionStoreFlushMode = "batched"

Обработка медленных или зависших ответов API

Подпроцесс CLI читает несколько переменных окружения, которые управляют тайм-аутами API и обнаружением зависания. Передайте их через ClaudeAgentOptions.env:

options = ClaudeAgentOptions(
    env={
        "API_TIMEOUT_MS": "120000",
        "CLAUDE_CODE_MAX_RETRIES": "2",
        "CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS": "120000",
    },
)

• API_TIMEOUT_MS: тайм-аут для каждого запроса на клиенте Anthropic в миллисекундах. По умолчанию 600000. Применяется к основному циклу и всем подагентам.
• CLAUDE_CODE_MAX_RETRIES: максимальное количество повторных попыток API. По умолчанию 10. Каждая повторная попытка получает свое собственное окно API_TIMEOUT_MS, поэтому наихудшее время стены примерно API_TIMEOUT_MS × (CLAUDE_CODE_MAX_RETRIES + 1) плюс отступ.
• CLAUDE_ASYNC_AGENT_STALL_TIMEOUT_MS: сторожевой таймер зависания для подагентов, запущенных с run_in_background. По умолчанию 600000. Сбрасывается при каждом событии потока; при зависании он прерывает подагента, отмечает задачу как неудачную и выводит ошибку родителю с любым частичным результатом. Не применяется к синхронным подагентам.
• CLAUDE_ENABLE_STREAM_WATCHDOG=1 с CLAUDE_STREAM_IDLE_TIMEOUT_MS: прерывает запрос, когда заголовки прибыли, но тело ответа перестает потоковать. По умолчанию отключено. CLAUDE_STREAM_IDLE_TIMEOUT_MS по умолчанию 300000 и зажимается до этого минимума. Прерванный запрос проходит через обычный путь повторной попытки.

OutputFormat

Конфигурация для валидации структурированного вывода. Передайте это как dict в поле output_format на ClaudeAgentOptions:

# Expected dict shape for output_format
{
    "type": "json_schema",
    "schema": {...},  # Your JSON Schema definition
}

SystemPromptPreset

Конфигурация для использования предустановленной системной подсказки Claude Code с дополнительными добавлениями.

class SystemPromptPreset(TypedDict):
    type: Literal["preset"]
    preset: Literal["claude_code"]
    append: NotRequired[str]
    exclude_dynamic_sections: NotRequired[bool]

SettingSource

Управляет тем, какие источники конфигурации на основе файловой системы загружает SDK.

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

Поведение по умолчанию

Когда setting_sources опущено или None, query() загружает те же параметры файловой системы, что и Claude Code CLI: пользовательские, проектные и локальные. Управляемые параметры политики загружаются во всех случаях. См. What settingSources does not control для входов, которые читаются независимо от этой опции, и как их отключить.

Почему использовать setting_sources

Отключить параметры файловой системы:

# Do not load user, project, or local settings from disk
from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="Analyze this code",
    options=ClaudeAgentOptions(
        setting_sources=[]
    ),
):
    print(message)

Загрузить все параметры файловой системы явно:

from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="Analyze this code",
    options=ClaudeAgentOptions(
        setting_sources=["user", "project", "local"]
    ),
):
    print(message)

Загрузить только определенные источники параметров:

# Load only project settings, ignore user and local
async for message in query(
    prompt="Run CI checks",
    options=ClaudeAgentOptions(
        setting_sources=["project"]  # Only .claude/settings.json
    ),
):
    print(message)

Тестирование и окружения CI:

# Ensure consistent behavior in CI by excluding local settings
async for message in query(
    prompt="Run tests",
    options=ClaudeAgentOptions(
        setting_sources=["project"],  # Only team-shared settings
        permission_mode="bypassPermissions",
    ),
):
    print(message)

Приложения только SDK:

# Define everything programmatically.
# Pass [] to opt out of filesystem setting sources.
async for message in query(
    prompt="Review this PR",
    options=ClaudeAgentOptions(
        setting_sources=[],
        agents={...},
        mcp_servers={...},
        allowed_tools=["Read", "Grep", "Glob"],
    ),
):
    print(message)

Загрузка инструкций проекта CLAUDE.md:

# Load project settings to include CLAUDE.md files
async for message in query(
    prompt="Add a new feature following project conventions",
    options=ClaudeAgentOptions(
        system_prompt={
            "type": "preset",
            "preset": "claude_code",  # Use Claude Code's system prompt
        },
        setting_sources=["project"],  # Loads CLAUDE.md from project
        allowed_tools=["Read", "Write", "Edit"],
    ),
):
    print(message)

Приоритет параметров

Когда загружаются несколько источников, параметры объединяются с этим приоритетом (от наивысшего к наименьшему):

1. Локальные параметры (.claude/settings.local.json)
2. Параметры проекта (.claude/settings.json)
3. Параметры пользователя (~/.claude/settings.json)

Программные опции, такие как agents и allowed_tools, переопределяют параметры пользователя, проекта и локальной файловой системы. Управляемые параметры политики имеют приоритет над программными опциями.

AgentDefinition

Конфигурация для подагента, определенного программно.

@dataclass
class AgentDefinition:
    description: str
    prompt: str
    tools: list[str] | None = None
    disallowedTools: list[str] | None = None
    model: str | None = None
    skills: list[str] | None = None
    memory: Literal["user", "project", "local"] | None = None
    mcpServers: list[str | dict[str, Any]] | None = None
    initialPrompt: str | None = None
    maxTurns: int | None = None
    background: bool | None = None
    effort: Literal["low", "medium", "high", "xhigh", "max"] | int | None = None
    permissionMode: PermissionMode | None = None

PermissionMode

Режимы разрешений для управления выполнением инструментов.

PermissionMode = Literal[
    "default",  # Standard permission behavior
    "acceptEdits",  # Auto-accept file edits
    "plan",  # Planning mode - read-only tools only
    "dontAsk",  # Deny anything not pre-approved instead of prompting
    "bypassPermissions",  # Bypass all permission checks (use with caution)
]

CanUseTool

Псевдоним типа для функций обратного вызова разрешения инструмента.

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

Обратный вызов получает:

• tool_name: Имя вызываемого инструмента
• input_data: Входные параметры инструмента
• context: ToolPermissionContext с дополнительной информацией

Возвращает PermissionResult (либо PermissionResultAllow, либо PermissionResultDeny).

ToolPermissionContext

Информация контекста, передаваемая в обратные вызовы разрешения инструмента.

@dataclass
class ToolPermissionContext:
    signal: Any | None = None  # Future: abort signal support
    suggestions: list[PermissionUpdate] = field(default_factory=list)
    blocked_path: str | None = None
    decision_reason: str | None = None
    title: str | None = None
    display_name: str | None = None
    description: str | None = None

PermissionResult

Тип объединения для результатов обратного вызова разрешения.

PermissionResult = PermissionResultAllow | PermissionResultDeny

PermissionResultAllow

Результат, указывающий, что вызов инструмента должен быть разрешен.

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

PermissionResultDeny

Результат, указывающий, что вызов инструмента должен быть отклонен.

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

PermissionUpdate

Конфигурация для программного обновления разрешений.

@dataclass
class PermissionUpdate:
    type: Literal[
        "addRules",
        "replaceRules",
        "removeRules",
        "setMode",
        "addDirectories",
        "removeDirectories",
    ]
    rules: list[PermissionRuleValue] | None = None
    behavior: Literal["allow", "deny", "ask"] | None = None
    mode: PermissionMode | None = None
    directories: list[str] | None = None
    destination: (
        Literal["userSettings", "projectSettings", "localSettings", "session"] | None
    ) = None

PermissionRuleValue

Правило для добавления, замены или удаления в обновлении разрешений.

@dataclass
class PermissionRuleValue:
    tool_name: str
    rule_content: str | None = None

ToolsPreset

Конфигурация предустановленных инструментов для использования набора инструментов Claude Code по умолчанию.

class ToolsPreset(TypedDict):
    type: Literal["preset"]
    preset: Literal["claude_code"]

ThinkingConfig

Управляет поведением расширенного мышления. Объединение трех конфигураций:

class ThinkingConfigAdaptive(TypedDict):
    type: Literal["adaptive"]


class ThinkingConfigEnabled(TypedDict):
    type: Literal["enabled"]
    budget_tokens: int


class ThinkingConfigDisabled(TypedDict):
    type: Literal["disabled"]


ThinkingConfig = ThinkingConfigAdaptive | ThinkingConfigEnabled | ThinkingConfigDisabled

Поскольку это классы TypedDict, они являются простыми словарями во время выполнения. Либо создавайте их как литералы словарей, либо вызывайте класс как конструктор; оба создают dict. Получайте доступ к полям с помощью config["budget_tokens"], а не config.budget_tokens:

from claude_agent_sdk import ClaudeAgentOptions, ThinkingConfigEnabled

# Option 1: dict literal (recommended, no import needed)
options = ClaudeAgentOptions(thinking={"type": "enabled", "budget_tokens": 20000})

# Option 2: constructor-style (returns a plain dict)
config = ThinkingConfigEnabled(type="enabled", budget_tokens=20000)
print(config["budget_tokens"])  # 20000
# config.budget_tokens would raise AttributeError

SdkBeta

Тип Literal для функций бета-версии SDK.

SdkBeta = Literal["context-1m-2025-08-07"]

Используйте с полем betas в ClaudeAgentOptions для включения функций бета-версии.

McpSdkServerConfig

Конфигурация для SDK MCP servers, созданных с помощью create_sdk_mcp_server().

class McpSdkServerConfig(TypedDict):
    type: Literal["sdk"]
    name: str
    instance: Any  # MCP Server instance

McpServerConfig

Тип объединения для конфигураций MCP server.

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

McpServerStatusConfig

Конфигурация MCP server, как сообщается get_mcp_status(). Это объединение всех вариантов транспорта McpServerConfig плюс вариант claudeai-proxy только для вывода для servers, проксированных через claude.ai.

McpServerStatusConfig = (
    McpStdioServerConfig
    | McpSSEServerConfig
    | McpHttpServerConfig
    | McpSdkServerConfigStatus
    | McpClaudeAIProxyServerConfig
)

McpSdkServerConfigStatus - это сериализуемая форма McpSdkServerConfig с только полями type ("sdk") и name (str); встроенный instance опущен. McpClaudeAIProxyServerConfig имеет поля type ("claudeai-proxy"), url (str) и id (str).

McpStatusResponse

Ответ от ClaudeSDKClient.get_mcp_status(). Оборачивает список статусов сервера под ключом mcpServers.

class McpStatusResponse(TypedDict):
    mcpServers: list[McpServerStatus]

McpServerStatus

Статус подключенного MCP server, содержащийся в McpStatusResponse.

class McpServerStatus(TypedDict):
    name: str
    status: McpServerConnectionStatus  # "connected" | "failed" | "needs-auth" | "pending" | "disabled"
    serverInfo: NotRequired[McpServerInfo]
    error: NotRequired[str]
    config: NotRequired[McpServerStatusConfig]
    scope: NotRequired[str]
    tools: NotRequired[list[McpToolInfo]]

SdkPluginConfig

Конфигурация для загрузки plugins в SDK.

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

Пример:

plugins = [
    {"type": "local", "path": "./my-plugin"},
    {"type": "local", "path": "/absolute/path/to/plugin"},
]

Для полной информации о создании и использовании plugins см. Plugins.

Типы сообщений

Message

Тип объединения всех возможных сообщений.

Message = (
    UserMessage
    | AssistantMessage
    | SystemMessage
    | ResultMessage
    | StreamEvent
    | RateLimitEvent
)

UserMessage

Сообщение пользовательского ввода.

@dataclass
class UserMessage:
    content: str | list[ContentBlock]
    uuid: str | None = None
    parent_tool_use_id: str | None = None
    tool_use_result: dict[str, Any] | None = None

AssistantMessage

Сообщение ответа помощника с блоками содержимого.

@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str
    parent_tool_use_id: str | None = None
    error: AssistantMessageError | None = None
    usage: dict[str, Any] | None = None
    message_id: str | None = None

AssistantMessageError

Возможные типы ошибок для сообщений помощника.

AssistantMessageError = Literal[
    "authentication_failed",
    "billing_error",
    "rate_limit",
    "invalid_request",
    "server_error",
    "max_output_tokens",
    "unknown",
]

SystemMessage

Системное сообщение с метаданными.

@dataclass
class SystemMessage:
    subtype: str
    data: dict[str, Any]

ResultMessage

Финальное сообщение результата с информацией о стоимости и использовании.

@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    stop_reason: str | None = None
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None
    structured_output: Any = None
    model_usage: dict[str, Any] | None = None
    permission_denials: list[Any] | None = None
    deferred_tool_use: DeferredToolUse | None = None
    errors: list[str] | None = None
    api_error_status: int | None = None
    uuid: str | None = None

Словарь usage содержит следующие ключи, если они присутствуют:

Словарь model_usage отображает имена моделей на использование для каждой модели. Внутренние ключи словаря используют camelCase, потому что значение передается без изменений из базового процесса CLI, соответствуя типу TypeScript ModelUsage:

StreamEvent

Событие потока для частичных обновлений сообщений во время потоковой передачи. Получается только когда include_partial_messages=True в ClaudeAgentOptions. Импортируйте через from claude_agent_sdk.types import StreamEvent.

@dataclass
class StreamEvent:
    uuid: str
    session_id: str
    event: dict[str, Any]  # The raw Claude API stream event
    parent_tool_use_id: str | None = None

RateLimitEvent

Выдается при изменении статуса ограничения скорости (например, с "allowed" на "allowed_warning"). Используйте это для предупреждения пользователей перед достижением жесткого лимита или для отката, когда статус "rejected".

@dataclass
class RateLimitEvent:
    rate_limit_info: RateLimitInfo
    uuid: str
    session_id: str

RateLimitInfo

Состояние ограничения скорости, переносимое RateLimitEvent.

RateLimitStatus = Literal["allowed", "allowed_warning", "rejected"]
RateLimitType = Literal[
    "five_hour", "seven_day", "seven_day_opus", "seven_day_sonnet", "overage"
]


@dataclass
class RateLimitInfo:
    status: RateLimitStatus
    resets_at: int | None = None
    rate_limit_type: RateLimitType | None = None
    utilization: float | None = None
    overage_status: RateLimitStatus | None = None
    overage_resets_at: int | None = None
    overage_disabled_reason: str | None = None
    raw: dict[str, Any] = field(default_factory=dict)

TaskStartedMessage

Выдается при запуске фоновой задачи. Фоновая задача - это все, что отслеживается вне основного хода: фоновая команда Bash, наблюдение Monitor, подагент, порожденный через инструмент Agent, или удаленный агент. Поле task_type говорит вам, какой. Это именование не связано с переименованием инструмента Task на Agent.

@dataclass
class TaskStartedMessage(SystemMessage):
    task_id: str
    description: str
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    task_type: str | None = None

TaskUsage

Данные токенов и времени для фоновой задачи.

class TaskUsage(TypedDict):
    total_tokens: int
    tool_uses: int
    duration_ms: int

TaskProgressMessage

Выдается периодически с обновлениями прогресса для выполняющейся фоновой задачи.

@dataclass
class TaskProgressMessage(SystemMessage):
    task_id: str
    description: str
    usage: TaskUsage
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    last_tool_name: str | None = None

TaskNotificationMessage

Выдается при завершении, сбое или остановке фоновой задачи. Фоновые задачи включают команды Bash run_in_background, наблюдения Monitor и фоновые подагенты.

@dataclass
class TaskNotificationMessage(SystemMessage):
    task_id: str
    status: TaskNotificationStatus  # "completed" | "failed" | "stopped"
    output_file: str
    summary: str
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    usage: TaskUsage | None = None

Типы блоков содержимого

ContentBlock

Тип объединения всех блоков содержимого.

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

TextBlock

Блок содержимого текста.

@dataclass
class TextBlock:
    text: str

ThinkingBlock

Блок содержимого мышления (для моделей с возможностью мышления).

@dataclass
class ThinkingBlock:
    thinking: str
    signature: str

ToolUseBlock

Блок запроса использования инструмента.

@dataclass
class ToolUseBlock:
    id: str
    name: str
    input: dict[str, Any]

ToolResultBlock

Блок результата выполнения инструмента.

@dataclass
class ToolResultBlock:
    tool_use_id: str
    content: str | list[dict[str, Any]] | None = None
    is_error: bool | None = None

Типы ошибок

ClaudeSDKError

Базовый класс исключения для всех ошибок SDK.

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

CLINotFoundError

Вызывается, когда Claude Code CLI не установлен или не найден.

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

Вызывается, когда соединение с Claude Code не удается.

class CLIConnectionError(ClaudeSDKError):
    """Failed to connect to Claude Code."""

ProcessError

Вызывается, когда процесс Claude Code не работает.

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

Вызывается, когда разбор JSON не удается.

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

Типы hooks

Для полного руководства по использованию hooks с примерами и общими шаблонами см. Hooks guide.

HookEvent

Поддерживаемые типы событий hooks.

HookEvent = Literal[
    "PreToolUse",  # Called before tool execution
    "PostToolUse",  # Called after tool execution
    "PostToolUseFailure",  # Called when a tool execution fails
    "UserPromptSubmit",  # Called when user submits a prompt
    "Stop",  # Called when stopping execution
    "SubagentStop",  # Called when a subagent stops
    "PreCompact",  # Called before message compaction
    "Notification",  # Called for notification events
    "SubagentStart",  # Called when a subagent starts
    "PermissionRequest",  # Called when a permission decision is needed
]

HookCallback

Определение типа для функций обратного вызова hooks.

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

Параметры:

• input: Строго типизированный ввод hooks с дискриминированными объединениями на основе hook_event_name (см. HookInput)
• tool_use_id: Дополнительный идентификатор использования инструмента (для hooks, связанных с инструментами)
• context: Контекст hooks с дополнительной информацией

Возвращает HookJSONOutput, который может содержать:

• decision: "block" для блокировки действия
• systemMessage: Системное сообщение для добавления в стенограмму
• hookSpecificOutput: Вывод, специфичный для hooks

HookContext

Информация контекста, передаваемая в обратные вызовы hooks.

class HookContext(TypedDict):
    signal: Any | None  # Future: abort signal support

HookMatcher

Конфигурация для сопоставления hooks с определенными событиями или инструментами.

@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
    timeout: float | None = (
        None  # Timeout in seconds for all hooks in this matcher (default: 60)
    )

HookInput

Тип объединения всех типов ввода hooks. Фактический тип зависит от поля hook_event_name.

HookInput = (
    PreToolUseHookInput
    | PostToolUseHookInput
    | PostToolUseFailureHookInput
    | UserPromptSubmitHookInput
    | StopHookInput
    | SubagentStopHookInput
    | PreCompactHookInput
    | NotificationHookInput
    | SubagentStartHookInput
    | PermissionRequestHookInput
)

BaseHookInput

Базовые поля, присутствующие во всех типах ввода hooks.

class BaseHookInput(TypedDict):
    session_id: str
    transcript_path: str
    cwd: str
    permission_mode: NotRequired[str]

PreToolUseHookInput

Входные данные для событий hooks PreToolUse.

class PreToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PreToolUse"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_use_id: str
    agent_id: NotRequired[str]
    agent_type: NotRequired[str]

PostToolUseHookInput

Входные данные для событий hooks PostToolUse.

class PostToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PostToolUse"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_response: Any
    tool_use_id: str
    agent_id: NotRequired[str]
    agent_type: NotRequired[str]

PostToolUseFailureHookInput

Входные данные для событий hooks PostToolUseFailure. Вызывается, когда выполнение инструмента не удается.

class PostToolUseFailureHookInput(BaseHookInput):
    hook_event_name: Literal["PostToolUseFailure"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_use_id: str
    error: str
    is_interrupt: NotRequired[bool]
    agent_id: NotRequired[str]
    agent_type: NotRequired[str]

UserPromptSubmitHookInput

Входные данные для событий hooks UserPromptSubmit.

class UserPromptSubmitHookInput(BaseHookInput):
    hook_event_name: Literal["UserPromptSubmit"]
    prompt: str

StopHookInput

Входные данные для событий hooks Stop.

class StopHookInput(BaseHookInput):
    hook_event_name: Literal["Stop"]
    stop_hook_active: bool

SubagentStopHookInput

Входные данные для событий hooks SubagentStop.

class SubagentStopHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStop"]
    stop_hook_active: bool
    agent_id: str
    agent_transcript_path: str
    agent_type: str

PreCompactHookInput

Входные данные для событий hooks PreCompact.

class PreCompactHookInput(BaseHookInput):
    hook_event_name: Literal["PreCompact"]
    trigger: Literal["manual", "auto"]
    custom_instructions: str | None

NotificationHookInput

Входные данные для событий hooks Notification.

class NotificationHookInput(BaseHookInput):
    hook_event_name: Literal["Notification"]
    message: str
    title: NotRequired[str]
    notification_type: str

SubagentStartHookInput

Входные данные для событий hooks SubagentStart.

class SubagentStartHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStart"]
    agent_id: str
    agent_type: str

PermissionRequestHookInput

Входные данные для событий hooks PermissionRequest. Позволяет hooks программно обрабатывать решения разрешений.

class PermissionRequestHookInput(BaseHookInput):
    hook_event_name: Literal["PermissionRequest"]
    tool_name: str
    tool_input: dict[str, Any]
    permission_suggestions: NotRequired[list[Any]]

HookJSONOutput

Тип объединения для возвращаемых значений обратного вызова hooks.

HookJSONOutput = AsyncHookJSONOutput | SyncHookJSONOutput

SyncHookJSONOutput

Синхронный вывод hooks с полями управления и решения.

class SyncHookJSONOutput(TypedDict):
    # Control fields
    continue_: NotRequired[bool]  # Whether to proceed (default: True)
    suppressOutput: NotRequired[bool]  # Hide stdout from transcript
    stopReason: NotRequired[str]  # Message when continue is False

    # Decision fields
    decision: NotRequired[Literal["block"]]
    systemMessage: NotRequired[str]  # Warning message for user
    reason: NotRequired[str]  # Feedback for Claude

    # Hook-specific output
    hookSpecificOutput: NotRequired[HookSpecificOutput]

HookSpecificOutput

TypedDict, содержащий имя события hooks и поля, специфичные для события. Форма зависит от значения hookEventName. Для полных деталей доступных полей для каждого события hooks см. Control execution with hooks.

Дискриминированное объединение типов вывода, специфичных для события. Поле hookEventName определяет, какие поля действительны.

class PreToolUseHookSpecificOutput(TypedDict):
    hookEventName: Literal["PreToolUse"]
    permissionDecision: NotRequired[Literal["allow", "deny", "ask", "defer"]]
    permissionDecisionReason: NotRequired[str]
    updatedInput: NotRequired[dict[str, Any]]
    additionalContext: NotRequired[str]


class PostToolUseHookSpecificOutput(TypedDict):
    hookEventName: Literal["PostToolUse"]
    additionalContext: NotRequired[str]
    updatedToolOutput: NotRequired[Any]
    updatedMCPToolOutput: NotRequired[Any]


class PostToolUseFailureHookSpecificOutput(TypedDict):
    hookEventName: Literal["PostToolUseFailure"]
    additionalContext: NotRequired[str]


class UserPromptSubmitHookSpecificOutput(TypedDict):
    hookEventName: Literal["UserPromptSubmit"]
    additionalContext: NotRequired[str]


class NotificationHookSpecificOutput(TypedDict):
    hookEventName: Literal["Notification"]
    additionalContext: NotRequired[str]


class SubagentStartHookSpecificOutput(TypedDict):
    hookEventName: Literal["SubagentStart"]
    additionalContext: NotRequired[str]


class PermissionRequestHookSpecificOutput(TypedDict):
    hookEventName: Literal["PermissionRequest"]
    decision: dict[str, Any]


HookSpecificOutput = (
    PreToolUseHookSpecificOutput
    | PostToolUseHookSpecificOutput
    | PostToolUseFailureHookSpecificOutput
    | UserPromptSubmitHookSpecificOutput
    | NotificationHookSpecificOutput
    | SubagentStartHookSpecificOutput
    | PermissionRequestHookSpecificOutput
)

AsyncHookJSONOutput

Асинхронный вывод hooks, который откладывает выполнение hooks.

class AsyncHookJSONOutput(TypedDict):
    async_: Literal[True]  # Set to True to defer execution
    asyncTimeout: NotRequired[int]  # Timeout in milliseconds

Пример использования hooks

Этот пример регистрирует два hooks: один, который блокирует опасные команды bash, такие как rm -rf /, и другой, который регистрирует все использование инструментов для аудита. Hooks безопасности работает только на командах Bash (через matcher), в то время как hooks логирования работает на всех инструментах.

from claude_agent_sdk import query, ClaudeAgentOptions, 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 = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash", hooks=[validate_bash_command], timeout=120
            ),  # 2 min for validation
            HookMatcher(
                hooks=[log_tool_use]
            ),  # Applies to all tools (default 60s timeout)
        ],
        "PostToolUse": [HookMatcher(hooks=[log_tool_use])],
    }
)

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

Типы ввода/вывода инструментов

Документация схем ввода/вывода для всех встроенных инструментов Claude Code. Хотя Python SDK не экспортирует их как типы, они представляют структуру входов и выходов инструментов в сообщениях.

Agent

Имя инструмента: Agent (ранее Task, который все еще принимается как псевдоним)

Ввод:

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

Вывод:

{
    "result": str,  # Final result from the subagent
    "usage": dict | None,  # Token usage statistics
    "total_cost_usd": float | None,  # Estimated total cost in USD
    "duration_ms": int | None,  # Execution duration in milliseconds
}

AskUserQuestion

Имя инструмента: AskUserQuestion

Задает пользователю уточняющие вопросы во время выполнения. См. Handle approvals and user input для деталей использования.

Ввод:

{
    "questions": [  # Questions to ask the user (1-4 questions)
        {
            "question": str,  # The complete question to ask the user
            "header": str,  # Very short label displayed as a chip/tag (max 12 chars)
            "options": [  # The available choices (2-4 options)
                {
                    "label": str,  # Display text for this option (1-5 words)
                    "description": str,  # Explanation of what this option means
                }
            ],
            "multiSelect": bool,  # Set to true to allow multiple selections
        }
    ],
    "answers": dict | None,  # User answers populated by the permission system
}

Вывод:

{
    "questions": [  # The questions that were asked
        {
            "question": str,
            "header": str,
            "options": [{"label": str, "description": str}],
            "multiSelect": bool,
        }
    ],
    "answers": dict[str, str],  # Maps question text to answer string
    # Multi-select answers are comma-separated
}

Bash

Имя инструмента: Bash

Ввод:

{
    "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": 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
}

Monitor

Имя инструмента: Monitor

Запускает фоновый скрипт и доставляет каждую строку stdout в Claude как событие, чтобы он мог реагировать без опроса. Monitor следует тем же правилам разрешений, что и Bash. См. Monitor tool reference для поведения и доступности поставщика.

Ввод:

{
    "command": str,  # Shell script; each stdout line is an event, exit ends the watch
    "description": str,  # Short description shown in notifications
    "timeout_ms": int | None,  # Kill after this deadline (default 300000, max 3600000)
    "persistent": bool | None,  # Run for the lifetime of the session; stop with TaskStop
}

Вывод:

{
    "taskId": str,  # ID of the background monitor task
    "timeoutMs": int,  # Timeout deadline in milliseconds (0 when persistent)
    "persistent": bool | None,  # True when running until TaskStop or session end
}

Edit

Имя инструмента: Edit

Ввод:

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

Вывод:

{
    "message": str,  # Confirmation message
    "replacements": int,  # Number of replacements made
    "file_path": str,  # File path that was edited
}

Read

Имя инструмента: Read

Ввод:

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

Вывод (текстовые файлы):

{
    "content": str,  # File contents with line numbers
    "total_lines": int,  # Total number of lines in file
    "lines_returned": int,  # Lines actually returned
}

Вывод (изображения):

{
    "image": str,  # Base64 encoded image data
    "mime_type": str,  # Image MIME type
    "file_size": int,  # File size in bytes
}

Write

Имя инструмента: Write

Ввод:

{
    "file_path": str,  # The absolute path to the file to write
    "content": str,  # The content to write to the file
}

Вывод:

{
    "message": str,  # Success message
    "bytes_written": int,  # Number of bytes written
    "file_path": str,  # File path that was written
}

Glob

Имя инструмента: Glob

Ввод:

{
    "pattern": str,  # The glob pattern to match files against
    "path": str | None,  # The directory to search in (defaults to cwd)
}

Вывод:

{
    "matches": list[str],  # Array of matching file paths
    "count": int,  # Number of matches found
    "search_path": str,  # Search directory used
}

Grep

Имя инструмента: Grep

Ввод:

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

Вывод (режим content):

{
    "matches": [
        {
            "file": str,
            "line_number": int | None,
            "line": str,
            "before_context": list[str] | None,
            "after_context": list[str] | None,
        }
    ],
    "total_matches": int,
}

Вывод (режим files_with_matches):

{
    "files": list[str],  # Files containing matches
    "count": int,  # Number of files with matches
}

NotebookEdit

Имя инструмента: NotebookEdit

Ввод:

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

Вывод:

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

Имя инструмента: WebFetch

Ввод:

{
    "url": str,  # The URL to fetch content from
    "prompt": str,  # The prompt to run on the fetched content
}

Вывод:

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

Имя инструмента: WebSearch

Ввод:

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

Вывод:

{
    "results": [{"title": str, "url": str, "snippet": str, "metadata": dict | None}],
    "total_results": int,
    "query": str,
}

TodoWrite

Имя инструмента: TodoWrite

Ввод:

{
    "todos": [
        {
            "content": str,  # The task description
            "status": "pending" | "in_progress" | "completed",  # Task status
            "activeForm": str,  # Active form of the description
        }
    ]
}

Вывод:

{
    "message": str,  # Success message
    "stats": {"total": int, "pending": int, "in_progress": int, "completed": int},
}

BashOutput

Имя инструмента: BashOutput

Ввод:

{
    "bash_id": str,  # The ID of the background shell
    "filter": str | None,  # Optional regex to filter output lines
}

Вывод:

{
    "output": str,  # New output since last check
    "status": "running" | "completed" | "failed",  # Current shell status
    "exitCode": int | None,  # Exit code when completed
}

KillBash

Имя инструмента: KillBash

Ввод:

{
    "shell_id": str  # The ID of the background shell to kill
}

Вывод:

{
    "message": str,  # Success message
    "shell_id": str,  # ID of the killed shell
}

ExitPlanMode

Имя инструмента: ExitPlanMode

Ввод:

{
    "plan": str  # The plan to run by the user for approval
}

Вывод:

{
    "message": str,  # Confirmation message
    "approved": bool | None,  # Whether user approved the plan
}

ListMcpResources

Имя инструмента: ListMcpResources

Ввод:

{
    "server": str | None  # Optional server name to filter resources by
}

Вывод:

{
    "resources": [
        {
            "uri": str,
            "name": str,
            "description": str | None,
            "mimeType": str | None,
            "server": str,
        }
    ],
    "total": int,
}

ReadMcpResource

Имя инструмента: ReadMcpResource

Ввод:

{
    "server": str,  # The MCP server name
    "uri": str,  # The resource URI to read
}

Вывод:

{
    "contents": [
        {"uri": str, "mimeType": str | None, "text": str | None, "blob": str | None}
    ],
    "server": str,
}

Расширенные функции с ClaudeSDKClient

Построение интерфейса непрерывного разговора

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
)
import asyncio


class ConversationSession:
    """Maintains a single conversation session with Claude."""

    def __init__(self, options: ClaudeAgentOptions | None = None):
        self.client = ClaudeSDKClient(options)
        self.turn_count = 0

    async def start(self):
        await self.client.connect()
        print("Starting conversation session. Claude will remember context.")
        print(
            "Commands: 'exit' to quit, 'interrupt' to stop current task, 'new' for new session"
        )

        while True:
            user_input = input(f"\n[Turn {self.turn_count + 1}] You: ")

            if user_input.lower() == "exit":
                break
            elif user_input.lower() == "interrupt":
                await self.client.interrupt()
                print("Task interrupted!")
                continue
            elif user_input.lower() == "new":
                # Disconnect and reconnect for a fresh session
                await self.client.disconnect()
                await self.client.connect()
                self.turn_count = 0
                print("Started new conversation session (previous context cleared)")
                continue

            # Send message - the session retains all previous messages
            await self.client.query(user_input)
            self.turn_count += 1

            # Process response
            print(f"[Turn {self.turn_count}] Claude: ", end="")
            async for message in self.client.receive_response():
                if isinstance(message, AssistantMessage):
                    for block in message.content:
                        if isinstance(block, TextBlock):
                            print(block.text, end="")
            print()  # New line after response

        await self.client.disconnect()
        print(f"Conversation ended after {self.turn_count} turns.")


async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"], permission_mode="acceptEdits"
    )
    session = ConversationSession(options)
    await session.start()


# Example conversation:
# Turn 1 - You: "Create a file called hello.py"
# Turn 1 - Claude: "I'll create a hello.py file for you..."
# Turn 2 - You: "What's in that file?"
# Turn 2 - Claude: "The hello.py file I just created contains..." (remembers!)
# Turn 3 - You: "Add a main function to it"
# Turn 3 - Claude: "I'll add a main function to hello.py..." (knows which file!)

asyncio.run(main())

Использование hooks для модификации поведения

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    HookMatcher,
    HookContext,
)
import asyncio
from typing import Any


async def pre_tool_logger(
    input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
    """Log all tool usage before execution."""
    tool_name = input_data.get("tool_name", "unknown")
    print(f"[PRE-TOOL] About to use: {tool_name}")

    # You can modify or block the tool execution here
    if tool_name == "Bash" and "rm -rf" in str(input_data.get("tool_input", {})):
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": "Dangerous command blocked",
            }
        }
    return {}


async def post_tool_logger(
    input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
    """Log results after tool execution."""
    tool_name = input_data.get("tool_name", "unknown")
    print(f"[POST-TOOL] Completed: {tool_name}")
    return {}


async def user_prompt_modifier(
    input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
) -> dict[str, Any]:
    """Add context to user prompts."""
    original_prompt = input_data.get("prompt", "")

    # Add a timestamp as additional context for Claude to see
    from datetime import datetime

    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": f"[Submitted at {timestamp}] Original prompt: {original_prompt}",
        }
    }


async def main():
    options = ClaudeAgentOptions(
        hooks={
            "PreToolUse": [
                HookMatcher(hooks=[pre_tool_logger]),
                HookMatcher(matcher="Bash", hooks=[pre_tool_logger]),
            ],
            "PostToolUse": [HookMatcher(hooks=[post_tool_logger])],
            "UserPromptSubmit": [HookMatcher(hooks=[user_prompt_modifier])],
        },
        allowed_tools=["Read", "Write", "Bash"],
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("List files in current directory")

        async for message in client.receive_response():
            # Hooks will automatically log tool usage
            pass


asyncio.run(main())

Мониторинг прогресса в реальном времени

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    ToolUseBlock,
    ToolResultBlock,
    TextBlock,
)
import asyncio


async def monitor_progress():
    options = ClaudeAgentOptions(
        allowed_tools=["Write", "Bash"], permission_mode="acceptEdits"
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Create 5 Python files with different sorting algorithms")

        # Monitor progress in real-time
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, ToolUseBlock):
                        if block.name == "Write":
                            file_path = block.input.get("file_path", "")
                            print(f"Creating: {file_path}")
                    elif isinstance(block, ToolResultBlock):
                        print("Completed tool execution")
                    elif isinstance(block, TextBlock):
                        print(f"Claude says: {block.text[:100]}...")

        print("Task completed!")


asyncio.run(monitor_progress())

Пример использования

Базовые операции с файлами (используя query)

from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ToolUseBlock
import asyncio


async def create_project():
    options = ClaudeAgentOptions(
        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())

Обработка ошибок

from claude_agent_sdk import query, CLINotFoundError, ProcessError, CLIJSONDecodeError

try:
    async for message in query(prompt="Hello"):
        print(message)
except CLINotFoundError:
    print(
        "Claude Code CLI not found. Try reinstalling: pip install --force-reinstall claude-agent-sdk"
    )
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}")

Режим потоковой передачи с клиентом

from claude_agent_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())

Использование пользовательских инструментов с ClaudeSDKClient

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    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 = ClaudeAgentOptions(
        mcp_servers={"utils": my_server},
        allowed_tools=["mcp__utils__calculate", "mcp__utils__get_time"],
    )

    # Use ClaudeSDKClient for interactive tool usage
    async with ClaudeSDKClient(options=options) as client:
        await client.query("What's 123 * 456?")

        # Process calculation response
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Calculation: {block.text}")

        # Follow up with time query
        await client.query("What time is it now?")

        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(f"Time: {block.text}")


asyncio.run(main())

Конфигурация sandbox

SandboxSettings

Конфигурация для поведения sandbox. Используйте это для включения sandboxing команд и программной настройки ограничений сети.

class SandboxSettings(TypedDict, total=False):
    enabled: bool
    autoAllowBashIfSandboxed: bool
    excludedCommands: list[str]
    allowUnsandboxedCommands: bool
    network: SandboxNetworkConfig
    ignoreViolations: SandboxIgnoreViolations
    enableWeakerNestedSandbox: bool

Пример использования

from claude_agent_sdk import query, ClaudeAgentOptions, SandboxSettings

sandbox_settings: SandboxSettings = {
    "enabled": True,
    "autoAllowBashIfSandboxed": True,
    "network": {"allowLocalBinding": True},
}

async for message in query(
    prompt="Build and test my project",
    options=ClaudeAgentOptions(sandbox=sandbox_settings),
):
    print(message)

SandboxNetworkConfig

Конфигурация, специфичная для сети, для режима sandbox.

class SandboxNetworkConfig(TypedDict, total=False):
    allowedDomains: list[str]
    deniedDomains: list[str]
    allowManagedDomainsOnly: bool
    allowUnixSockets: list[str]
    allowAllUnixSockets: bool
    allowLocalBinding: bool
    allowMachLookup: list[str]
    httpProxyPort: int
    socksProxyPort: int

SandboxIgnoreViolations

Конфигурация для игнорирования определенных нарушений sandbox.

class SandboxIgnoreViolations(TypedDict, total=False):
    file: list[str]
    network: list[str]

Fallback разрешений для команд без sandbox

Когда allowUnsandboxedCommands включен, модель может запросить выполнение команд вне sandbox, установив dangerouslyDisableSandbox: True во входе инструмента. Эти запросы переходят к существующей системе разрешений, что означает, что ваш обработчик can_use_tool будет вызван, позволяя вам реализовать пользовательскую логику авторизации.

from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    HookMatcher,
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext,
)


async def can_use_tool(
    tool: str, input: dict, context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    # Check if the model is requesting to bypass the sandbox
    if tool == "Bash" and input.get("dangerouslyDisableSandbox"):
        # The model is requesting to run this command outside the sandbox
        print(f"Unsandboxed command requested: {input.get('command')}")

        if is_command_authorized(input.get("command")):
            return PermissionResultAllow()
        return PermissionResultDeny(
            message="Command not authorized for unsandboxed execution"
        )
    return PermissionResultAllow()


# Required: dummy hook keeps the stream open for can_use_tool
async def dummy_hook(input_data, tool_use_id, context):
    return {"continue_": True}


async def prompt_stream():
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Deploy my application"},
    }


async def main():
    async for message in query(
        prompt=prompt_stream(),
        options=ClaudeAgentOptions(
            sandbox={
                "enabled": True,
                "allowUnsandboxedCommands": True,  # Model can request unsandboxed execution
            },
            permission_mode="default",
            can_use_tool=can_use_tool,
            hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[dummy_hook])]},
        ),
    ):
        print(message)

Этот шаблон позволяет вам:

• Аудит запросов модели: Логируйте, когда модель запрашивает выполнение без sandbox
• Реализовать allowlists: Разрешайте только определенные команды работать без sandbox
• Добавить рабочие процессы одобрения: Требуйте явной авторизации для привилегированных операций

См. также

• SDK overview - Общие концепции SDK
• TypeScript SDK reference - Документация TypeScript SDK
• CLI reference - Интерфейс командной строки
• Common workflows - Пошаговые руководства