Claude Agent SDK 源码与实战（一）：整体架构与快速上手

用户代码
   ↓
┌───────────────────────────────────────────────┐
│  公共 API 层（claude_agent_sdk/）              │
│  ├── query()       → 简单的一次性查询          │
│  ├── ClaudeSDKClient → 双向交互式对话客户端     │
│  └── types.py       → 所有类型定义              │
└───────────────────────────────────────────────┘
   ↓
┌───────────────────────────────────────────────┐
│  内部实现层（_internal/）                       │
│  ├── InternalClient → 核心客户端实现           │
│  ├── Query          → 控制协议 + 消息路由      │
│  └── message_parser → 消息解析器              │
└───────────────────────────────────────────────┘
   ↓
┌───────────────────────────────────────────────┐
│  传输层（_internal/transport/）                │
│  └── SubprocessCLITransport → 子进程通信       │
└───────────────────────────────────────────────┘
   ↓
Claude Code CLI（子进程）
   ↓
Claude API

claude-agent-sdk-python/
  pyproject.toml
  README.md
  CHANGELOG.md
  CLAUDE.md

  src/claude_agent_sdk/
    __init__.py          # 对外暴露的主入口
    client.py            # ClaudeSDKClient 实现
    query.py             # query() 函数
    types.py             # 所有类型定义
    _errors.py           # 错误体系
    _version.py          # 版本信息
    py.typed

    _internal/
      __init__.py
      client.py          # InternalClient 实现
      query.py           # Query 类（控制协议核心）
      message_parser.py  # 消息解析器
      transport/
        __init__.py
        subprocess_cli.py # 子进程传输层

  examples/
    quick_start.py       # 快速上手示例
    agents.py            # 多轮对话示例
    streaming_mode.py    # 流式输出示例
    mcp_calculator.py    # MCP 工具示例
    ...

  tests/
    test_client.py
    test_types.py
    test_message_parser.py
    ...

  e2e-tests/
    test_agents_and_settings.py
    test_hooks.py
    ...

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

from claude_agent_sdk import query

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

class ClaudeSDKClient:
    async def connect(self, prompt: str | AsyncIterable[dict] | None = None)
    async def send_user_message(self, content: str | list, parent_tool_use_id: str | None = None)
    async def messages() -> AsyncIterator[Message]
    async def disconnect()

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

client = ClaudeSDKClient(options=ClaudeAgentOptions(...))
await client.connect("Hello!")

async for msg in client.messages():
    print(msg)
    if need_followup:
        await client.send_user_message("Tell me more")

await client.disconnect()

@dataclass
class UserMessage:
    """用户消息"""
    content: str | list[ContentBlock]
    parent_tool_use_id: str | None = None

@dataclass
class AssistantMessage:
    """助手消息，包含内容块"""
    content: list[ContentBlock]
    model: str
    parent_tool_use_id: str | None = None
    error: AssistantMessageError | None = None

@dataclass
class SystemMessage:
    """系统消息，包含元数据"""
    subtype: str
    data: dict[str, Any]

@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

# 消息联合类型
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent

@dataclass
class TextBlock:
    """文本内容块"""
    text: str

@dataclass
class ThinkingBlock:
    """思考内容块（Claude 的推理过程）"""
    thinking: str
    signature: str

@dataclass
class ToolUseBlock:
    """工具调用块"""
    id: str
    name: str
    input: dict[str, Any]

@dataclass
class ToolResultBlock:
    """工具结果块"""
    tool_use_id: str
    content: str | list[dict] | None = None
    is_error: bool | None = None

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

@dataclass
class ClaudeAgentOptions:
    # 工具相关
    tools: list[str] | ToolsPreset | None = None
    allowed_tools: list[str] = field(default_factory=list)
    disallowed_tools: list[str] = field(default_factory=list)
    
    # 提示词
    system_prompt: str | SystemPromptPreset | None = None
    
    # MCP 服务器
    mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
    
    # 权限与安全
    permission_mode: PermissionMode | None = None
    can_use_tool: CanUseTool | None = None
    sandbox: SandboxSettings | None = None
    
    # 模型配置
    model: str | None = None
    fallback_model: str | None = None
    max_turns: int | None = None
    max_budget_usd: float | None = None
    
    # Hook 系统
    hooks: dict[HookEvent, list[HookMatcher]] | None = None
    
    # 高级选项
    cwd: str | Path | None = None
    cli_path: str | Path | None = None
    include_partial_messages: bool = False
    agents: dict[str, AgentDefinition] | None = None
    output_format: dict[str, Any] | None = None  # 结构化输出
    ...

# SDK → CLI 的控制请求
class SDKControlRequest(TypedDict):
    type: Literal["control_request"]
    request_id: str
    request: (
        SDKControlInterruptRequest      # 中断
        | SDKControlPermissionRequest   # 工具权限请求
        | SDKControlInitializeRequest   # 初始化
        | SDKHookCallbackRequest        # Hook 回调
        | SDKControlMcpMessageRequest   # MCP 消息
    )

# CLI → SDK 的控制响应
class SDKControlResponse(TypedDict):
    type: Literal["control_response"]
    response: ControlResponse | ControlErrorResponse

