第 7 章：权限控制与安全

Claude 很强大——它能读文件、写代码、执行命令。但"能力越大，责任越大"。 这一章教你怎么给 Claude 画好边界，让它在你允许的范围内干活。

前提条件

1. 已完成前 6 章的学习
2. 了解 query() 和 ClaudeAgentOptions 的基本用法

1. 为什么需要权限控制？

想象你招了一个超级能干的实习生。他会写代码、能跑脚本、还能操作数据库。 但你肯定不会第一天就让他随便动生产环境的代码吧？

Claude 也是一样。它有能力做很多事，但不代表你应该让它什么都做。

几个真实的场景：

• 误删文件：Claude 在整理项目时，把它认为"没用"的配置文件删了
• 跑飞了：Claude 不断调用工具，跑了 50 轮还没停，API 费用蹭蹭涨
• 执行危险命令：Claude 觉得 rm -rf 能"清理"项目目录
• 访问敏感数据：Claude 读到了 .env 里的数据库密码

权限控制就是你的"安全网"。它不会限制 Claude 的能力，只是让你决定—— 哪些事 Claude 可以自己做，哪些事要先问你。

2. 四种权限模式

SDK 提供了四种内置的权限模式，从最严格到最宽松：

from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    permission_mode="default",  # 四选一
)

模式对比

default — 每次都问

这是默认模式。Claude 每次想使用工具（读文件、写文件、跑命令），都需要你确认。

就像实习生每做一步都来问你："这个文件我可以改吗？" "这条命令我可以执行吗？"

安全，但很烦。适合你第一次跑某个任务、还不确定 Claude 会做什么的时候。

options = ClaudeAgentOptions(
    permission_mode="default",  # 可以省略，这就是默认值
)

acceptEdits — 自动接受文件编辑

Claude 可以自由地读写文件，但执行 Bash 命令等操作仍需确认。

就像告诉实习生："代码你随便改，但要跑脚本的话先跟我说。"

options = ClaudeAgentOptions(
    permission_mode="acceptEdits",
)

这是日常开发最常用的模式。大部分时候，Claude 改代码不会出什么大问题， 但执行命令（尤其是 rm、git push 这类）还是让你过个眼比较好。

plan — 先展示计划

Claude 会先告诉你它打算做什么，等你批准后再执行。

就像实习生先写一份计划书："我打算先读取配置文件，然后修改第 42 行， 最后运行测试。" 你看了觉得没问题，说"行，干吧"。

options = ClaudeAgentOptions(
    permission_mode="plan",
)

适合任务比较复杂、你想对全局有把控的场景。

bypassPermissions — 全部放行

Claude 做什么都不需要你确认。完全自主。

options = ClaudeAgentOptions(
    permission_mode="bypassPermissions",
)

警告： 这个模式意味着 Claude 可以不经你同意执行任何操作。 只在以下条件全部满足时使用：

1. 运行在隔离的沙盒环境中（比如 Docker 容器）
2. 没有敏感数据可以被访问
3. 有其他安全措施兜底（比如预算限制、沙盒设置）

生产环境中，永远不要用这个模式。

3. can_use_tool 回调 — 自定义权限策略

四种内置模式是"粗粒度"的控制。如果你想精细控制——比如"读操作自动放行， 写操作要审批"，就需要 can_use_tool 回调。

基本原理

can_use_tool 是一个你自己写的异步函数。每次 Claude 想使用某个工具时， SDK 就会调用你的函数，问你："这个操作允许吗？"

from claude_agent_sdk import (
    ClaudeAgentOptions,
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext,
)

# 你的权限判断函数
async def my_permission_callback(
    tool_name: str,                    # 工具名称，如 "Read", "Write", "Bash"
    input_data: dict,                  # 工具的输入参数
    context: ToolPermissionContext,     # 上下文信息
) -> PermissionResultAllow | PermissionResultDeny:
    ...

options = ClaudeAgentOptions(
    can_use_tool=my_permission_callback,
)

三个参数的含义：

放行：PermissionResultAllow

