Agent SDK 參考 - Python

安裝

pip install claude-agent-sdk

在 `query()` 和 `ClaudeSDKClient` 之間選擇

Python SDK 提供了兩種與 Claude Code 互動的方式：

快速比較

功能	`query()`	`ClaudeSDKClient`
會話	每次建立新會話	重複使用相同會話
對話	單次交換	同一上下文中的多次交換
連接	自動管理	手動控制
串流輸入	✅ 支援	✅ 支援
中斷	❌ 不支援	✅ 支援
鉤子	❌ 不支援	✅ 支援
自訂工具	❌ 不支援	✅ 支援
繼續聊天	❌ 每次新會話	✅ 保持對話
使用案例	一次性任務	連續對話

何時使用 `query()`（每次新會話）

最適合：

不需要對話歷史的一次性問題
不需要來自先前交換的上下文的獨立任務
簡單的自動化指令碼
當您想每次都重新開始時

何時使用 `ClaudeSDKClient`（連續對話）

最適合：

繼續對話 - 當您需要 Claude 記住上下文時
後續問題 - 基於先前的回應進行構建
互動式應用程式 - 聊天介面、REPL
回應驅動邏輯 - 當下一步行動取決於 Claude 的回應時
會話控制 - 明確管理對話生命週期

函數

`query()`

為與 Claude Code 的每次互動建立新會話。返回一個非同步迭代器，在消息到達時產生消息。每次呼叫 query() 都會重新開始，不記得先前的互動。

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

參數

參數	類型	描述
`prompt`	`str \| AsyncIterable[dict]`	輸入提示，可以是字串或非同步可迭代物件（用於串流模式）
`options`	`ClaudeAgentOptions \| None`	可選的配置物件（如果為 None，預設為 `ClaudeAgentOptions()`）

返回一個 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 工具的裝飾器。

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

參數

參數	類型	描述
`name`	`str`	工具的唯一識別碼
`description`	`str`	工具功能的人類可讀描述
`input_schema`	`type \| dict[str, Any]`	定義工具輸入參數的架構（見下文）

輸入架構選項

簡單類型對應（推薦）：

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

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

`create_sdk_mcp_server()`

建立在 Python 應用程式內執行的進程內 MCP 伺服器。

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

參數

參數	類型	預設值	描述
`name`	`str`	-	伺服器的唯一識別碼
`version`	`str`	`"1.0.0"`	伺服器版本字串
`tools`	`list[SdkMcpTool[Any]] \| None`	`None`	使用 `@tool` 裝飾器建立的工具函數清單

返回一個 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"]
)

類別

`ClaudeSDKClient`

在多次交換中維持對話會話。 這是 TypeScript SDK 的 query() 函數內部工作方式的 Python 等效物 - 它建立一個可以繼續對話的客戶端物件。

主要功能

會話連續性：在多個 query() 呼叫中維持對話上下文
相同對話：Claude 記住會話中的先前消息
中斷支援：可以在執行中停止 Claude
明確生命週期：您控制會話何時開始和結束
回應驅動流程：可以對回應做出反應並發送後續消息
自訂工具和鉤子：支援自訂工具（使用 @tool 裝飾器建立）和鉤子