class Query:
    async def initialize(self) -> dict | None:
        """初始化控制协议，注册 hooks"""

    async def _read_messages(self):
        """从 transport 读取消息并路由"""

    async def _handle_control_request(self, request):
        """处理 CLI 发来的控制请求"""

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

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

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

PermissionResult = PermissionResultAllow | PermissionResultDeny

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

permission_request: SDKControlPermissionRequest = request_data

context = ToolPermissionContext(
    signal=None,
    suggestions=permission_request.get("permission_suggestions", []) or [],
)

response = await self.can_use_tool(
    permission_request["tool_name"],
    permission_request["input"],
    context,
)

if isinstance(response, PermissionResultAllow):
    # behavior = allow，必要时可以修改输入 / 权限
elif isinstance(response, PermissionResultDeny):
    # behavior = deny，可以附带 message、interrupt 标记

if total_length > _CMD_LENGTH_LIMIT:
    # 把 prompt 写入临时文件
    temp_file = tempfile.NamedTemporaryFile(...)
    temp_file.write(prompt.encode())
    cmd.extend(["--prompt-file", temp_file.name])
else:
    cmd.extend(["--prompt", prompt])

def parse_message(data: dict[str, Any]) -> Message:
    message_type = data.get("type")

    match message_type:
        case "user":
            return UserMessage(...)

        case "assistant":
            return AssistantMessage(...)

        case "system":
            return SystemMessage(...)

        case "result":
            return ResultMessage(...)

        case "stream_event":
            return StreamEvent(...)

options = ClaudeAgentOptions(
    mcp_servers={
        "my_server": {
            "type": "stdio",
            "command": "python",
            "args": ["my_mcp_server.py"],
        }
    }
)

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"Result: {args['a'] + args['b']}"}]}

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

# 创建服务器
calculator = create_sdk_mcp_server("calculator", tools=[add, multiply])

# 使用
options = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["add", "multiply"],
)

async def _handle_sdk_mcp_request(self, server_name: str, message: dict[str, Any]) -> dict[str, Any]:
    method = message.get("method")

    if method == "initialize":
        # 返回初始化结果（只声明 tools 能力）
    elif method == "tools/list":
        # 调用 server.request_handlers[ListToolsRequest]
    elif method == "tools/call":
        # 调用 server.request_handlers[CallToolRequest]
    elif method == "notifications/initialized":
        # 简单返回成功
    else:
        # 其他方法暂不支持

HookEvent = Literal[
    "PreToolUse",       # 工具调用前
    "PostToolUse",      # 工具调用后
    "UserPromptSubmit", # 用户提交 prompt 前
    "Stop",             # 停止时
    "SubagentStop",     # 子 Agent 停止时
    "PreCompact",       # 压缩上下文前
]

@dataclass
class HookMatcher:
    matcher: str | None = None    # 匹配规则，如 "Bash" 或 "Write|Edit"
    hooks: list[HookCallback] = field(default_factory=list)
    timeout: float | None = None

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

class SyncHookJSONOutput(TypedDict):
    continue_: NotRequired[bool]           # 是否继续（注意是 continue_，避免关键字）
    suppressOutput: NotRequired[bool]      # 隐藏输出
    stopReason: NotRequired[str]           # 停止原因
    decision: NotRequired[Literal["block"]]
    hookSpecificOutput: NotRequired[HookSpecificOutput]

class PreToolUseHookSpecificOutput(TypedDict):
    hookEventName: Literal["PreToolUse"]
    permissionDecision: NotRequired[Literal["allow", "deny", "ask"]]
    permissionDecisionReason: NotRequired[str]
    updatedInput: NotRequired[dict[str, Any]]  # 可以直接修改工具输入

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

class ClaudeSDKError(Exception): pass
class CLINotFoundError(ClaudeSDKError): pass   # CLI 找不到
class CLIConnectionError(ClaudeSDKError): pass # 连接失败
class CLIJSONDecodeError(ClaudeSDKError): pass # JSON 解析失败
class ProcessError(ClaudeSDKError): pass       # 进程错误
class MessageParseError(ClaudeSDKError): pass  # 消息解析失败

pip install claude-agent-sdk

cd claude-agent-sdk-python
python -m venv .venv
source .venv/bin/activate
pip install -e .

import asyncio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="用一句话解释什么是 Claude Agent SDK"):
        print(message)

asyncio.run(main())

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        system_prompt="你是一名专业的 Python 开发者",
        model="claude-3-5-sonnet-latest",
        max_turns=5,
    )

    async for message in query(
        prompt="写一个快速排序的 Python 实现",
        options=options
    ):
        print(message)

asyncio.run(main())

# 1. 系统消息：会话开始
SystemMessage(subtype="init", data={"session_id": "..."})

# 2. 用户消息
UserMessage(content="用一句话解释什么是 Claude Agent SDK")

# 3. 助手消息（可能包含多个内容块）
AssistantMessage(
    content=[
        TextBlock(text="Claude Agent SDK 是一个 Python 客户端库，..."),
    ],
    model="claude-3-5-sonnet-latest",
)

# 4. 结果消息：包含费用和统计
ResultMessage(
    subtype="result",
    duration_ms=1234,
    is_error=False,
    total_cost_usd=0.003,
    usage={...},
)