Dec 17, 2025

Claude Agent SDK Python 架構深度解析：Anthropic 官方 Agent 開發框架

0. 為什麼要關注 Claude Agent SDK？

Claude Agent SDK 是 Anthropic 官方提供的 Python SDK，用於與 Claude Code CLI 進行程式化交互。與其他框架相比，它有幾個獨特的特點：

官方支持：直接由 Anthropic 維護，與 Claude 模型深度整合
輕量設計：核心代碼精簡，專注於與 Claude Code 的交互
雙模式交互：支持一次性查詢和交互式客戶端兩種模式
進程內 MCP：無需外部進程，直接在 Python 中定義和運行 MCP 工具
完整的權限控制：Hook 系統和權限回調支持細粒度控制

整體架構可以用一張圖表示：

flowchart TD
  classDef pad fill:transparent,stroke:transparent,color:transparent;

  subgraph UserAPI["用戶 API 層"]
    direction TB
    api_pad[" "]:::pad
    Query["query()"]
    Client["ClaudeSDKClient"]
  end

  subgraph Internal["內部實現層"]
    direction TB
    int_pad[" "]:::pad
    InternalClient["InternalClient"]
    QueryClass["Query"]
    MessageParser["MessageParser"]
  end

  subgraph Transport["傳輸層"]
    direction TB
    trans_pad[" "]:::pad
    TransportABC["Transport (ABC)"]
    SubprocessCLI["SubprocessCLITransport"]
  end

  subgraph CLI["Claude Code CLI"]
    direction TB
    cli_pad[" "]:::pad
    Binary["打包的二進制文件"]
  end

  UserAPI --> Internal
  Internal --> Transport
  Transport --> CLI

接下來深入探討 Claude Agent SDK 的核心架構。

1. 專案結構與規模

Claude Agent SDK 是一個精簡的 SDK，核心代碼約 2,000 行：

claude-agent-sdk-python/
├── src/claude_agent_sdk/
│   ├── __init__.py                 # 主入口和公共 API 導出
│   ├── client.py                   # ClaudeSDKClient - 交互式客戶端
│   ├── query.py                    # query() - 一次性查詢函數
│   ├── types.py                    # 類型定義 (754 行)
│   ├── _errors.py                  # 錯誤類定義
│   ├── _version.py                 # 版本信息
│   ├── py.typed                    # PEP 561 類型標記
│   ├── _bundled/                   # 打包的 Claude Code CLI 二進制文件
│   └── _internal/
│       ├── __init__.py
│       ├── client.py               # InternalClient - 內部客戶端實現
│       ├── message_parser.py       # 消息解析和類型轉換
│       ├── query.py                # Query - 控制協議處理
│       └── transport/
│           ├── __init__.py         # Transport 抽象基類
│           └── subprocess_cli.py   # 子進程 CLI 傳輸實現
├── tests/                          # 單元測試套件 (12 個測試文件)
├── e2e-tests/                      # 端到端測試
├── examples/                       # 使用示例 (16 個示例文件)
└── scripts/                        # 構建和發布腳本

版本資訊：

當前版本：0.1.17
許可證：MIT
Python 要求：>= 3.10

2. 核心架構：分層設計

2.1 三層架構

Claude Agent SDK 採用清晰的三層架構：

flowchart TD
  subgraph Layer1["用戶 API 層"]
    Query["query()"]
    Client["ClaudeSDKClient"]
    Options["ClaudeAgentOptions"]
    Tool["@tool decorator"]
  end

  subgraph Layer2["內部實現層"]
    InternalClient["InternalClient"]
    QueryClass["Query"]
    MessageParser["MessageParser"]
    HookHandler["Hook 處理"]
    PermHandler["權限管理"]
  end

  subgraph Layer3["傳輸層"]
    Transport["Transport (ABC)"]
    SubprocessCLI["SubprocessCLITransport"]
    IO["I/O 流處理"]
  end

  subgraph External["外部進程"]
    CLI["Claude Code CLI"]
  end

  Layer1 --> Layer2
  Layer2 --> Layer3
  Layer3 --> External

2.2 核心類別

類/函數	位置	用途
`query()`	`query.py`	一次性、無狀態的異步查詢函數
`ClaudeSDKClient`	`client.py`	雙向、有狀態的交互式客戶端
`ClaudeAgentOptions`	`types.py`	配置選項數據類
`@tool`	`__init__.py`	MCP 工具裝飾器
`create_sdk_mcp_server()`	`__init__.py`	創建進程內 MCP 服務器