返回 PermissionResultAllow() 表示"允许这个操作"：

async def allow_reads(tool_name, input_data, context):
    if tool_name == "Read":
        return PermissionResultAllow()  # 读操作一律放行

    # 其他操作拒绝
    return PermissionResultDeny(message="只允许读操作")

PermissionResultAllow 还有一个很有用的功能——修改工具输入：

async def redirect_writes(tool_name, input_data, context):
    if tool_name == "Write":
        # 把所有写操作重定向到安全目录
        original_path = input_data.get("file_path", "")
        safe_path = f"/tmp/sandbox/{original_path.lstrip('/')}"
        return PermissionResultAllow(
            updated_input={"file_path": safe_path, "content": input_data.get("content", "")}
        )
    return PermissionResultAllow()

这就像告诉实习生："你可以写文件，但只能写到 /tmp/sandbox/ 目录下。" Claude 以为自己写到了原始路径，实际上被你悄悄重定向了。

拒绝：PermissionResultDeny

返回 PermissionResultDeny() 表示"不允许"：

async def block_dangerous_commands(tool_name, input_data, context):
    if tool_name == "Bash":
        command = input_data.get("command", "")
        # 拦截危险命令
        dangerous_patterns = ["rm -rf", "mkfs", "dd if=", "> /dev/"]
        for pattern in dangerous_patterns:
            if pattern in command:
                return PermissionResultDeny(
                    message=f"危险命令被拦截: {pattern}",
                    interrupt=True,  # 中断整个对话
                )
    return PermissionResultAllow()

PermissionResultDeny 有两个参数：

大多数时候，interrupt=False（默认值）就够了。Claude 收到拒绝后， 通常会换一种更安全的方式来完成任务。只有遇到特别危险的操作， 你才需要设 interrupt=True 来强制停止。

实战：一个完整的权限策略

把上面的思路组合起来，就是一个实用的权限回调：

async def smart_permissions(tool_name, input_data, context):
    """
    权限策略：
    - 读操作：自动放行
    - 写操作：只允许写到项目目录
    - Bash 命令：拦截危险命令，其他放行
    """
    # 读操作一律放行
    if tool_name in ("Read", "Glob", "Grep"):
        return PermissionResultAllow()

    # 写操作：检查路径
    if tool_name in ("Write", "Edit"):
        file_path = input_data.get("file_path", "")
        allowed_dirs = ["/home/user/project/", "/tmp/"]
        if not any(file_path.startswith(d) for d in allowed_dirs):
            return PermissionResultDeny(
                message=f"只能写入项目目录或 /tmp，不能写入 {file_path}"
            )
        return PermissionResultAllow()

    # Bash 命令：拦截危险操作
    if tool_name == "Bash":
        command = input_data.get("command", "")
        blocklist = ["rm -rf", "sudo", "chmod 777", "curl | bash"]
        for bad in blocklist:
            if bad in command:
                return PermissionResultDeny(message=f"不允许执行: {bad}")
        return PermissionResultAllow()

    # 其他工具默认放行
    return PermissionResultAllow()

完整可运行代码见 examples/permission_demo.py

4. 沙盒设置 — 让 Claude 在围栏里干活

can_use_tool 是在 SDK 层面做权限控制。沙盒（Sandbox）则是在操作系统层面 给 Claude 画围栏——限制它能访问的文件系统和网络。

两者的区别：

建议两者结合使用：can_use_tool 做细粒度策略，沙盒做兜底防护。

开启沙盒

from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    sandbox={
        "enabled": True,                       # 开启沙盒
        "autoAllowBashIfSandboxed": True,      # 沙盒内自动放行 Bash 命令
    },
)

autoAllowBashIfSandboxed 很实用：既然 Bash 命令已经被限制在沙盒里了， 就不需要每次都问你"可以执行吗？"。

关键配置项

排除特定命令

有些命令需要访问沙盒外的资源。比如 git 要连远程仓库，docker 要访问守护进程：

options = ClaudeAgentOptions(
    sandbox={
        "enabled": True,
        "autoAllowBashIfSandboxed": True,
        "excludedCommands": ["git", "docker"],  # 这些命令不受沙盒限制
    },
)

