夜雨聆风Claude Agent SDK源码深度拆解
它不是一个Agent框架，而是一个CLI子进程的Python包装器。理解这一点，你就理解了整个SDK的设计哲学。
一、项目概览：一个"不走寻常路"的SDK
anthropics/claude-agent-sdk-python 在GitHub上拥有超过7600颗星，但它的架构与你想象的完全不同。大多数AI Agent SDK（如LangChain、CrewAI）是纯Python框架——它们在进程内调用API、管理状态、编排工具。而Claude Agent SDK走了一条截然不同的路：它在底层spawn一个Claude Code CLI子进程，通过stdin/stdout的NDJSON流式协议与之通信。
这意味着什么？你用Python写的每一行SDK代码，最终都是在和一个Node.js进程对话。Claude Code CLI是用TypeScript写的独立应用，拥有自己的工具系统、权限模型、会话管理和MCP集成。Python SDK只是给这个CLI套了一层Pythonic的壳。
这种设计的优劣我们稍后分析，但先看它的核心数据：
项目名：anthropics/claude-agent-sdk-python
Stars：7646
核心文件：~13个Python模块
依赖：anyio（异步IO）、mcp（MCP协议）
最低CLI版本：2.0.0
二、目录结构：精简而层次分明
SDK的目录结构遵循了清晰的分层设计：
claude_agent_sdk/
├── __init__.py # 公共API导出 + MCP server工具
├── _version.py # 版本号
├── _errors.py # 错误类型定义
├── query.py # query() 便捷函数
├── client.py # ClaudeSDKClient 类
├── types.py # 全部类型定义（2200行！）
└── _internal/
├── client.py # InternalClient 内部实现
├── query.py # Query 类（控制协议处理）
├── sessions.py # 会话列表/查询
├── session_store.py # InMemorySessionStore
├── session_resume.py # 会话恢复与物化
├── session_mutations.py # 会话重命名/标签/删除/fork
├── session_summary.py # 增量会话摘要
├── message_parser.py # CLI输出→Python类型
├── transport/
│ ├── __init__.py # Transport抽象基类
│ └── subprocess_cli.py # SubprocessCLITransport
└── ...（其他辅助模块）
最引人注目的是 types.py——2200行，是整个SDK中最大的文件。它定义了从消息类型、权限模型到MCP配置、Hook系统的所有类型。这反映了SDK的一个核心特点：它本质上是一个类型安全的协议适配层。
三、架构设计：子进程模式的完整链路
3.1 核心架构图
┌─────────────────────────────────────────────────┐
│ 你的Python应用 │
│ │
│ async for msg in query(prompt="...", options=…): │
│ print(msg) │
│ │
│ ┌─────────────┐ ┌──────────────────┐ │
│ │ query() │───>│ InternalClient │ │
│ │ (便捷入口) │ │ (编排层) │ │
│ └─────────────┘ └───────┬──────────┘ │
│ │ │
│ ┌──────────────────────────▼──────────────────┐ │
│ │ SubprocessCLITransport │ │
│ │ · 构建CLI命令行参数 │ │
│ │ · spawn claude CLI子进程 │ │
│ │ · 管理stdin/stdout流 │ │
│ └────────────┬───────────────────┬─────────────┘ │
│ │ stdin (NDJSON) │ stdout (NDJSON) │
└───────────────┼───────────────────┼─────────────────┘
│ │
┌───────────▼───────────────────▼──────────┐
│ claude CLI (Node.js进程) │
│ │
│ · Anthropic API调用 │
│ · 工具执行（Bash/Read/Edit/WebFetch…） │
│ · 权限管理 │
│ · 会话持久化（~/.claude/projects/…） │
│ · MCP服务器管理 │
│ · 沙箱执行 │
└───────────────────────────────────────────┘
3.2 为什么选择子进程模式？
这不是偶然的设计决策。Claude Code CLI本身就是一个功能完备的终端应用，拥有：
完整的工具系统：Bash、Read、Edit、Write、WebFetch等内置工具 沙箱执行环境：macOS/Linux的sandbox隔离 会话管理系统：JSONL格式的会话持久化 MCP协议支持：连接各种外部MCP服务器 权限控制：细粒度的工具权限管理 Hook系统：PreToolUse、PostToolUse等生命周期钩子
Python SDK不需要从零实现这些——它只需要把指令传给CLI，把结果解析回来。
四、query() 函数：最简洁的入口
query.py 是整个SDK最简单的文件之一，只有126行。它是面向"一次性查询"场景的便捷入口：
async def query(
*,
prompt: str | AsyncIterable[dict[str, Any]],
options: ClaudeAgentOptions | None = None,
transport: Transport | None = None,
) -> AsyncIterator[Message]:
if options is None:
options = ClaudeAgentOptions()
client = InternalClient()
async for message in client.process_query(
prompt=prompt, options=options, transport=transport
):
yield message
几个关键设计点：
async generator模式：返回 AsyncIterator[Message]，调用方用async for消费消息流prompt支持两种类型： str（单次查询）或AsyncIterable[dict]（流式输入）transport可注入：允许自定义传输层实现 options默认为空： ClaudeAgentOptions()的所有字段都有合理默认值
prompt 参数的类型设计很精妙。当你传入一个字符串时，SDK会自动构建一个user message并通过stdin发送：
# InternalClient._process_query_inner 中的处理
if isinstance(prompt, str):
user_message = {
"type": "user",
"session_id": "",
"message": {"role": "user", "content": prompt},
"parent_tool_use_id": None,
}
await chosen_transport.write(json.dumps(user_message) + "\n")
query.spawn_task(query.wait_for_result_and_end_input())
五、ClaudeSDKClient 类：有状态的双向交互
如果 query() 是"发一枪就跑"，那 ClaudeSDKClient 就是"长连接对话"。它支持：
双向通信：随时发送和接收消息 会话管理：维持跨消息的上下文 中断能力：可以中断正在进行的执行 MCP状态查询：获取MCP服务器状态 上下文使用分析：查看token消耗
class ClaudeSDKClient:
"""Client for bidirectional, interactive conversations with Claude Code."""
async def connect(
self,
prompt: str | AsyncIterable[dict[str, Any]],
options: ClaudeAgentOptions | None = None,
transport: Transport | None = None,
) -> None:
# ... 建立连接
async def receive_messages(self) -> AsyncIterator[Message]:
# ... 接收消息流
async def send_message(self, message: dict[str, Any]) -> None:
# ... 发送后续消息
async def interrupt(self) -> None:
# ... 中断当前执行
Client与query的核心区别在于会话连续性。Client通过 session_id 维持会话状态，支持在收到响应后继续发送后续消息，而query是单向的——发完prompt就等着收完所有消息。
六、类型系统：2200行的类型安全网
types.py 是SDK的心脏，定义了整个协议的类型契约。
6.1 消息类型层次
# 内容块类型
ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock | ServerToolUseBlock | ServerToolResultBlock
@dataclass
class TextBlock:
text: str
@dataclass
class ThinkingBlock:
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[str, Any]] | None = None
is_error: bool | None = None
6.2 顶层消息类型
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent | RateLimitEvent
@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
stop_reason: str | None = None
session_id: str | None = None
uuid: str | None = None
@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
6.3 ClaudeAgentOptions：SDK的完整配置中心
ClaudeAgentOptions 是一个纯数据类（不是dataclass，是类变量定义），包含了控制CLI子进程行为的所有参数。核心字段包括：
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 | SystemPromptFile | None = None
permission_mode: PermissionMode | None = None # 权限模式
max_turns: int | None = None # 最大轮次
max_budget_usd: float | None = None # 最大预算
model: str | None = None # 模型选择
cwd: str | Path | None = None # 工作目录
mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict)
can_use_tool: CanUseTool | None = None # 权限回调
hooks: dict[HookEvent, list[HookMatcher]] | None = None
agents: dict[str, AgentDefinition] | None = None # 子Agent定义
session_store: SessionStore | None = None # 外部会话存储
thinking: ThinkingConfig | None = None # 思考配置
effort: EffortLevel | None = None # 推理力度
output_format: dict[str, Any] | None = None # 结构化输出
# ... 还有约20个其他字段
这个类的设计体现了一个核心原则：所有CLI能力都通过options暴露。SubprocessCLITransport._build_command() 方法将这些选项逐一转换为CLI命令行参数。
6.4 权限模型的类型设计
权限系统是SDK中最复杂的类型设计之一：
PermissionMode = Literal["default", "acceptEdits", "plan", "bypassPermissions", "dontAsk", "auto"]
@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]
]
CanUseTool 是一个异步回调类型，接收工具名、工具参数和权限上下文，返回允许或拒绝的结果。这是SDK替代CLI交互式权限提示的核心机制。
6.5 Hook系统类型
Hook系统是SDK中最丰富的类型集合，支持10种事件：
HookEvent = (
Literal["PreToolUse"] | Literal["PostToolUse"] | Literal["PostToolUseFailure"] |
Literal["UserPromptSubmit"] | Literal["Stop"] | Literal["SubagentStop"] |
Literal["PreCompact"] | Literal["Notification"] | Literal["SubagentStart"] |
Literal["PermissionRequest"]
)
每种事件都有对应的输入类型和输出类型。例如 PreToolUseHookInput 包含工具名、工具输入和工具使用ID，而 PreToolUseHookSpecificOutput 可以返回权限决策（allow/deny/ask/defer）。
七、内部实现层：_internal/ 模块深度解析
7.1 InternalClient：编排层
_internal/client.py 是 query() 和 ClaudeSDKClient 共享的内部实现。它的核心方法 process_query() 完成以下工作：
验证选项： validate_session_store_options(options)物化会话恢复：如果指定了 resume+session_store，从外部存储加载会话到临时目录配置权限：如果设置了 can_use_tool，自动设置permission_prompt_tool_name="stdio"创建Transport：使用注入的transport或创建默认的 SubprocessCLITransport创建Query：处理控制协议 启动会话镜像：如果配置了 session_store，启动transcript mirror batcher处理输入：字符串prompt直接写入stdin，AsyncIterable在后台流式写入 产出消息：通过 parse_message()将原始JSON解析为类型化消息
async def _process_query_inner(self, prompt, options, transport, materialized):
# 验证can_use_tool的前置条件
if options.can_use_tool:
if isinstance(prompt, str):
raise ValueError("can_use_tool requires streaming mode (AsyncIterable prompt)")
if options.permission_prompt_tool_name:
raise ValueError("can_use_tool and permission_prompt_tool_name are mutually exclusive")
configured_options = replace(options, permission_prompt_tool_name="stdio")
# 创建并连接transport
chosen_transport = SubprocessCLITransport(prompt=prompt, options=configured_options)
await chosen_transport.connect()
# 创建Query处理控制协议
query = Query(
transport=chosen_transport,
is_streaming_mode=True,
can_use_tool=configured_options.can_use_tool,
hooks=self._convert_hooks_to_internal_format(configured_options.hooks),
sdk_mcp_servers=sdk_mcp_servers,
initialize_timeout=initialize_timeout,
agents=agents_dict,
exclude_dynamic_sections=exclude_dynamic_sections,
skills=configured_options.skills,
)
# 启动并初始化
await query.start()
await query.initialize()
# 处理输入
if isinstance(prompt, str):
user_message = {"type": "user", "session_id": "", "message": {"role": "user", "content": prompt}}
await chosen_transport.write(json.dumps(user_message) + "\n")
query.spawn_task(query.wait_for_result_and_end_input())
elif isinstance(prompt, AsyncIterable):
query.spawn_task(query.stream_input(prompt))
# 产出解析后的消息
async for data in query.receive_messages():
message = parse_message(data)
if message is not None:
yield message
7.2 sessions.py：会话列表与查询
sessions.py 是一个约1900行的大文件，负责扫描 ~/.claude/projects/<sanitized-cwd>/ 目录下的 .jsonl 会话文件。它的工作方式是从文件头尾读取元数据，而不是完整解析JSONL——这是一种性能优化：
LITE_READ_BUF_SIZE = 65536 # 64KB的头尾缓冲区
class _LiteSessionFile:
"""读取会话文件的头、尾、mtime和size"""
__slots__ = ("mtime", "size", "head", "tail")
def _read_session_lite(file_path: Path) -> _LiteSessionFile | None:
"""打开会话文件，stat它，读取头+尾"""
with file_path.open("rb") as f:
stat = os.fstat(f.fileno())
size = stat.st_size
mtime = int(stat.st_mtime * 1000)
head_bytes = f.read(LITE_READ_BUF_SIZE)
head = head_bytes.decode("utf-8", errors="replace")
tail_offset = max(0, size - LITE_READ_BUF_SIZE)
f.seek(tail_offset)
tail_bytes = f.read(LITE_READ_BUF_SIZE)
tail = tail_bytes.decode("utf-8", errors="replace")
return _LiteSessionFile(mtime=mtime, size=size, head=head, tail=tail)
从头尾缓冲区中提取字段使用的是正则扫描而不是JSON解析：
def _extract_json_string_field(text: str, key: str) -> str | None:
"""在不完整解析JSON的情况下提取简单字符串字段"""
patterns = [f'"{key}":"', f'"{key}": "']
for pattern in patterns:
idx = text.find(pattern)
if idx < 0:
continue
value_start = idx + len(pattern)
i = value_start
while i < len(text):
if text[i] == "\\":
i += 2
continue
if text[i] == '"':
return _unescape_json_string(text[value_start:i])
i += 1
return None
路径清理使用了与CLI一致的djb2哈希算法：
def _simple_hash(s: str) -> str:
"""32位整数哈希转base36，匹配CLI的目录命名"""
h = 0
for ch in s:
char = ord(ch)
h = (h << 5) - h + char
h = h & 0xFFFFFFFF
if h >= 0x80000000:
h -= 0x100000000
h = abs(h)
# JS toString(36)
digits = "0123456789abcdefghijklmnopqrstuvwxyz"
# ... 转base36
7.3 session_store.py：外部会话存储
InMemorySessionStore 是 SessionStore 协议的内存实现，主要用于测试。但 SessionStore 协议本身定义了SDK与外部存储的完整接口：
class SessionStore(Protocol):
async def append(self, key: SessionKey, entries: list[SessionStoreEntry]) -> None:
"""镜像一批转录条目。在本地写入成功后调用"""
...
async def load(self, key: SessionKey) -> list[SessionStoreEntry] | None:
"""加载完整会话用于恢复。在SDK父进程中调用一次"""
...
async def list_sessions(self, project_key: str) -> list[SessionStoreListEntry]:
"""列出某个project_key下的会话"""
raise NotImplementedError
async def delete(self, key: SessionKey) -> None:
"""删除会话"""
raise NotImplementedError
这个协议的设计非常务实：只有 append 和 load 是必须实现的，其他方法都是可选的。SDK在调用前会检查存储是否实现了特定方法：
def _store_implements(store: SessionStore, method: str) -> bool:
"""检查存储是否实现了特定方法（鸭子类型检测）"""
# 不使用isinstance，而是检查方法是否存在
7.4 session_resume.py：会话恢复的物化
当 resume 或 continue_conversation 与 session_store 组合使用时，SDK需要把外部存储中的会话"物化"到临时目录，因为CLI只能从本地文件恢复：
async def materialize_resume_session(options: ClaudeAgentOptions) -> MaterializedResume | None:
"""从session_store加载会话并写入临时目录"""
store = options.session_store
if store is None:
return None
if options.resume is None and not options.continue_conversation:
return None
# 解析session ID
if options.resume is not None:
if _validate_uuid(options.resume) is None:
return None
resolved = await _load_candidate(store, project_key, options.resume, timeout_s)
else:
resolved = await _resolve_continue_candidate(store, project_key, timeout_s)
# 写入临时目录
tmp_base = Path(tempfile.mkdtemp(prefix="claude-resume-"))
project_dir = tmp_base / "projects" / project_key
project_dir.mkdir(parents=True, exist_ok=True)
_write_jsonl(project_dir / f"{session_id}.jsonl", entries)
# 复制认证文件
_copy_auth_files(tmp_base, options.env)
return MaterializedResume(
config_dir=tmp_base,
resume_session_id=session_id,
cleanup=cleanup, # 清理临时目录的协程
)
这里有一个关键的安全细节：refreshToken被删除。恢复后的子进程运行在重定向的 CLAUDE_CONFIG_DIR 下，如果它刷新了token，新的token会写到父进程永远不会读取的位置，导致父进程的存储凭证被撤销。所以SDK在复制凭证时删除了 refreshToken。
7.5 session_mutations.py：会话的CRUD操作
这个模块提供了会话的增删改查操作，包括：
重命名：通过追加 custom-titleJSONL条目实现标签：通过追加 tagJSONL条目实现删除：删除 .jsonl文件和子agent目录Fork：复制会话并重新映射所有UUID
Fork操作是最复杂的——它需要重新映射每条消息的UUID和parentUuid链：
def _build_fork_lines(transcript, content_replacements, session_id, up_to_message_id, title, derive_title):
# 过滤掉sidechain
transcript = [e for e in transcript if not e.get("isSidechain")]
# 生成UUID映射
uuid_mapping = {}
for entry in transcript:
uuid_mapping[entry["uuid"]] = str(uuid.uuid4())
forked_session_id = str(uuid.uuid4())
for original in writable:
new_uuid = uuid_mapping[original["uuid"]]
# 解析parentUuid，跳过progress祖先
new_parent_uuid = None
parent_id = original.get("parentUuid")
while parent_id:
parent = by_uuid.get(parent_id)
if parent.get("type") != "progress":
new_parent_uuid = uuid_mapping.get(parent_id)
break
parent_id = parent.get("parentUuid")
forked = {
**original,
"uuid": new_uuid,
"parentUuid": new_parent_uuid,
"sessionId": forked_session_id,
"forkedFrom": {"sessionId": session_id, "messageUuid": original["uuid"]},
}
7.6 session_summary.py：增量摘要推导
fold_session_summary 是一个纯函数式的设计——它让存储适配器在 append() 内部增量维护每个会话的摘要，而不需要重新读取整个转录：
def fold_session_summary(prev, key, entries):
"""将一批追加的条目折叠到key的运行摘要中"""
if prev is not None:
summary = {"session_id": prev["session_id"], "mtime": prev["mtime"], "data": dict(prev["data"])}
else:
summary = {"session_id": key["session_id"], "mtime": 0, "data": {}}
data = summary["data"]
for raw in entries:
entry = cast("dict[str, Any]", raw)
ms = _iso_to_epoch_ms(entry.get("timestamp"))
# 设置一次性字段
if "is_sidechain" not in data:
data["is_sidechain"] = entry.get("isSidechain") is True
if "created_at" not in data and ms is not None:
data["created_at"] = ms
if "cwd" not in data:
cwd = entry.get("cwd")
if isinstance(cwd, str) and cwd:
data["cwd"] = cwd
_fold_first_prompt(data, entry)
# last-wins字段
for src, dst in _LAST_WINS_FIELDS.items():
val = entry.get(src)
if isinstance(val, str):
data[dst] = val
return summary
这个设计的精妙之处在于：所有派生状态都是append-incremental的——要么set-once（如created_at），要么last-wins（如customTitle）。存储适配器永远不需要重新读取已追加的条目。
7.7 message_parser.py：CLI输出到Python类型的映射
parse_message 函数是协议解析的核心——它将CLI输出的原始JSON字典转换为类型化的Python对象：
def parse_message(data: dict[str, Any]) -> Message | None:
message_type = data.get("type")
match message_type:
case "user":
# 解析用户消息，支持string和list两种content格式
content_blocks = []
for block in data["message"]["content"]:
match block["type"]:
case "text":
content_blocks.append(TextBlock(text=block["text"]))
case "tool_use":
content_blocks.append(ToolUseBlock(id=block["id"], name=block["name"], input=block["input"]))
case "tool_result":
content_blocks.append(ToolResultBlock(tool_use_id=block["tool_use_id"], content=block.get("content")))
return UserMessage(content=content_blocks, uuid=data.get("uuid"))
case "assistant":
# 解析助手消息，包含thinking/tool_use/server_tool_use等
for block in raw_content:
match block["type"]:
case "text": ...
case "thinking": ...
case "tool_use": ...
case "tool_result": ...
case "server_tool_use": ...
case "advisor_tool_result": ...
return AssistantMessage(content=content_blocks, model=data["message"]["model"])
case "result":
return ResultMessage(
subtype=data["subtype"],
duration_ms=data["duration_ms"],
is_error=data["is_error"],
num_turns=data["num_turns"],
session_id=data["session_id"],
total_cost_usd=data.get("total_cost_usd"),
...
)
case _:
# 前向兼容：跳过未知消息类型
logger.debug("Skipping unknown message type: %s", message_type)
return None
注意最后的 case _: 分支——这是一种前向兼容设计。新版本CLI可能发送旧版SDK不认识的消息类型，直接跳过而不是报错。
八、传输层：SubprocessCLITransport
subprocess_cli.py 是SDK中技术含量最高的文件（876行），负责CLI子进程的完整生命周期管理。
8.1 CLI发现机制
def _find_cli(self) -> str:
# 1. 优先查找打包的CLI
bundled_cli = self._find_bundled_cli()
if bundled_cli:
return bundled_cli
# 2. 系统PATH搜索
if cli := shutil.which("claude"):
return cli
# 3. 预定义路径搜索
locations = [
Path.home() / ".npm-global/bin/claude",
Path("/usr/local/bin/claude"),
Path.home() / ".local/bin/claude",
Path.home() / "node_modules/.bin/claude",
Path.home() / ".yarn/bin/claude",
Path.home() / ".claude/local/claude",
]
for path in locations:
if path.exists() and path.is_file():
return str(path)
raise CLINotFoundError("Claude Code not found...")
8.2 命令行构建
_build_command() 方法将 ClaudeAgentOptions 的每个字段映射为CLI参数：
def _build_command(self) -> list[str]:
cmd = [self._cli_path, "--output-format", "stream-json", "--verbose"]
# 系统提示
if self._options.system_prompt is None:
cmd.extend(["--system-prompt", ""])
elif isinstance(self._options.system_prompt, str):
cmd.extend(["--system-prompt", self._options.system_prompt])
# 工具配置
if self._options.allowed_tools:
cmd.extend(["--allowedTools", ",".join(effective_allowed_tools)])
# 权限模式
if self._options.permission_mode:
cmd.extend(["--permission-mode", self._options.permission_mode])
# 会话恢复（使用=形式防止值被解析为flag）
if self._options.resume:
cmd.append(f"--resume={self._options.resume}")
# MCP服务器配置
if self._options.mcp_servers:
cmd.extend(["--mcp-config", json.dumps({"mcpServers": servers_for_cli})])
# 思考配置
if self._options.thinking is not None:
t = self._options.thinking
if t["type"] == "adaptive":
cmd.extend(["--thinking", "adaptive"])
elif t["type"] == "enabled":
cmd.extend(["--max-thinking-tokens", str(t["budget_tokens"])])
# 始终使用stream-json输入格式
cmd.extend(["--input-format", "stream-json"])
return cmd
8.3 子进程生命周期
connect() 方法负责启动子进程，核心流程：
async def connect(self) -> None:
# 查找CLI
self._cli_path = await anyio.to_thread.run_sync(self._find_cli)
# 版本检查
if not os.environ.get("CLAUDE_AGENT_SDK_SKIP_VERSION_CHECK"):
await self._check_claude_version()
cmd = self._build_command()
# 构建环境变量
process_env = {
**{k: v for k, v in os.environ.items() if k != "CLAUDECODE"},
"CLAUDE_CODE_ENTRYPOINT": "sdk-py",
"CLAUDE_AGENT_SDK_VERSION": __version__,
**self._options.env,
}
# 注入OTEL trace context
try:
from opentelemetry import propagate
carrier = {}
propagate.inject(carrier)
if "traceparent" in carrier:
for k, v in carrier.items():
process_env[k.upper()] = v
except Exception:
pass
# 启动子进程
self._process = await anyio.open_process(
cmd, stdin=PIPE, stdout=PIPE, stderr=stderr_dest,
cwd=self._cwd, env=process_env, user=self._options.user,
)
_ACTIVE_CHILDREN.add(self._process)
注意环境变量处理中的几个细节：
过滤CLAUDECODE：防止SDK生成的子进程认为自己在Claude Code父进程中运行 设置ENTRYPOINT：标识来源为Python SDK OTEL传播：注入分布式追踪上下文
8.4 NDJSON帧重组
CLI的stdout输出是NDJSON格式（每行一个JSON对象），但 anyio 的 TextReceiveStream 返回的是chunk而不是行。_LineFramer 负责重组：
class _LineFramer:
"""从返回任意chunk的流中重组完整行"""
def __init__(self) -> None:
self._pending: list[str] = []
self.pending_len = 0
def push(self, chunk: str) -> list[str]:
self._pending.append(chunk)
self.pending_len += len(chunk)
if "\n" not in chunk:
return []
# 拼接所有pending chunk并按\n分割
*lines, tail = "".join(self._pending).split("\n")
self._pending, self.pending_len = [tail], len(tail)
return lines
8.5 优雅关闭与信号处理
close() 方法实现了三级进程终止策略：
async def close(self) -> None:
with anyio.CancelScope(shield=True): # 屏蔽取消，确保清理完成
# 1. 关闭stdin（发送EOF）
if self._stdin_stream:
await self._stdin_stream.aclose()
# 2. 等待优雅退出（5秒）
try:
with anyio.fail_after(5):
await self._process.wait()
except TimeoutError:
# 3. SIGTERM
self._process.terminate()
try:
with anyio.fail_after(5):
await self._process.wait()
except TimeoutError:
# 4. SIGKILL
self._process.kill()
SDK还注册了 atexit 处理器，确保父进程退出时清理所有活跃的子进程：
_ACTIVE_CHILDREN: set[Process] = set()
def _kill_active_children() -> None:
for p in list(_ACTIVE_CHILDREN):
with suppress(Exception):
p.send_signal(signal.SIGTERM)
_ACTIVE_CHILDREN.clear()
atexit.register(_kill_active_children)
九、权限模型：多层防护
SDK的权限模型有三个层次：
9.1 工具白名单/黑名单
allowed_tools: list[str] = ["Read", "Bash(ls:*)", "Edit(src/**/*.py)"]
disallowed_tools: list[str] = ["Write", "Bash(rm:*)"]
allowed_tools 支持模式匹配：
"Read"— 允许整个Read工具"Bash(ls:*)"— 只允许以ls开头的Bash命令"Edit(src/**/*.py)"— 只允许编辑匹配glob的文件
9.2 权限模式
PermissionMode = Literal[
"default", # 标准模式，危险操作需要确认
"acceptEdits", # 自动接受文件编辑
"plan", # 只规划不执行
"bypassPermissions", # 绕过所有权限检查（慎用！）
"dontAsk", # 不询问，未预批准的直接拒绝
"auto", # 模型分类器自动审批
]
9.3 can_use_tool回调
最灵活的权限控制方式——完全自定义的异步回调：
async def my_permission_handler(
tool_name: str,
tool_input: dict[str, Any],
context: ToolPermissionContext
) -> PermissionResult:
if tool_name == "Bash" and "rm" in tool_input.get("command", ""):
return PermissionResultDeny(message="rm命令被禁止")
return PermissionResultAllow()
回调通过控制协议与CLI通信。当CLI需要权限确认时，它通过stdin发送一个 permission_request 控制请求，SDK调用回调，然后将结果通过 permission_response 发回：
# Query._handle_permission_request 中的处理
async def _handle_permission_request(self, request):
tool_name = request.get("tool_name")
tool_input = request.get("tool_input", {})
context = ToolPermissionContext(...)
result = await self.can_use_tool(tool_name, tool_input, context)
# 发回CLI
response = {
"type": "control_response",
"request_id": request_id,
"response": {
"behavior": result.behavior,
"message": getattr(result, "message", ""),
}
}
十、MCP集成：外部工具的桥梁
MCP（Model Context Protocol）是Anthropic推出的工具协议标准。SDK支持三种MCP服务器类型：
McpServerConfig = McpStdioServerConfig | McpSSEServerConfig | McpHttpServerConfig | McpSdkServerConfig
10.1 外部MCP服务器
标准的stdio/SSE/HTTP服务器配置直接传递给CLI：
mcp_servers = {
"my-server": {"type": "stdio", "command": "my-mcp-server", "args": ["--port", "3000"]},
"remote-server": {"type": "http", "url": "https://mcp.example.com"},
}
10.2 SDK内置MCP服务器
SDK还支持进程内MCP服务器——工具函数直接在Python进程中运行，无需IPC开销：
@tool("greet", "Greet a user", {"name": str})
async def greet(args):
return {"content": [{"type": "text", "text": f"Hello, {args['name']}!"}]}
server = create_sdk_mcp_server("my-tools", tools=[greet])
options = ClaudeAgentOptions(mcp_servers={"calc": server}, allowed_tools=["greet"])
create_sdk_mcp_server 在内部创建了一个MCP Server实例，注册了 list_tools 和 call_tool 处理器。当CLI请求工具列表或调用工具时，请求通过控制协议转发到Python端：
# Query._handle_sdk_mcp_request 中的处理
async def _handle_sdk_mcp_request(self, server_name, message):
server = self.sdk_mcp_servers[server_name]
method = message.get("method")
if method == "tools/list":
handler = server.request_handlers.get(ListToolsRequest)
result = await handler(ListToolsRequest(method=method))
# 转换为JSONRPC响应
return {"jsonrpc": "2.0", "id": message.get("id"), "result": {"tools": tools_data}}
elif method == "tools/call":
handler = server.request_handlers.get(CallToolRequest)
result = await handler(call_request)
return {"jsonrpc": "2.0", "id": message.get("id"), "result": response_data}
十一、与框架型SDK的对比
Claude Agent SDK的"CLI包装器"架构与LangChain、CrewAI等"框架型"SDK有本质区别：
| 维度 | Claude Agent SDK | 框架型SDK（LangChain等） |
|---|---|---|
| 运行模型 | spawn子进程 | 进程内直接调用API |
| 工具执行 | CLI在沙箱中执行 | 框架在宿主进程中执行 |
| 状态管理 | CLI的JSONL文件 | 框架的内存/数据库 |
| 权限系统 | CLI原生多层权限 | 框架自行实现 |
| 沙箱隔离 | 操作系统级沙箱 | 通常无沙箱 |
| MCP支持 | CLI原生支持 | 需要自行集成 |
| 会话持久化 | CLI的JSONL + SessionStore协议 | 框架自定义 |
| 更新频率 | 跟随CLI发布 | 独立版本管理 |
| 调试难度 | 需要同时理解Python和CLI | 纯Python栈 |
| 性能开销 | 子进程spawn + IPC | 无额外开销 |
优势：
天然沙箱安全——工具在CLI的沙箱中执行，不直接暴露给宿主进程 功能跟随CLI——CLI更新的所有工具、权限、MCP功能自动可用 权限系统完整——复用CLI的多层权限模型
劣势：
spawn开销——每次查询都需要启动Node.js子进程 调试困难——问题可能出在Python层或CLI层 依赖CLI安装——必须先安装 @anthropic-ai/claude-codenpm包环境耦合——CLI的行为受其配置文件、环境变量影响
十二、总结：一个务实的工程决策
Claude Agent SDK的架构选择反映了一个务实的工程决策：不重新发明轮子。Claude Code CLI已经是一个功能完备的AI编程助手，拥有复杂的工具系统、权限模型和沙箱环境。Python SDK的价值不在于重新实现这些能力，而在于提供一个类型安全的、Pythonic的接口来调用它们。
这种架构最适合以下场景：
CI/CD集成：在自动化流水线中调用Claude Code 后端服务：在Python后端中嵌入Claude Code能力 脚本自动化：批量处理任务 已有CLI用户：从CLI迁移到编程接口
对于需要深度定制Agent行为、自定义工具链、或追求最低延迟的场景，框架型SDK可能更合适。但如果你想要Claude Code的全部能力通过Python接口访问——包括沙箱、MCP、权限、Hook——Claude Agent SDK是唯一的选择。
理解了"CLI包装器"这个核心定位，SDK中所有的设计决策——从2200行的类型定义到子进程的NDJSON协议，从SessionStore的物化恢复到MCP的控制协议转发——都变得合理而自洽。这不是一个偷懒的设计，而是一个在给定约束下的最优解。
