Claude Agent SDK for Python — 架构与原理

30 秒导读: 这是一个 Python 库，让你几行代码就能把 Claude 当成一个能读写文件、跑命令、调工具的"agent"来用。关键点：它自己不跑模型、不实现工具，而是把一个捆绑好的 claude 命令行程序拉起成子进程，两边用 JSON 文本行对讲；权限判断、钩子、你自定义的工具，都通过一条"控制协议"回调进你的 Python 代码。

1. 这是什么（零基础也能懂）

一句话定义: claude-agent-sdk 是 Claude Code 的 Python 封装——你写 async for message in query(...)，它替你启动 Claude Code CLI、喂 prompt、把 CLI 吐出的 JSON 流解析成带类型的 Python 对象。

解决什么问题 / 给谁用。 假设你想在自己的脚本或后端服务里，让 Claude 自动改一个项目的代码、跑测试、查资料——你不想自己去拼 API 请求、管工具循环、处理权限弹窗。这个 SDK 把这些都包好了。典型用户：

写自动化脚本 / CI 流水线的工程师（"帮我把这个 bug 修了并跑通测试"）。
做聊天式、交互式 agent 应用的开发者（需要多轮、能打断、能中途换模型）。
想给 Claude 加自定义工具的人（用 Python 函数当工具，模型能直接调）。

它能做什么（功能）:

一发一收地问（query()）或多轮交互地聊（ClaudeSDKClient）。
把 Python async 函数注册成模型可调用的工具（进程内 MCP，无需另起进程）。
用回调决定"这个工具调用准不准"（can_use_tool 权限钩子）。
在工具执行前后插入钩子（hooks，如 PreToolUse 拦截危险命令）。
列举 / 改名 / 打标签 / 删除 / 分叉历史会话；把会话镜像到外部存储（S3、Redis、Postgres）。
中途 interrupt() 打断、set_model() 换模型、set_permission_mode() 改权限模式。

用起来什么样。 最小示例（真实来自 README 与 examples/quick_start.py）：

import anyio
from claude_agent_sdk import query, AssistantMessage, TextBlock

async def main():
    # query() 返回一个异步迭代器，逐条产出对话消息
    async for message in query(prompt="What is 2 + 2?"):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)

一句话直觉/类比。 把这个 SDK 想成"遥控器 + 翻译官":真正干活的电视机是那个 claude 命令行程序（它连 API、跑工具）。SDK 负责按开机键（拉起子进程）、把你的话翻成它懂的 JSON（写 stdin）、再把它满屏的 JSON 流翻回 Python 对象（读 stdout）。它自己不放节目。

本节到此不碰代码细节。记住一件事：SDK 是 CLI 的驱动器，不是 agent 本体。

2. 顶层全景（它大概怎么转）

2.1 一张图看清分层

怎么读这张图:从上往下是"你的代码 → SDK → 子进程 → 云";中间那条虚线箭头是反向回调——CLI 想问"这工具能用吗"时，控制协议把问题送回你的 Python 进程。

  你的 Python 代码
     │  query(prompt=...)  或  ClaudeSDKClient(options)
     ▼
  ┌─────────────────────────────────────────────┐
  │ InternalClient / ClaudeSDKClient             │  组装 options、校验参数
  └─────────────────────────────────────────────┘
     │
     ▼
  ┌─────────────────────────────────────────────┐         回调你的进程
  │ Query（控制协议层）                          │◀ ─ ─ ─ ─ ─ ─ ─ ─ ─┐
  │  路由 control_request / control_response     │   权限 can_use_tool │
  │  跑钩子、跑进程内 MCP 工具、发 initialize     │   钩子 hooks         │
  └─────────────────────────────────────────────┘   工具 @tool         │
     │  写 JSON 行到 stdin / 读 JSON 行从 stdout                        │
     ▼                                                                 │
  ┌─────────────────────────────────────────────┐                     │
  │ SubprocessCLITransport                       │                     │
  │  找 claude 二进制、拼命令行、开子进程         │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─┘
  │  管 stdin/stdout/stderr、缓冲、清理           │
  └─────────────────────────────────────────────┘
     │  stdin: stream-json    stdout: stream-json
     ▼
  ┌─────────────────────────────────────────────┐
  │ claude 二进制（Claude Code CLI，随包捆绑）     │
  │  真正的 agent 循环 + 真实工具执行             │
  └─────────────────────────────────────────────┘
     │
     ▼
  Anthropic API  +  Bash/Read/Edit/… 真实工具

2.2 部件一句话职责

