跳到主要内容

数据截至 (上游 commit a4eaba4a56f9)

两个入口 — query() 与 ClaudeSDKClient

本章讲:SDK 给你两个门，怎么选、各自的消息循环长什么样、它们内部其实共用同一套机制。

1. 先看结论:该用哪个​

维度query()ClaudeSDKClient
交互方向单向:一次性发完、收完双向:随时发、随时收
状态无状态,每次独立有状态,维持一条连接
能否打断否能（interrupt()）
能否中途换模型/权限否能（set_model / set_permission_mode）
典型场景脚本、批处理、CI、一问一答聊天 UI、REPL、需要看响应再决定下一步
代码形态async for m in query(...)async with ClaudeSDKClient() as c:

这张对照表直接抄自两者的 docstring（query.py:24 与 client.py:34 都专门写了"何时用我 / 何时用另一个"）。

一句话选择: 输入一次给全、不用中途插话 → query();需要来回、需要打断、需要根据回答再发下一句 → ClaudeSDKClient。

2. query():一发一收的迭代器​

2.1 它要解决的小问题​

最常见的需求就是"问一句、把回答流式读出来"。query() 把连接、初始化、收尾全藏起来,只暴露一个异步迭代器。

2.2 原理演示​

# 示意,非源码：query() 的使用骨架
from claude_agent_sdk import query, ClaudeAgentOptions

options = ClaudeAgentOptions(system_prompt="You are terse", max_turns=1)
async for message in query(prompt="Tell me a joke", options=options):
    print(message)   # 逐条产出：可能是 AssistantMessage、ResultMessage 等

2.3 真实实现​

query() 本身极薄——它只是把活儿转给 InternalClient.process_query(query.py:118-126）：

# 真实源码 query.py:118
if options is None:
    options = ClaudeAgentOptions()
client = InternalClient()
async for message in client.process_query(prompt=prompt, options=options, transport=transport):
    yield message

关键点:prompt 可以是 str(一次性),也可以是 AsyncIterable[dict](流式多条,但仍是单向——全发完才收）。prompt 的字典结构在 docstring 里写死了:{"type":"user","message":{...}}(query.py:47）。

2.4 一个易错点:query() 也"永远 streaming"​

很多人以为 str prompt 走的是"非流式"路径。实则不然:InternalClient 里创建 Query 时写死 is_streaming_mode=True(_internal/client.py:137,注释明说"Always streaming internally, matching TypeScript SDK"）。差别只在于:字符串 prompt 会在 initialize 之后作为一条 user 消息写进 stdin,然后 spawn_task(query.wait_for_result_and_end_input()) 等第一个 result 到了再关 stdin(_internal/client.py:181-186）。

3. ClaudeSDKClient:有状态的连接​

3.1 它要解决的小问题​

聊天类应用需要:连上一次、来回多轮、根据 Claude 的回答再决定发什么、必要时打断。query() 的"发完才收"满足不了,于是有了 ClaudeSDKClient。

3.2 典型用法（真实来自 examples/streaming_mode.py）​

# 示意,非源码：交互循环的骨架
async with ClaudeSDKClient(options=options) as client:
    await client.query("What's the capital of France?")
    async for msg in client.receive_response():   # 收到 ResultMessage 自动停
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print("Claude:", block.text)
    # 还能继续 await client.query("下一句") ...

3.3 两个收消息的方法,别搞混​

方法行为出处
receive_messages()无限产出所有消息,不会自己停client.py:271
receive_response()产出到并包含一条 ResultMessage 后停止client.py:567

receive_response() 就是在 receive_messages() 外面套了一层"见到 result 就 return"（client.py:532）。单轮问答用它最省心。单轮问答用它最省心。

3.4 连接生命周期​

async with 进入时调 connect(),退出时总是调 disconnect()(client.py:589-598）。connect() 里做了一串关键动作:

connect()
  ├─ validate_session_store_options()        校验参数,早失败
  ├─ materialize_resume_session()            需要时把 resume 会话落地到临时目录
  └─ _connect_inner()
       ├─ 若设了 can_use_tool → 强制 streaming，permission_prompt_tool_name="stdio"
       ├─ 建 SubprocessCLITransport 并 connect()
       ├─ 建 Query(is_streaming_mode=True, 传入钩子/agents/skills)
       ├─ query.start()  → 启后台读循环
       ├─ query.initialize()  → 发 initialize 控制请求
       └─ 把初始 prompt 写进 stdin（或后台 stream_input）

依据:client.py:99(connect)、client.py:147(_connect_inner)、client.py:160-176（can_use_tool 校验与 stdio 设置）。

3.5 交互能力一览​

ClaudeSDKClient 比 query() 多出的方法,底层全是"发一条控制请求":

方法作用出处
interrupt()打断当前生成client.py:313
set_permission_mode(mode)中途改权限模式client.py:319
set_model(model)中途换模型client.py:346
rewind_files(id)回滚文件到某条 user 消息时的状态client.py:370
get_mcp_status()查 MCP 服务器连接状态client.py:473
get_context_usage()查上下文窗口占用明细（同 /context）client.py:506
reconnect_mcp_server / toggle_mcp_server重连/开关某个 MCP 服务器client.py:402 / :424
stop_task(task_id)停一个后台任务client.py:450

这些方法有个共同前提:只在 streaming 模式可用,且必须先 connect()——没连就抛 CLIConnectionError（每个方法开头都有 if not self._query: raise ...）。

4. 一个跨两者的注意点:async 上下文亲和性​

ClaudeSDKClient 的 docstring 警告(client.py:58-64):从 connect() 到 disconnect(),它内部维持一个持久的 anyio 任务组读消息,不能跨不同的 async 运行时上下文（比如不同 trio nursery 或 asyncio task group）用同一个实例。所有操作要在"连接它的那个 async 上下文"里完成。这是已知局限,作者也在注释里写了"理想情况不该存在"。

5. 代码地图​

主题文件路径符号名
一次性入口src/claude_agent_sdk/query.pyquery
query 编排src/claude_agent_sdk/_internal/client.pyInternalClient.process_query _process_query_inner
交互客户端src/claude_agent_sdk/client.pyClaudeSDKClient connect _connect_inner
收消息src/claude_agent_sdk/client.pyreceive_messages receive_response
交互控制src/claude_agent_sdk/client.pyinterrupt set_model set_permission_mode