2.3 消息類型系統

SDK 定義了豐富的消息類型：

# 消息類型
UserMessage        # 用戶輸入消息
AssistantMessage   # Claude 響應消息
SystemMessage      # 系統消息
ResultMessage      # 結果和成本信息
StreamEvent        # 流式部分消息

# 內容塊類型
TextBlock          # 文本內容
ThinkingBlock      # 思考過程 (Extended Thinking)
ToolUseBlock       # 工具調用
ToolResultBlock    # 工具結果

消息流程示意：

flowchart LR
  User["UserMessage"]
  Assistant["AssistantMessage"]
  Tool["ToolUseBlock"]
  Result["ToolResultBlock"]
  Final["ResultMessage"]

  User --> Assistant
  Assistant --> Tool
  Tool --> Result
  Result --> Assistant
  Assistant --> Final

2.4 錯誤類型層次

ClaudeSDKError (基類)
├── CLIConnectionError
│   └── CLINotFoundError
├── ProcessError
├── CLIJSONDecodeError
└── MessageParseError

3. 兩種交互模式

3.1 query() - 一次性查詢

最簡單的使用方式，適合單次問答：

import anyio
from claude_agent_sdk import query

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

anyio.run(main)

函數簽名：

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

執行流程：

flowchart LR
  User["用戶"] --> Prompt["提示詞"]
  Prompt --> Query["query()"]
  Query --> Transport["傳輸層"]
  Transport --> CLI["Claude Code CLI"]
  CLI --> Response["流式響應"]
  Response --> Iterator["消息迭代器"]
  Iterator --> User

特點：

單向、無狀態
所有消息預先發送，然後接收所有響應
適合簡單問題、批處理、腳本自動化

使用場景：

一次性問題
CI/CD 管道中的代碼生成
批量處理任務

3.2 ClaudeSDKClient - 交互式客戶端

支持多輪對話和動態控制：

from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def main():
    async with ClaudeSDKClient() as client:
        # 第一個查詢
        await client.query("What is Python?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # 後續查詢（保持對話歷史）
        await client.query("Tell me more about its libraries")
        async for msg in client.receive_response():
            # 處理響應...
            pass

anyio.run(main)

核心方法：

class ClaudeSDKClient:
    # 基本交互
    async def connect(prompt: str | None = None) -> None
    async def query(prompt: str, session_id: str = "default") -> None
    async def receive_response() -> AsyncIterator[Message]
    async def receive_messages() -> AsyncIterator[Message]

    # 高級控制
    async def interrupt() -> None                           # 中斷當前操作
    async def set_permission_mode(mode: str) -> None        # 設置權限模式
    async def set_model(model: str | None = None) -> None   # 切換模型
    async def rewind_files(user_message_id: str) -> None    # 回滾文件
    async def get_server_info() -> dict[str, Any] | None    # 獲取服務器信息

執行流程：

flowchart TD
  Connect["connect()"]
  Query1["query('問題1')"]
  Receive1["receive_response()"]
  Query2["query('問題2')"]
  Receive2["receive_response()"]
  Close["close()"]

  Connect --> Query1 --> Receive1
  Receive1 --> Query2 --> Receive2
  Receive2 --> Close

特點：

雙向通信
有狀態對話（保持對話歷史）
支持中斷操作
支持動態配置更改
支持文件檢查點和回滾

4. 配置選項詳解

4.1 ClaudeAgentOptions

from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    # 基本配置
    system_prompt="You are a Python expert",
    model="claude-opus-4-1-20250805",
    cwd="/path/to/project",

    # 工具配置
    allowed_tools=["Read", "Write", "Bash"],
    mcp_servers={"calc": calculator_server},

    # 權限控制
    permission_mode="acceptEdits",
    can_use_tool=my_permission_callback,

    # 限制
    max_turns=10,
    max_budget_usd=1.0,

    # 高級功能
    hooks={...},
    sandbox=SandboxSettings(...),
    enable_file_checkpointing=True,
)

4.2 主要選項說明

選項	類型	用途
`system_prompt`	`str`	系統提示詞，定義 Claude 的行為
`model`	`str`	模型選擇 (claude-sonnet-4-20250514, claude-opus-4-1-20250805 等)
`allowed_tools`	`list[str]`	允許 Claude 使用的工具列表
`mcp_servers`	`dict`	MCP 服務器配置
`permission_mode`	`str`	權限模式：default, acceptEdits, bypassPermissions
`max_turns`	`int`	最大對話輪數
`max_budget_usd`	`float`	最大花費預算（美元）
`hooks`	`dict`	Hook 配置
`can_use_tool`	`callable`	工具權限回調函數
`sandbox`	`SandboxSettings`	沙箱配置
`cwd`	`str\|Path`	工作目錄
`enable_file_checkpointing`	`bool`	啟用文件檢查點