网络控制

沙盒默认会限制网络访问。如果 Claude 需要访问某些网络资源，可以按需开放：

options = ClaudeAgentOptions(
    sandbox={
        "enabled": True,
        "autoAllowBashIfSandboxed": True,
        "network": {
            "allowUnixSockets": ["/var/run/docker.sock"],  # 允许 Docker socket
            "allowLocalBinding": True,                      # 允许绑定本地端口
        },
    },
)

完整可运行代码见 examples/sandbox_demo.py

5. 预算控制 — 防止烧钱

Claude 的 API 是按量计费的。如果 Claude 进入了某种"死循环"——不断调用工具、 不断生成回复——你的账单可能会很惊人。

SDK 提供了两把"刹车"：

max_budget_usd — 费用上限

options = ClaudeAgentOptions(
    max_budget_usd=0.50,  # 最多花 0.5 美元
)

超过这个金额，Claude 会自动停止。简单粗暴，但非常有效。

怎么定这个值？ 经验法则：

max_turns — 轮次上限

options = ClaudeAgentOptions(
    max_turns=10,  # 最多跑 10 轮
)

每次 Claude 调用工具算一轮。设个上限，防止 Claude 无限循环。

双重保险

建议两个一起用：

options = ClaudeAgentOptions(
    max_budget_usd=1.00,  # 最多花 1 美元
    max_turns=20,          # 最多跑 20 轮
)

哪个先到就停。这样既防了费用暴涨，也防了无限循环。

运行后检查花费

每次查询结束后，ResultMessage 里有实际花费：

from claude_agent_sdk import ResultMessage, query

async for message in query(prompt="重构这个文件", options=options):
    if isinstance(message, ResultMessage):
        print(f"本次花费: ${message.total_cost_usd:.4f}")
        print(f"对话轮数: {message.num_turns}")
        if message.total_cost_usd and message.total_cost_usd > 0.80:
            print("警告: 花费接近预算上限！")

完整可运行代码见 examples/budget_guard.py

6. 安全最佳实践清单

最后，总结一份安全清单。不管你用 SDK 做什么项目，都建议过一遍：

必做

• [ ] 始终设置 max_budget_usd：哪怕设得很高，也比不设好。防止意外烧钱。
• [ ] 生产环境不用 bypassPermissions：这个模式只用于完全隔离的沙盒环境。
• [ ] 设置 max_turns：防止 Claude 进入无限循环。一般 20-50 轮就够了。

建议做

• [ ] 用 can_use_tool 拦截危险命令：rm -rf、sudo、chmod 777 这类操作应该被拦截。
• [ ] 限制可写路径：用 can_use_tool 确保 Claude 只能写入特定目录。
• [ ] 开启沙盒：在支持的平台上（macOS/Linux），开启沙盒作为兜底防护。
• [ ] 检查 ResultMessage：每次查询后检查花费和轮数，记录日志。

高级

• [ ] 组合使用多层防护：权限模式 + can_use_tool + 沙盒 + 预算限制，层层设防。
• [ ] 审计工具使用日志：记录每次 can_use_tool 的调用，方便事后审查。
• [ ] 按环境切换配置：开发环境用 acceptEdits，CI 环境用沙盒 + 预算限制。

一个生产级配置示例

把本章所有知识点组合起来：

from claude_agent_sdk import (
    ClaudeAgentOptions,
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext,
)

# 危险命令黑名单
DANGEROUS_COMMANDS = [
    "rm -rf", "mkfs", "dd if=", "> /dev/",
    "sudo", "chmod 777", "curl | bash", "wget | sh",
]

# 允许写入的目录
ALLOWED_WRITE_DIRS = ["/home/user/project/", "/tmp/workspace/"]


