Claude Code SDK #46：ClaudeSDKClient 全解：query、receive_response、interrupt，把 Agent 变成可控多轮会话

先看一个反直觉点

query() 和 ClaudeSDKClient 不是同一个抽象

一个客户端，三轮对话

import asyncio

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ClaudeSDKClient,
    ResultMessage,
    TextBlock,
)

async def print_response(message) -> None:
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                print(f"Claude: {block.text}")

elif isinstance(message, ResultMessage):
        if message.subtype == "success":
            print(f"[done] {message.result}")
        else:
            print(f"[stopped] {message.subtype}")

async def main() -> None:
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep"],
        max_turns=8,
    )

async with ClaudeSDKClient(options=options) as client:
        await client.query("分析 auth 模块的登录流程")
        async for message in client.receive_response():
            await print_response(message)

await client.query("刚才提到的 token 在哪些文件里被创建？")
        async for message in client.receive_response():
            await print_response(message)

await client.query("只读审查完成后，列出三个最值得先修的风险")
        async for message in client.receive_response():
            await print_response(message)

asyncio.run(main())

Agent loop 仍然在客户端里面运行

中断：先停任务，再清空消息边界

1. 发起长任务。
2. 调用 interrupt()。
3. 继续消费 receive_response()，直到拿到被中断任务的终态。
4. 再发起新 query。

import asyncio

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, ResultMessage

async def interruptible_task() -> None:
    options = ClaudeAgentOptions(
        allowed_tools=["Bash"],
        permission_mode="acceptEdits",
    )

async with ClaudeSDKClient(options=options) as client:
        await client.query("慢慢执行一个长任务，逐步检查当前目录中的文件")
        await asyncio.sleep(2)
        await client.interrupt()

interrupted_result = None
        async for message in client.receive_response():
            if isinstance(message, ResultMessage):
                interrupted_result = message

print(f"上一任务结束状态：{interrupted_result.subtype}")

await client.query("停止刚才的计划，只输出当前已经确认的文件列表")
        async for message in client.receive_response():
            if isinstance(message, ResultMessage):
                print(f"新任务结束状态：{message.subtype}")

asyncio.run(interruptible_task())

不要用 break 结束消息流

result = None

async for message in client.receive_response():
    if isinstance(message, ResultMessage):
        result = message

# 迭代器自然结束后，再读取 result
if result is not None and result.subtype == "success":
    print(result.result)

会话内还能动态改什么

• set_permission_mode(mode)：改变当前会话的权限模式。
• set_model(model)：切换当前会话使用的模型，传入 None 可恢复默认模型。
• get_mcp_status()：读取 MCP 服务状态。
• reconnect_mcp_server(name)：重新连接失效的 MCP 服务。
• toggle_mcp_server(name, enabled)：在会话中启用或停用某个 MCP 服务。
• rewind_files(user_message_id)：在启用 enable_file_checkpointing=True 后，把文件恢复到指定用户消息对应的状态。

长会话要管住上下文

最后的选型规则

• 一个 prompt 对应一个结果，选 query()。
• 下一条输入依赖上一轮上下文，选 ClaudeSDKClient。
• 用户可以点击停止、改权限、切换模型，选 ClaudeSDKClient。
• 进程会重启，先保存 session_id，再设计 resume。
• 要保留原会话并尝试另一种方案，使用 fork，不要把文件复制一份就当成会话分支。