4.3 權限模式

模式	描述
`default`	默認模式，需要確認危險操作
`acceptEdits`	自動接受文件編輯
`bypassPermissions`	跳過所有權限檢查（謹慎使用）

5. MCP 工具系統

5.1 進程內 MCP 服務器

Claude Agent SDK 的一大亮點是支持進程內 MCP 服務器，無需啟動外部進程：

from claude_agent_sdk import tool, create_sdk_mcp_server

# 定義工具
@tool("add", "Add two numbers", {"a": float, "b": float})
async def add_numbers(args: dict[str, Any]) -> dict[str, Any]:
    result = args["a"] + args["b"]
    return {"content": [{"type": "text", "text": f"Result: {result}"}]}

@tool("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply_numbers(args: dict[str, Any]) -> dict[str, Any]:
    result = args["a"] * args["b"]
    return {"content": [{"type": "text", "text": f"Result: {result}"}]}

# 創建服務器
calculator = create_sdk_mcp_server(
    name="calculator",
    tools=[add_numbers, multiply_numbers]
)

# 使用服務器
options = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=[
        "mcp__calc__add",
        "mcp__calc__multiply",
    ]
)

5.2 @tool 裝飾器

@tool(
    name: str,              # 工具名稱
    description: str,       # 工具描述
    input_schema: dict,     # 輸入參數 schema
)
async def tool_function(args: dict[str, Any]) -> dict[str, Any]:
    """
    返回格式：
    {
        "content": [
            {"type": "text", "text": "結果文本"},
            {"type": "image", "data": "base64...", "mimeType": "image/png"}
        ]
    }
    """
    pass

5.3 進程內 MCP 的優勢

flowchart LR
  subgraph Traditional["傳統 MCP"]
    SDK1["SDK"] --> IPC["IPC"] --> Server["外部進程"]
  end

  subgraph InProcess["進程內 MCP"]
    SDK2["SDK"] --> Direct["直接調用"] --> Tools["工具函數"]
  end

傳統 MCP	進程內 MCP
需要外部進程	單一 Python 進程
IPC 延遲	無 IPC 開銷
進程管理複雜	簡化部署
調試困難	容易調試
獨立狀態	直接訪問應用狀態

6. Hook 系統

6.1 支持的 Hook 事件

事件	觸發時機	用途
`PreToolUse`	工具使用前	權限控制、參數驗證
`PostToolUse`	工具執行後	處理輸出、日誌記錄
`UserPromptSubmit`	用戶提示提交時	輸入過濾、預處理
`Stop`	停止事件	清理資源
`SubagentStop`	子代理停止	子代理管理
`PreCompact`	上下文壓縮前	自定義壓縮邏輯

6.2 Hook 執行流程

flowchart TD
  User["用戶輸入"]
  Submit["UserPromptSubmit Hook"]
  Process["Claude 處理"]
  PreTool["PreToolUse Hook"]
  Tool["工具執行"]
  PostTool["PostToolUse Hook"]
  Response["響應"]

  User --> Submit --> Process
  Process --> PreTool --> Tool --> PostTool --> Process
  Process --> Response

6.3 Hook 配置示例

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher

# 定義 Hook 函數
async def check_bash_command(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """檢查 Bash 命令的安全性"""
    if input_data["tool_name"] == "Bash":
        command = input_data["tool_input"].get("command", "")

        # 檢查危險命令
        dangerous_patterns = ["rm -rf", "sudo", "chmod 777"]
        for pattern in dangerous_patterns:
            if pattern in command:
                return {
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": "deny",
                        "permissionDecisionReason": f"Blocked: contains '{pattern}'"
                    }
                }

    # 允許執行
    return {}

async def log_tool_result(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """記錄工具執行結果"""
    print(f"Tool: {input_data['tool_name']}")
    print(f"Result: {input_data.get('tool_output', {})}")
    return {}

# 配置 Hooks
options = ClaudeAgentOptions(
    allowed_tools=["Bash", "Read", "Write"],
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Bash", hooks=[check_bash_command])
        ],
        "PostToolUse": [
            HookMatcher(matcher="*", hooks=[log_tool_result])
        ]
    }
)

6.4 Hook 返回值