部件	干什么	在哪个文件（符号）
`query()`	一次性提问入口，返回消息异步迭代器	`src/claude_agent_sdk/query.py:11` `query`
`ClaudeSDKClient`	多轮交互入口，能打断/换模型/发后续	`src/claude_agent_sdk/client.py:26` `ClaudeSDKClient`
`InternalClient`	`query()` 背后的编排：建传输、建 Query、跑循环	`src/claude_agent_sdk/_internal/client.py:28` `InternalClient`
`Query`	控制协议大脑：路由控制消息、跑回调、发 initialize	`src/claude_agent_sdk/_internal/query.py:61` `Query`
`Transport`（抽象）	低层 I/O 接口，可自定义（如远程连接）	`src/claude_agent_sdk/_internal/transport/__init__.py:8` `Transport`
`SubprocessCLITransport`	默认传输：拉起 `claude` 子进程、读写管道	`src/claude_agent_sdk/_internal/transport/subprocess_cli.py:50`
`parse_message`	把 CLI 的原始 JSON 变成带类型的消息对象	`src/claude_agent_sdk/_internal/message_parser.py:35`
`tool` / `create_sdk_mcp_server`	把 Python 函数做成进程内工具	`src/claude_agent_sdk/__init__.py:169` / `:310`
`ClaudeAgentOptions`	所有配置的大 dataclass	`src/claude_agent_sdk/types.py:1634`
`SessionStore`（协议）	外部存储镜像会话的适配器接口	`src/claude_agent_sdk/types.py:1425`

2.3 主线走一遍（高层，不进代码）

一次 query(prompt="...") 从头到尾:

InternalClient.process_query 先校验 session_store 相关参数，需要时把要 resume 的会话从外部存储"落地"到临时目录（_internal/client.py:52）。
创建 SubprocessCLITransport,它把 options 翻译成一长串 claude --output-format stream-json --verbose ... 命令行，开子进程。
创建 Query（永远用 streaming 模式），启动后台读循环，发一条 initialize 控制请求（把钩子、agents、skills 配置递过去）。
把你的 prompt 作为一条 user JSON 消息写进子进程 stdin。
子进程开始干活，往 stdout 吐 assistant / user / system / result 等 JSON 行；读循环解析后逐条 yield 给你。
期间若 CLI 需要"这工具能用吗"或"跑一下这个进程内工具",它发 control_request 回来，Query 调你的回调，把结果 control_response 写回去。
收到 result 消息表示这一轮结束;query() 会话随之收尾、终止子进程、清临时目录。

3. 该读哪一章（阅读地图）

建议顺序:先 01 建立"两个入口"的心智模型,再 02 吃透核心的控制协议,后面三章按需读。

你想搞懂…	去这章
`query()` 和 `ClaudeSDKClient` 到底差在哪、消息循环怎么写	`01-two-entrypoints.md`
SDK 和 CLI 之间那条协议怎么工作、子进程怎么管（最核心）	`02-transport-control-protocol.md`
怎么给模型加 Python 工具、怎么管权限和钩子	`03-tools-permissions-hooks.md`
resume 怎么落地、怎么把会话存到 S3/Redis/Postgres	`04-sessions-and-store.md`
消息解析、跨 asyncio/trio 的任务管理、它会在哪崩	`05-craft-and-boundaries.md`

4. 全局代码地图（导航索引）

用符号名 grep 定位比行号更抗漂移。

主题	文件路径	符号名
一次性入口	`src/claude_agent_sdk/query.py`	`query`
交互式客户端	`src/claude_agent_sdk/client.py`	`ClaudeSDKClient` `connect` `receive_response`
query 编排	`src/claude_agent_sdk/_internal/client.py`	`InternalClient.process_query` `_process_query_inner`
控制协议大脑	`src/claude_agent_sdk/_internal/query.py`	`Query` `_read_messages` `_handle_control_request` `_send_control_request`
子进程传输	`src/claude_agent_sdk/_internal/transport/subprocess_cli.py`	`SubprocessCLITransport` `_build_command` `_read_messages_impl`
传输抽象	`src/claude_agent_sdk/_internal/transport/__init__.py`	`Transport`
消息解析	`src/claude_agent_sdk/_internal/message_parser.py`	`parse_message`
进程内工具	`src/claude_agent_sdk/__init__.py`	`tool` `create_sdk_mcp_server` `_python_type_to_json_schema`
配置	`src/claude_agent_sdk/types.py`	`ClaudeAgentOptions`
权限/钩子类型	`src/claude_agent_sdk/types.py`	`CanUseTool` `PermissionResultAllow` `HookMatcher` `PermissionUpdate`
会话存储协议	`src/claude_agent_sdk/types.py`	`SessionStore` `SessionKey`
resume 落地	`src/claude_agent_sdk/_internal/session_resume.py`	`materialize_resume_session` `apply_materialized_options`
镜像批处理	`src/claude_agent_sdk/_internal/transcript_mirror_batcher.py`	`TranscriptMirrorBatcher`
会话增改	`src/claude_agent_sdk/_internal/session_mutations.py`	`fork_session` `rename_session`
跨后端任务	`src/claude_agent_sdk/_internal/_task_compat.py`	`spawn_detached` `TaskHandle`
错误类型	`src/claude_agent_sdk/_errors.py`	`ClaudeSDKError` `CLINotFoundError` `ProcessError`

1. 这是什么（零基础也能懂）​

2. 顶层全景（它大概怎么转）​

2.1 一张图看清分层​

2.2 部件一句话职责​

2.3 主线走一遍（高层，不进代码）​

3. 该读哪一章（阅读地图）​

4. 全局代码地图（导航索引）​