class ClaudeSDKClient:
    def __init__(self, options: ClaudeAgentOptions | 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

方法

方法	描述
`__init__(options)`	使用可選配置初始化客戶端
`connect(prompt)`	使用可選初始提示或消息串流連接到 Claude
`query(prompt, session_id)`	以串流模式發送新請求
`receive_messages()`	以非同步迭代器的形式接收來自 Claude 的所有消息
`receive_response()`	接收消息直到並包括 ResultMessage
`interrupt()`	發送中斷信號（僅在串流模式中有效）
`disconnect()`	從 Claude 斷開連接

上下文管理器支援

客戶端可以用作非同步上下文管理器以進行自動連接管理：

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 - Claude remembers 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": "text", "text": "Analyze the following data:"}
    await asyncio.sleep(0.5)
    yield {"type": "text", "text": "Temperature: 25°C"}
    await asyncio.sleep(0.5)
    yield {"type": "text", "text": "Humidity: 60%"}
    await asyncio.sleep(0.5)
    yield {"type": "text", "text": "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

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

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

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

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

        async for message in client.receive_response():
            # Process the new response
            pass

asyncio.run(interruptible_task())

範例 - 進階權限控制

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions
)

async def custom_permission_handler(
    tool_name: str,
    input_data: dict,
    context: dict
):
    """Custom logic for tool permissions."""

    # Block writes to system directories
    if tool_name == "Write" and input_data.get("file_path", "").startswith("/system/"):
        return {
            "behavior": "deny",
            "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 {
            "behavior": "allow",
            "updatedInput": {**input_data, "file_path": safe_path}
        }

    # Allow everything else
    return {
        "behavior": "allow",
        "updatedInput": 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`

使用 @tool 裝飾器建立的 SDK MCP 工具的定義。

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

屬性	類型	描述
`name`	`str`	工具的唯一識別碼
`description`	`str`	人類可讀的描述
`input_schema`	`type[T] \| dict[str, Any]`	輸入驗證的架構
`handler`	`Callable[[T], Awaitable[dict[str, Any]]]`	處理工具執行的非同步函數

`ClaudeAgentOptions`

Claude Code 查詢的配置資料類別。

@dataclass
class ClaudeAgentOptions:
    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)
    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)
    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
    fork_session: bool = False
    agents: dict[str, AgentDefinition] | None = None
    setting_sources: list[SettingSource] | None = None

屬性	類型	預設值	描述
`allowed_tools`	`list[str]`	`[]`	允許的工具名稱清單
`system_prompt`	`str \| SystemPromptPreset \| None`	`None`	系統提示配置。傳遞字串以取得自訂提示，或使用 `{"type": "preset", "preset": "claude_code"}` 以取得 Claude Code 的系統提示。新增 `"append"` 以擴展預設
`mcp_servers`	`dict[str, McpServerConfig] \| str \| Path`	`{}`	MCP 伺服器配置或配置檔案的路徑
`permission_mode`	`PermissionMode \| None`	`None`	工具使用的權限模式
`continue_conversation`	`bool`	`False`	繼續最近的對話
`resume`	`str \| None`	`None`	要恢復的會話 ID
`max_turns`	`int \| None`	`None`	最大對話輪次
`disallowed_tools`	`list[str]`	`[]`	不允許的工具名稱清單
`model`	`str \| None`	`None`	要使用的 Claude 模型
`permission_prompt_tool_name`	`str \| None`	`None`	權限提示的 MCP 工具名稱
`cwd`	`str \| Path \| None`	`None`	目前工作目錄
`settings`	`str \| None`	`None`	設定檔案的路徑
`add_dirs`	`list[str \| Path]`	`[]`	Claude 可以存取的其他目錄
`env`	`dict[str, str]`	`{}`	環境變數
`extra_args`	`dict[str, str \| None]`	`{}`	要直接傳遞給 CLI 的其他 CLI 引數
`max_buffer_size`	`int \| None`	`None`	緩衝 CLI stdout 時的最大位元組數
`debug_stderr`	`Any`	`sys.stderr`	已棄用 - 用於偵錯輸出的類似檔案的物件。改用 `stderr` 回呼
`stderr`	`Callable[[str], None] \| None`	`None`	用於 CLI 的 stderr 輸出的回呼函數
`can_use_tool`	`CanUseTool \| None`	`None`	工具權限回呼函數
`hooks`	`dict[HookEvent, list[HookMatcher]] \| None`	`None`	用於攔截事件的鉤子配置
`user`	`str \| None`	`None`	使用者識別碼
`include_partial_messages`	`bool`	`False`	包括部分消息串流事件
`fork_session`	`bool`	`False`	使用 `resume` 恢復時，分支到新的會話 ID 而不是繼續原始會話
`agents`	`dict[str, AgentDefinition] \| None`	`None`	以程式設計方式定義的子代理
`plugins`	`list[SdkPluginConfig]`	`[]`	從本機路徑載入自訂外掛程式。如需詳細資訊，請參閱外掛程式
`setting_sources`	`list[SettingSource] \| None`	`None`（無設定）	控制要載入哪些檔案系統設定。省略時，不會載入任何設定。注意：必須包括 `"project"` 以載入 CLAUDE.md 檔案

`SystemPromptPreset`

使用 Claude Code 的預設系統提示和可選新增項目的配置。

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

欄位	必需	描述
`type`	是	必須是 `"preset"` 以使用預設系統提示
`preset`	是	必須是 `"claude_code"` 以使用 Claude Code 的系統提示
`append`	否	要附加到預設系統提示的其他指示

`SettingSource`

控制 SDK 從哪些檔案系統配置來源載入設定。

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

值	描述	位置
`"user"`	全域使用者設定	`~/.claude/settings.json`
`"project"`	共用的專案設定（版本控制）	`.claude/settings.json`
`"local"`	本機專案設定（gitignored）	`.claude/settings.local.json`

預設行為

當 setting_sources 被省略或為 None 時，SDK 不會載入任何檔案系統設定。這為 SDK 應用程式提供了隔離。

為什麼使用 setting_sources？

載入所有檔案系統設定（舊版行為）：

# Load all settings like SDK v0.0.x did
from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="Analyze this code",
    options=ClaudeAgentOptions(
        setting_sources=["user", "project", "local"]  # Load all settings
    )
):
    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 (default behavior)
# No filesystem dependencies - setting_sources defaults to None
async for message in query(
    prompt="Review this PR",
    options=ClaudeAgentOptions(
        # setting_sources=None is the default, no need to specify
        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"],  # Required to load CLAUDE.md from project
        allowed_tools=["Read", "Write", "Edit"]
    )
):
    print(message)

設定優先順序

載入多個來源時，設定會以此優先順序（從高到低）合併：

本機設定（.claude/settings.local.json）
專案設定（.claude/settings.json）
使用者設定（~/.claude/settings.json）

程式設計選項（如 agents、allowed_tools）始終會覆蓋檔案系統設定。

`AgentDefinition`

以程式設計方式定義的子代理的配置。

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

欄位	必需	描述
`description`	是	何時使用此代理的自然語言描述
`tools`	否	允許的工具名稱陣列。如果省略，繼承所有工具
`prompt`	是	代理的系統提示
`model`	否	此代理的模型覆蓋。如果省略，使用主模型

`PermissionMode`

用於控制工具執行的權限模式。

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`

使用 create_sdk_mcp_server() 建立的 SDK MCP 伺服器的配置。

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

`McpServerConfig`

MCP 伺服器配置的聯合類型。

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

`SdkPluginConfig`

SDK 中載入外掛程式的配置。

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

欄位	類型	描述
`type`	`Literal["local"]`	必須是 `"local"`（目前僅支援本機外掛程式）
`path`	`str`	外掛程式目錄的絕對或相對路徑

範例：

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

如需建立和使用外掛程式的完整資訊，請參閱外掛程式。

消息類型

`Message`

所有可能消息的聯合類型。

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage

`UserMessage`

使用者輸入消息。

@dataclass
class UserMessage:
    content: str | list[ContentBlock]

`AssistantMessage`

具有內容區塊的助手回應消息。

@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str

`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
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | 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

鉤子類型

`HookEvent`

支援的鉤子事件類型。請注意，由於設定限制，Python SDK 不支援 SessionStart、SessionEnd 和 Notification 鉤子。

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`

鉤子回呼函數的類型定義。

HookCallback = Callable[
    [dict[str, Any], str | None, HookContext],
    Awaitable[dict[str, Any]]
]

參數：

input_data：鉤子特定的輸入資料（請參閱鉤子文件）
tool_use_id：可選的工具使用識別碼（適用於工具相關的鉤子）
context：具有其他資訊的鉤子上下文

返回一個可能包含以下內容的字典：

decision："block" 以阻止操作
systemMessage：要新增到文字記錄的系統消息
hookSpecificOutput：鉤子特定的輸出資料

`HookContext`

傳遞給鉤子回呼的上下文資訊。

@dataclass
class HookContext:
    signal: Any | None = None  # Future: abort signal support

`HookMatcher`

用於將鉤子與特定事件或工具相符的配置。

@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

鉤子使用範例

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

工具輸入/輸出類型

所有內建 Claude Code 工具的輸入/輸出架構文件。雖然 Python SDK 不會將這些匯出為類型，但它們代表消息中工具輸入和輸出的結構。

Task

工具名稱： 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,  # Total cost in USD
    "duration_ms": int | None         # Execution duration in milliseconds
}

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
}

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
}