# PreToolUse Hook 可以返回權限決策
return {
    "hookSpecificOutput": {
        "hookEventName": "PreToolUse",
        "permissionDecision": "allow" | "deny" | "ask",
        "permissionDecisionReason": "原因說明",
        # 可選：修改工具輸入
        "updatedInput": {"modified": "value"}
    }
}

# PostToolUse Hook 可以修改輸出
return {
    "hookSpecificOutput": {
        "hookEventName": "PostToolUse",
        "output": "modified output"
    }
}

7. 工具權限回調

除了 Hook 系統，SDK 還提供了專門的工具權限回調：

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

async def my_permission_callback(
    tool_name: str,
    input_data: dict,
    context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    """細粒度的工具權限控制"""

    # 保護系統文件
    if tool_name == "Write":
        file_path = input_data.get("file_path", "")
        if file_path.startswith("/etc/") or file_path.startswith("/sys/"):
            return PermissionResultDeny(
                message="System files are protected"
            )

    # 限制 Bash 命令
    if tool_name == "Bash":
        command = input_data.get("command", "")
        if "curl" in command or "wget" in command:
            return PermissionResultDeny(
                message="Network commands are not allowed"
            )

    # 修改輸入（例如重定向文件路徑）
    if tool_name == "Write" and input_data.get("file_path", "").startswith("/tmp/"):
        modified_input = input_data.copy()
        modified_input["file_path"] = "/safe/dir/" + input_data["file_path"].split("/")[-1]
        return PermissionResultAllow(updated_input=modified_input)

    # 默認允許
    return PermissionResultAllow()

options = ClaudeAgentOptions(
    can_use_tool=my_permission_callback,
    permission_prompt_tool_name="stdio"
)

8. 控制協議

8.1 子進程通信架構

SDK 通過 JSON 控制協議與 Claude Code CLI 進行雙向通信：

flowchart TD
  subgraph SDK["Python SDK"]
    Query["Query/Client"]
    Transport["SubprocessCLITransport"]
  end

  subgraph Protocol["JSON 控制協議"]
    Stdin["stdin (SDK → CLI)"]
    Stdout["stdout (CLI → SDK)"]
  end

  subgraph CLI["Claude Code CLI"]
    Process["子進程"]
  end

  Query --> Transport
  Transport --> Stdin --> Process
  Process --> Stdout --> Transport

8.2 協議消息類型

請求類型（SDK → CLI）：

# 初始化請求
SDKControlInitializeRequest = {
    "type": "initialize",
    "options": {...}
}

# 權限響應
SDKControlPermissionRequest = {
    "type": "permission_response",
    "allowed": True,
    "updated_input": {...}  # 可選
}

# 中斷信號
SDKControlInterruptRequest = {
    "type": "interrupt"
}

# Hook 回調響應
SDKHookCallbackRequest = {
    "type": "hook_callback",
    "output": {...}
}

響應類型（CLI → SDK）：

# 消息流
Message = {
    "type": "assistant_message" | "user_message" | "system_message",
    "content": [...]
}

# 控制請求
SDKControlRequest = {
    "type": "permission_request" | "hook_callback",
    ...
}

# 結果
ResultMessage = {
    "type": "result",
    "cost_usd": 0.05,
    "tokens_used": {...}
}

8.3 通信流程示例

sequenceDiagram
  participant SDK as Python SDK
  participant CLI as Claude Code CLI

  SDK->>CLI: InitializeRequest (options)
  CLI->>SDK: InitializeResponse

  SDK->>CLI: UserMessage (prompt)
  CLI->>SDK: AssistantMessage (thinking...)
  CLI->>SDK: ToolUseBlock (Bash)

  CLI->>SDK: PermissionRequest (Bash)
  SDK->>CLI: PermissionResponse (allowed: true)

  CLI->>SDK: ToolResultBlock (output)
  CLI->>SDK: AssistantMessage (final response)
  CLI->>SDK: ResultMessage (cost, tokens)

9. 文件檢查點與回滾

9.1 啟用文件檢查點

options = ClaudeAgentOptions(
    enable_file_checkpointing=True,
    extra_args={"replay-user-messages": None}
)

9.2 使用檢查點回滾

async with ClaudeSDKClient(options=options) as client:
    # 第一次交互 - 記錄檢查點 ID
    await client.query("Create a new file called test.py")
    checkpoint_id = None
    async for msg in client.receive_response():
        if isinstance(msg, UserMessage) and msg.uuid:
            checkpoint_id = msg.uuid
        # 處理響應...

    # 第二次交互 - 修改文件
    await client.query("Add a function to test.py")
    async for msg in client.receive_response():
        # 處理響應...

    # 回滾到第一個檢查點
    if checkpoint_id:
        await client.rewind_files(checkpoint_id)
        print("Files rolled back to checkpoint")

9.3 檢查點流程

flowchart TD
  Q1["Query 1: 創建文件"]
  CP1["Checkpoint 1"]
  Q2["Query 2: 修改文件"]
  CP2["Checkpoint 2"]
  Rewind["rewind_files(CP1)"]
  State1["狀態回滾到 CP1"]

  Q1 --> CP1 --> Q2 --> CP2
  CP2 --> Rewind --> State1

10. 完整示例

10.1 快速開始

import anyio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="Explain Python decorators"):
        print(message)