async def production_permissions(tool_name, input_data, context):
    """生产环境权限策略"""

    # 读操作放行
    if tool_name in ("Read", "Glob", "Grep"):
        return PermissionResultAllow()

    # 写操作检查路径
    if tool_name in ("Write", "Edit"):
        path = input_data.get("file_path", "")
        if not any(path.startswith(d) for d in ALLOWED_WRITE_DIRS):
            return PermissionResultDeny(message=f"禁止写入: {path}")
        return PermissionResultAllow()

    # Bash 命令检查黑名单
    if tool_name == "Bash":
        cmd = input_data.get("command", "")
        for bad in DANGEROUS_COMMANDS:
            if bad in cmd:
                return PermissionResultDeny(
                    message=f"危险命令: {bad}",
                    interrupt=True,
                )
        return PermissionResultAllow()

    return PermissionResultAllow()


# 组装最终配置
production_options = ClaudeAgentOptions(
    permission_mode="acceptEdits",
    can_use_tool=production_permissions,
    max_budget_usd=2.00,
    max_turns=30,
    sandbox={
        "enabled": True,
        "autoAllowBashIfSandboxed": True,
        "excludedCommands": ["git"],
    },
)

这个配置做了五层防护：

1. 权限模式：acceptEdits 自动接受文件编辑，Bash 需确认
2. can_use_tool：精细控制每个工具的使用权限
3. 沙盒：操作系统级别的隔离
4. 预算：最多花 2 美元
5. 轮次：最多跑 30 轮

本章完整示例代码

以下是本章涉及的完整示例文件，可以直接复制运行。

permission_demo.py — can_use_tool 权限回调

import anyio

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    PermissionResultAllow,
    PermissionResultDeny,
    ResultMessage,
    TextBlock,
    ToolPermissionContext,
    query,
)

# ============================================================
# 配置区：可以根据你的项目需求修改
# ============================================================

# 允许 Claude 写入的目录列表
ALLOWED_WRITE_DIRS = [
    "/tmp/claude-workspace/",
    "/home/user/project/src/",
]

# 危险命令黑名单——包含这些字符串的 Bash 命令会被拦截
DANGEROUS_COMMANDS = [
    "rm -rf",       # 递归强制删除
    "rm -r /",      # 删除根目录
    "mkfs",         # 格式化磁盘
    "dd if=",       # 低级磁盘写入
    "> /dev/",      # 写入设备文件
    "sudo",         # 提权操作
    "chmod 777",    # 过度开放权限
    "curl | bash",  # 从网络下载并执行脚本
    "wget | sh",    # 同上
]


# ============================================================
# 权限回调函数
# ============================================================

async def smart_permission_callback(
    tool_name: str,
    input_data: dict,
    context: ToolPermissionContext,
) -> PermissionResultAllow | PermissionResultDeny:
    """
    自定义权限策略。

    参数:
        tool_name: Claude 想使用的工具名称（如 "Read", "Write", "Bash"）
        input_data: 工具的输入参数（如 {"file_path": "/src/main.py"}）
        context: 上下文信息，包含 signal 和 suggestions

    返回:
        PermissionResultAllow — 放行（可选：修改输入）
        PermissionResultDeny  — 拒绝（可选：附带原因、是否中断）
    """

    # ----------------------------------------------------------
    # 策略 1：读操作自动放行
    # ----------------------------------------------------------
    if tool_name in ("Read", "Glob", "Grep"):
        print(f"  [权限] 放行读操作: {tool_name}")
        return PermissionResultAllow()

    # ----------------------------------------------------------
    # 策略 2：写操作检查目标路径
    # ----------------------------------------------------------
    if tool_name in ("Write", "Edit"):
        file_path = input_data.get("file_path", "")

        # 检查是否在允许的目录内
        is_allowed = any(file_path.startswith(d) for d in ALLOWED_WRITE_DIRS)

        if is_allowed:
            print(f"  [权限] 放行写操作: {tool_name} -> {file_path}")
            return PermissionResultAllow()
        else:
            print(f"  [权限] 拒绝写操作: {tool_name} -> {file_path}")
            return PermissionResultDeny(
                message=f"不允许写入 {file_path}。只能写入以下目录: {ALLOWED_WRITE_DIRS}"
            )

    # ----------------------------------------------------------
    # 策略 3：Bash 命令拦截危险操作
    # ----------------------------------------------------------
    if tool_name == "Bash":
        command = input_data.get("command", "")

        for dangerous in DANGEROUS_COMMANDS:
            if dangerous in command:
                print(f"  [权限] 拦截危险命令: {command}")
                return PermissionResultDeny(
                    message=f"安全策略禁止执行包含 '{dangerous}' 的命令",
                    interrupt=True,  # 危险命令直接中断对话
                )

        print(f"  [权限] 放行 Bash 命令: {command[:50]}...")
        return PermissionResultAllow()

    # ----------------------------------------------------------
    # 兜底：其他工具默认放行
    # ----------------------------------------------------------
    print(f"  [权限] 放行其他工具: {tool_name}")
    return PermissionResultAllow()