輸出（內容模式）：

{
    "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):
        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 - Claude remembers all previous messages in this session
            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())

使用鉤子進行行為修改

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 timestamp to all prompts
    from datetime import datetime
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    return {
        'hookSpecificOutput': {
            'hookEventName': 'UserPromptSubmit',
            'updatedPrompt': f"[{timestamp}] {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
        files_created = []
        async for message in client.receive_messages():
            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(f"✅ Completed tool execution")
                    elif isinstance(block, TextBlock):
                        print(f"💭 Claude says: {block.text[:100]}...")

            # Check if we've received the final result
            if hasattr(message, 'subtype') and message.subtype in ['success', 'error']:
                print(f"\n🎯 Task completed!")
                break

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

使用客戶端的串流模式

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

另請參閱

Python SDK 指南 - 教學和範例
SDK 概述 - 一般 SDK 概念
TypeScript SDK 參考 - TypeScript SDK 文件
CLI 參考 - 命令列介面
常見工作流程 - 逐步指南

第一步

模型與定價

使用 Claude 建構

功能

工具

代理技能

Agent SDK

API 中的 MCP

Claude 在第三方平台上

提示工程

測試與評估

加強防護措施

管理和監控

​安裝

​在 query() 和 ClaudeSDKClient 之間選擇

​快速比較

​何時使用 query()（每次新會話）

​何時使用 ClaudeSDKClient（連續對話）

​函數

​query()

​參數

​返回

​範例 - 使用選項

​tool()

​參數

​輸入架構選項

​返回

​範例

​create_sdk_mcp_server()

​參數

​返回

​範例

​類別

​ClaudeSDKClient

​主要功能

​方法

​上下文管理器支援

​範例 - 繼續對話

​範例 - 使用 ClaudeSDKClient 進行串流輸入

​範例 - 使用中斷

​範例 - 進階權限控制

​類型

​SdkMcpTool

​ClaudeAgentOptions

​SystemPromptPreset

​SettingSource

​預設行為

​為什麼使用 setting_sources？

​設定優先順序

​AgentDefinition

​PermissionMode

​McpSdkServerConfig

​McpServerConfig

​McpStdioServerConfig

​McpSSEServerConfig

​McpHttpServerConfig

​SdkPluginConfig

​消息類型

​Message

​UserMessage

​AssistantMessage

​SystemMessage

​ResultMessage

​內容區塊類型

​ContentBlock

​TextBlock

​ThinkingBlock

​ToolUseBlock

​ToolResultBlock

​錯誤類型

​ClaudeSDKError

​CLINotFoundError

​CLIConnectionError

​ProcessError

​CLIJSONDecodeError

​鉤子類型

​HookEvent

​HookCallback

​HookContext

​HookMatcher

安裝

在 `query()` 和 `ClaudeSDKClient` 之間選擇

快速比較

何時使用 `query()`（每次新會話）

何時使用 `ClaudeSDKClient`（連續對話）

函數

`query()`

參數

返回

範例 - 使用選項

`tool()`

參數

輸入架構選項

返回

範例

`create_sdk_mcp_server()`

參數

返回

範例

類別

`ClaudeSDKClient`

主要功能

方法

上下文管理器支援

範例 - 繼續對話

範例 - 使用 ClaudeSDKClient 進行串流輸入

範例 - 使用中斷

範例 - 進階權限控制

類型

`SdkMcpTool`

`ClaudeAgentOptions`

`SystemPromptPreset`

`SettingSource`

預設行為

為什麼使用 setting_sources？

設定優先順序

`AgentDefinition`

`PermissionMode`

`McpSdkServerConfig`

`McpServerConfig`

`McpStdioServerConfig`

`McpSSEServerConfig`

`McpHttpServerConfig`

`SdkPluginConfig`

消息類型

`Message`

`UserMessage`

`AssistantMessage`

`SystemMessage`

`ResultMessage`

內容區塊類型

`ContentBlock`

`TextBlock`

`ThinkingBlock`

`ToolUseBlock`

`ToolResultBlock`

錯誤類型

`ClaudeSDKError`

`CLINotFoundError`

`CLIConnectionError`

`ProcessError`

`CLIJSONDecodeError`

鉤子類型

`HookEvent`

`HookCallback`

`HookContext`

`HookMatcher`

鉤子使用範例

工具輸入/輸出類型

Task

Bash

Edit

Read

Write

Glob

Grep

NotebookEdit

WebFetch

WebSearch

TodoWrite