anyio.run(main)

10.2 帶配置的查詢

from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a Python expert. Be concise.",
        allowed_tools=["Read", "Bash"],
        permission_mode="acceptEdits",
        max_turns=5,
        model="claude-sonnet-4-20250514"
    )

    async for message in query(
        prompt="List all Python files in the current directory",
        options=options
    ):
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    print(block.text)

anyio.run(main)

10.3 帶 MCP 工具的交互式客戶端

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    tool,
    create_sdk_mcp_server,
    AssistantMessage,
    TextBlock
)

# 定義計算器工具
@tool("calculate", "Perform calculations", {"expression": str})
async def calculate(args: dict) -> dict:
    try:
        result = eval(args["expression"])  # 注意：生產環境應使用安全的解析器
        return {"content": [{"type": "text", "text": f"Result: {result}"}]}
    except Exception as e:
        return {"content": [{"type": "text", "text": f"Error: {e}"}]}

# 創建 MCP 服務器
calculator_server = create_sdk_mcp_server(
    name="calculator",
    tools=[calculate]
)

async def main():
    options = ClaudeAgentOptions(
        mcp_servers={"calc": calculator_server},
        allowed_tools=["mcp__calc__calculate"],
        system_prompt="You have access to a calculator. Use it for math."
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("What is 123 * 456 + 789?")

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

anyio.run(main)

10.4 帶 Hook 的安全執行

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    HookMatcher
)

# 安全檢查 Hook
async def security_check(input_data, tool_use_id, context):
    tool_name = input_data.get("tool_name", "")
    tool_input = input_data.get("tool_input", {})

    # 記錄所有工具調用
    print(f"[AUDIT] Tool: {tool_name}, Input: {tool_input}")

    # 阻止危險操作
    if tool_name == "Bash":
        command = tool_input.get("command", "")
        if any(danger in command for danger in ["rm", "sudo", ">"]):
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": "Dangerous command blocked"
                }
            }

    return {}

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Bash", "Read", "Write"],
        hooks={
            "PreToolUse": [
                HookMatcher(matcher="*", hooks=[security_check])
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("List files and show system info")

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

anyio.run(main)

11. 設計特點總結

特性	描述	優勢
異步優先	基於 anyio 的跨運行時支持	支持 asyncio 和 trio
類型安全	完整的 Python 類型注解	IDE 支持、編譯時檢查
分層架構	用戶 API / 內部實現 / 傳輸層	關注點分離、可測試
雙模式交互	query() 和 ClaudeSDKClient	靈活適應不同場景
進程內 MCP	無需外部進程	簡化部署、降低延遲
Hook 系統	細粒度的生命週期控制	安全審計、自定義邏輯
權限回調	工具級別的權限控制	精細安全控制
文件檢查點	支持回滾	安全實驗、錯誤恢復

12. 與其他框架比較

特性	Claude Agent SDK	LangGraph	Google ADK
核心理念	簡潔優先	圖優先	代碼優先
狀態管理	CLI 子進程	BSP/Channel	無狀態 Runner
工具系統	進程內 MCP	LangChain Tools	50+ 內建工具
多代理	子代理 Hook	StateGraph	代理樹
持久化	文件檢查點	Checkpointer	SessionService
代碼量	~2,000 行	~15,000 行	~79,500 行

13. 結語

Claude Agent SDK 是一個精簡而強大的官方 SDK，特別適合：

需要官方支持：由 Anthropic 維護，與 Claude 模型深度整合
快速原型開發：API 簡潔，開箱即用
安全敏感場景：完整的 Hook 和權限控制系統
進程內工具開發：無需管理外部 MCP 進程

如果你正在構建基於 Claude 的 AI Agent 應用，Claude Agent SDK 是一個非常值得考慮的選擇。