# ============================================================
# 高级用法：修改工具输入（路径重定向）
# ============================================================

async def redirect_permission_callback(
    tool_name: str,
    input_data: dict,
    context: ToolPermissionContext,
) -> PermissionResultAllow | PermissionResultDeny:
    """
    高级示例：把所有写操作重定向到安全目录。

    Claude 以为自己写到了原始路径，实际上文件被写到了 /tmp/sandbox/ 下。
    这样既不影响 Claude 的工作流程，又保护了你的文件系统。
    """

    if tool_name == "Write":
        original_path = input_data.get("file_path", "")
        # 把路径重定向到安全目录
        safe_path = f"/tmp/sandbox{original_path}"
        print(f"  [重定向] {original_path} -> {safe_path}")
        return PermissionResultAllow(
            updated_input={
                "file_path": safe_path,
                "content": input_data.get("content", ""),
            }
        )

    # 其他工具正常放行
    return PermissionResultAllow()


# ============================================================
# 主程序
# ============================================================

async def main():
    print("=" * 60)
    print("can_use_tool 权限回调演示")
    print("=" * 60)

    # 组装配置
    options = ClaudeAgentOptions(
        permission_mode="acceptEdits",          # 基础模式：自动接受编辑
        can_use_tool=smart_permission_callback,  # 叠加自定义权限策略
        max_turns=5,                             # 限制轮数
        max_budget_usd=0.10,                     # 限制预算
    )

    prompt = "请读取当前目录下的文件列表，然后在 /tmp/claude-workspace/ 目录创建一个 hello.txt 文件"
    print(f"\n提示词: {prompt}\n")

    async for message in query(prompt=prompt, options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")

        if isinstance(message, ResultMessage):
            print(f"\n--- 完成 ---")
            print(f"轮数: {message.num_turns}")
            print(f"花费: ${message.total_cost_usd}")


if __name__ == "__main__":
    anyio.run(main)

运行方式：

python examples/permission_demo.py

sandbox_demo.py — 沙盒配置

import anyio

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


# ============================================================
# 配置示例 1：基础沙盒
# ============================================================

# 最简单的沙盒配置——开启沙盒，自动放行被沙盒限制的 Bash 命令
basic_sandbox_options = ClaudeAgentOptions(
    sandbox={
        "enabled": True,                    # 开启沙盒
        "autoAllowBashIfSandboxed": True,   # 沙盒内 Bash 命令自动放行
    },
    max_turns=5,
    max_budget_usd=0.10,
)


# ============================================================
# 配置示例 2：开发环境沙盒
# ============================================================

# 开发环境：沙盒开启，但 git 和 docker 可以绕过沙盒
dev_sandbox_options = ClaudeAgentOptions(
    permission_mode="acceptEdits",
    sandbox={
        "enabled": True,
        "autoAllowBashIfSandboxed": True,
        # git 需要访问 .git 目录和远程仓库，docker 需要访问守护进程
        # 这些命令放在排除列表中，不受沙盒限制
        "excludedCommands": ["git", "docker"],
    },
    max_turns=10,
    max_budget_usd=0.50,
)


# ============================================================
# 配置示例 3：严格沙盒（CI/CD 环境）
# ============================================================

# CI/CD 环境：沙盒 + 严格网络控制
strict_sandbox_options = ClaudeAgentOptions(
    permission_mode="bypassPermissions",   # CI 环境可以全部放行
    sandbox={
        "enabled": True,
        "autoAllowBashIfSandboxed": True,
        "excludedCommands": ["git"],
        # 不允许通过 dangerouslyDisableSandbox 绕过沙盒
        "allowUnsandboxedCommands": False,
    },
    max_turns=30,
    max_budget_usd=2.00,
)


# ============================================================
# 配置示例 4：Docker 环境沙盒
# ============================================================

# 在 Docker 容器中运行时，可能需要弱化的嵌套沙盒
docker_sandbox_options = ClaudeAgentOptions(
    sandbox={
        "enabled": True,
        "autoAllowBashIfSandboxed": True,
        # Docker 容器通常是非特权的，需要启用弱化沙盒（仅 Linux）
        "enableWeakerNestedSandbox": True,
        # 网络配置：允许访问 Docker socket 和绑定本地端口
        "network": {
            "allowUnixSockets": ["/var/run/docker.sock"],
            "allowLocalBinding": True,
        },
    },
    max_turns=20,
    max_budget_usd=1.00,
)


# ============================================================
# 配置示例 5：沙盒 + 忽略特定违规
# ============================================================

# 某些情况下，你可能需要忽略特定文件或网络的违规
relaxed_sandbox_options = ClaudeAgentOptions(
    sandbox={
        "enabled": True,
        "autoAllowBashIfSandboxed": True,
        # 忽略对特定文件和网络地址的访问违规
        "ignoreViolations": {
            "file": ["/proc/", "/sys/"],        # 忽略对 /proc 和 /sys 的访问
            "network": ["localhost", "127.0.0.1"],  # 忽略本地网络访问
        },
    },
    max_turns=10,
    max_budget_usd=0.50,
)


# ============================================================
# 主程序：演示基础沙盒
# ============================================================

async def main():
    print("=" * 60)
    print("沙盒配置演示")
    print("=" * 60)

    # 使用基础沙盒配置
    print("\n使用配置: 基础沙盒")
    print("  - 沙盒已开启")
    print("  - Bash 命令在沙盒内自动放行")
    print()

    prompt = "列出当前目录下的文件"

    async for message in query(prompt=prompt, options=basic_sandbox_options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")

        if isinstance(message, ResultMessage):
            print(f"\n--- 完成 ---")
            print(f"轮数: {message.num_turns}")
            print(f"花费: ${message.total_cost_usd}")


if __name__ == "__main__":
    anyio.run(main)

运行方式：

python examples/sandbox_demo.py

budget_guard.py — 预算和轮次控制

import anyio

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


# ============================================================
# 示例 1：简单问答 — 低预算
# ============================================================

async def simple_qa():
    """简单问答场景：预算 $0.05，最多 1 轮"""
    print("=" * 60)
    print("场景 1：简单问答（低预算）")
    print("=" * 60)

    options = ClaudeAgentOptions(
        max_budget_usd=0.05,  # 5 美分足够一次问答
        max_turns=1,           # 只问一次，不需要工具调用
    )

    async for message in query(prompt="Python 的 GIL 是什么？", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text[:200]}...")  # 截取前 200 字符

        if isinstance(message, ResultMessage):
            print(f"\n花费: ${message.total_cost_usd}")
            print(f"轮数: {message.num_turns}")


# ============================================================
# 示例 2：代码任务 — 中等预算
# ============================================================

async def code_task():
    """代码任务场景：预算 $0.50，最多 10 轮"""
    print("\n" + "=" * 60)
    print("场景 2：代码任务（中等预算）")
    print("=" * 60)

    options = ClaudeAgentOptions(
        max_budget_usd=0.50,  # 0.5 美元，够做一次代码审查
        max_turns=10,          # 最多 10 轮工具调用
        permission_mode="acceptEdits",  # 自动接受文件编辑
    )

    prompt = "请查看当前目录的结构，并简要描述这是什么项目"

    async for message in query(prompt=prompt, options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")

        if isinstance(message, ResultMessage):
            print(f"\n花费: ${message.total_cost_usd}")
            print(f"轮数: {message.num_turns}")

            # 检查是否接近预算上限
            if message.total_cost_usd and message.total_cost_usd > 0.40:
                print("注意: 花费已超过预算的 80%！")


# ============================================================
# 示例 3：预算追踪器
# ============================================================

class BudgetTracker:
    """
    预算追踪器：跨多次查询追踪总花费。

    SDK 的 max_budget_usd 是单次查询的限制。
    如果你需要跨多次查询控制总预算，可以用这个类。
    """

    def __init__(self, total_budget: float):
        self.total_budget = total_budget     # 总预算
        self.total_spent = 0.0               # 已花费
        self.query_count = 0                 # 查询次数

    @property
    def remaining(self) -> float:
        """剩余预算"""
        return max(0, self.total_budget - self.total_spent)

    def can_afford(self, estimated_cost: float = 0.05) -> bool:
        """检查是否还有足够预算进行下一次查询"""
        return self.remaining >= estimated_cost

    def record(self, cost: float | None, turns: int):
        """记录一次查询的花费"""
        actual_cost = cost or 0.0
        self.total_spent += actual_cost
        self.query_count += 1
        print(f"  [预算] 本次: ${actual_cost:.4f} | "
              f"累计: ${self.total_spent:.4f} / ${self.total_budget:.2f} | "
              f"剩余: ${self.remaining:.4f}")

    def get_per_query_budget(self) -> float:
        """计算单次查询应分配的预算"""
        # 保守策略：剩余预算的一半分给下一次查询
        return self.remaining / 2


async def tracked_budget_demo():
    """演示跨多次查询的预算追踪"""
    print("\n" + "=" * 60)
    print("场景 3：跨查询预算追踪")
    print("=" * 60)

    # 创建预算追踪器，总预算 $0.50
    tracker = BudgetTracker(total_budget=0.50)

    # 模拟多次查询
    questions = [
        "Python 的列表和元组有什么区别？",
        "什么时候该用字典，什么时候该用集合？",
        "请用一句话解释装饰器。",
    ]

    for i, question in enumerate(questions, 1):
        # 检查预算是否充足
        if not tracker.can_afford():
            print(f"\n预算不足，跳过第 {i} 个问题: {question}")
            break

        print(f"\n--- 问题 {i}: {question} ---")

        # 为单次查询分配预算
        per_query_budget = tracker.get_per_query_budget()

        options = ClaudeAgentOptions(
            max_budget_usd=per_query_budget,
            max_turns=1,
        )

        async for message in query(prompt=question, options=options):
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        # 只打印前 100 个字符
                        text = block.text[:100]
                        print(f"  Claude: {text}...")

            if isinstance(message, ResultMessage):
                tracker.record(message.total_cost_usd, message.num_turns)

    # 最终统计
    print(f"\n{'=' * 40}")
    print(f"总预算: ${tracker.total_budget:.2f}")
    print(f"总花费: ${tracker.total_spent:.4f}")
    print(f"查询次数: {tracker.query_count}")
    print(f"剩余预算: ${tracker.remaining:.4f}")


# ============================================================
# 主程序
# ============================================================

async def main():
    # 运行三个演示场景
    await simple_qa()
    await code_task()
    await tracked_budget_demo()


if __name__ == "__main__":
    anyio.run(main)

运行方式：

python examples/budget_guard.py

7. 小结

这一章我们学了四层安全防护：

核心思想很简单：不要把所有安全措施放在一个篮子里。 多层防护，层层递进，任何一层被突破，下一层还能兜住。

下一章，我们进入实战篇——开始设计一个真正的 AI Agent 项目架构。

本章文件清单

07-权限控制与安全/
  README.md                          # 你正在读的这个文件
  examples/
    permission_demo.py               # can_use_tool 回调完整示例
    sandbox_demo.py                  # 沙盒配置示例
    budget_guard.py                  # 预算和轮次控制示例