多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

learn-claude-code -s03 权限管线拆解:check_permission 在 agent loop 里怎么拦

learn-claude-code -s03 权限管线拆解:check_permission 在 agent loop 里怎么拦 1. 为什么 agent loop 里必须有一道权限闸门如果你跟着 learn-claude-code 的 s01、s02 一路写下来会发现一个很自然的状态模型返回tool_use程序拿到block.name和block.input直接去TOOL_HANDLERS里查 handler查到就执行。整个链路短、直接、跑得通。但这条链路有个隐患模型说删就删说写就写运行时没有任何否决权。s02 解决的是模型如何连续调用多个工具s03 解决的是另一个维度的问题——模型调用工具之前系统要不要允许它执行。这就是 Permission Pipeline 要干的事。learn-claude-code s03 的设计思路不是重写循环也不是重写工具系统而是在原来的工具分发前面加一道门。教程文档把它总结为三道闸门先硬拒绝再规则匹配最后需要时让用户审批三道都没命中才直接执行。这三道闸门对应三种结果。allow 是直接放行比如读取工作区内的 README.md只读操作不写文件、不跑 shell没必要拦。ask 是暂停询问比如删除某个本地临时目录不一定绝对禁止但可能造成不可逆影响不能自动执行。deny 是直接拒绝比如包含 root 删除、sudo、关机、格式化磁盘这类高危模式的命令不应该进入 handler更不应该让用户随手确认后执行。这套机制最重要的地方在于权限判断发生在工具执行之前。模型仍然可以提出一个工具调用但能不能真正跑起来由运行时决定。这就把模型建议和系统执行分开了——模型负责生成意图Permission Pipeline 负责判断这条意图是否允许落地。我试过把这段逻辑直接塞进 handler 内部做检查结果发现两个问题一是每个 handler 都要重复写一遍权限判断二是有些 handler 根本不该被调用检查写在里面已经晚了。s03 的做法是把检查提到 dispatch 之前用一个统一的check_permission(block)卡住所有工具调用干净很多。这篇文章会拆解check_permission在 agent loop 中的调用时机与拦截逻辑给出可复制的权限规则配置片段并演示一次被拦截和一次放行的验证动作。适合已经写完 s02、正在往 s03 推进的读者也适合想给自己的 agent 加一层运行时边界的开发者。2. TaoToken 前置准备让 agent loop 能稳定跑起来在拆权限管线之前得先保证你的 agent loop 能正常发请求、正常拿到tool_use。learn-claude-code 的示例代码用的是 Anthropic 风格的client.messages.create你需要一个能兼容这套接口的调用入口。TaoToken 在这里的角色是提供一个统一的 API 入口让你不用在多个模型供应商之间来回切换配置。它的 API 地址是https://taotoken.net/api兼容 Anthropic 的 messages 接口格式所以 learn-claude-code 里的client.messages.create基本可以原样保留只需要改 base_url 和 api_key。先拿到 API Key。打开https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi_keys创建一个新的 key复制出来。这个 key 后面会写进环境变量不要硬编码在代码里。然后确认你的模型 ID。learn-claude-code 里用的是一个MODEL常量你需要把它换成 TaoToken 支持的模型 ID。具体支持哪些模型可以在模型对话页面里看到当前可用的列表https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel_chat。选一个支持 tool_use 的模型因为 s03 的权限管线依赖模型返回tool_use类型的 block。接下来是环境变量配置。在项目根目录建一个.env文件或者直接 exportexport TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODEL你的模型ID然后在 Python 代码里初始化 clientimport os from anthropic import Anthropic client Anthropic( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) MODEL os.environ[TAOTOKEN_MODEL]这里有个容易踩的坑base_url不要带尾部斜杠也不要自己拼/v1/messagesAnthropic SDK 会自己处理路径。如果你写成了https://taotoken.net/api/v1请求路径会变成/api/v1/v1/messages直接 404。如果你用的是 Claude Code 或者类似的编码工具想把 TaoToken 接进去可以参考接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc。文档里有针对不同客户端的配置示例包括 Base URL、Key、Model ID 三件套的填法。对于长期跑 agent loop 的场景比如你要反复调试权限管线、跑多轮工具调用可以考虑用 Coding Plan避免按次计费带来的成本波动https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding_plan。配置好之后先跑一个最小请求验证连通性response client.messages.create( modelMODEL, systemYou are a helpful assistant., messages[{role: user, content: Say OK}], max_tokens100, ) print(response.content)如果能看到返回的 text block说明前置准备完成。接下来进入 s03 的核心把check_permission嵌进 agent loop。3. 可复制配置三道闸门的完整代码与规则片段s03 的权限管线由三部分组成硬拒绝列表、规则匹配、用户审批。最后用一个check_permission()把它们串起来。这一节给出可以直接复制进项目的代码以及一份可扩展的规则配置片段。先看第一道闸门硬拒绝。它对应DENY_LIST和check_deny_list()专门处理rm -rf /、sudo、shutdown这类永远不应该继续往下走的命令。只要这一层命中就直接得到 deny不会再进入用户审批。# Gate 1: Hard deny list — always forbidden DENY_LIST [ rm -rf /, sudo, shutdown, reboot, mkfs, dd if, /dev/sda, ] def check_deny_list(command: str) - str | None: for pattern in DENY_LIST: if pattern in command: return fBlocked: {pattern} is on the deny list return None注意这里的匹配方式是子串匹配不是正则。对于rm -rf /这种模式子串匹配已经够用而且不容易因为正则写错导致漏拦。如果你要加更复杂的模式比如匹配rm后面跟任意参数再跟根路径可以换成re.search但要小心正则回溯和误伤。第二道闸门是规则匹配对应PERMISSION_RULES和check_rules()。它处理的是上下文相关的风险操作比如写工作区外、执行删除命令、修改系统路径等。它和第一道闸门不同命中规则并不意味着立刻拒绝而是说明这次调用需要用户确认。from pathlib import Path WORKDIR Path.cwd() # Gate 2: Rule matching — context-dependent checks PERMISSION_RULES [ { tools: [write_file, edit_file], check: lambda args: not (WORKDIR / args.get(path, )).resolve().is_relative_to(WORKDIR), message: Writing outside workspace, }, { tools: [bash], check: lambda args: any( kw in args.get(command, ) for kw in [rm , /etc/, chmod 777] ), message: Potentially destructive command, }, ] def check_rules(tool_name: str, args: dict) - str | None: for rule in PERMISSION_RULES: if tool_name in rule[tools] and rule[check](args): return rule[message] return None这份规则配置片段可以直接扩展。比如你想禁止 agent 修改.git目录下的文件加一条{ tools: [write_file, edit_file], check: lambda args: .git in str(args.get(path, )), message: Modifying .git directory, },如果你想限制 bash 只能在工作区内操作可以加一条检查cd命令的目标路径{ tools: [bash], check: lambda args: cd / in args.get(command, ), message: Changing to root directory, },第三道闸门是用户审批对应ask_user()。当第二道闸门判断某次工具调用有风险时程序会暂停把工具名、参数和原因展示给用户让用户决定是否允许。用户允许才继续执行 handler用户拒绝则返回Permission denied.。# Gate 3: User approval — wait for confirmation after rule match def ask_user(tool_name: str, args: dict, reason: str) - str: print(f\n\033[33m {reason}\033[0m) print(f Tool: {tool_name}({args})) choice input( Allow? [y/N] ).strip().lower() return allow if choice in (y, yes) else deny最后三道闸门被check_permission()串起来def check_permission(block) - bool: if block.name bash: reason check_deny_list(block.input.get(command, )) if reason: print(f\n\033[31m⛔ {reason}\033[0m) return False reason check_rules(block.name, block.input) if reason: decision ask_user(block.name, block.input, reason) if decision deny: return False return True这段代码的顺序非常重要。第一步先查硬拒绝。如果是 bash先看命令里有没有命中DENY_LIST。命中就直接返回 False不执行不询问。第二步再查规则。如果工具名和参数命中某条风险规则就进入ask_user()。第三步如果用户拒绝也返回 False。如果用户允许或者没有命中任何规则就返回 True。所以check_permission()的返回值非常简单True 表示可以执行 handlerFalse 表示不允许执行 handler。如果你想把这份配置写成独立的 JSON 文件方便非开发者修改可以这样组织{ deny_list: [rm -rf /, sudo, shutdown, reboot, mkfs, dd if, /dev/sda], permission_rules: [ { tools: [write_file, edit_file], message: Writing outside workspace, type: path_outside_workdir }, { tools: [bash], message: Potentially destructive command, type: keyword_match, keywords: [rm , /etc/, chmod 777] } ] }然后在 Python 里加载这个 JSON把type映射到对应的检查函数。这样做的好处是规则和代码分离改规则不用动 Python 文件。但要注意JSON 里不能直接写 lambda所以需要一层映射逻辑。4. 验证请求一次被拦截与一次放行的完整过程配置写好了得验证它真的在工作。这一节演示两个场景一次被硬拒绝拦截一次被规则匹配后用户放行。通过这两个场景你能看到check_permission在 agent loop 里的实际调用时机。先看 agent loop 里接入权限管线的关键位置def agent_loop(messages: list): while True: response client.messages.create( modelMODEL, systemSYSTEM, messagesmessages, toolsTOOLS, max_tokens8000, ) messages.append({role: assistant, content: response.content}) if response.stop_reason ! tool_use: return results [] for block in response.content: if block.type ! tool_use: continue print(f\033[36m {block.name}\033[0m) # s03 change: run through permission pipeline before executing if not check_permission(block): results.append({ type: tool_result, tool_use_id: block.id, content: Permission denied., }) continue handler TOOL_HANDLERS.get(block.name) output handler(**block.input) if handler else fUnknown: {block.name} print(str(output)[:200]) results.append({ type: tool_result, tool_use_id: block.id, content: output, }) messages.append({role: user, content: results})关键就是这一行if not check_permission(block):它正好卡在模型生成tool_use之后、handler 执行之前。也就是说模型已经提出了工具调用但程序还没有真正碰文件系统、跑 shell、改代码。Permission Pipeline 就是在这个边界上做判断。如果权限拒绝程序不会静默跳过而是追加一个普通的tool_result内容是Permission denied.。这样模型下一轮会看到这个结果知道这次调用被拒绝了可以调整策略。现在验证第一个场景硬拒绝拦截。假设模型生成了一个 bash 工具调用命令是sudo rm -rf /tmp/test。这个命令会先经过check_deny_list因为包含sudo命中DENY_LIST直接返回 False。运行 agent loop你会看到 bash ⛔ Blocked: sudo is on the deny list然后results里会追加一条tool_result内容是Permission denied.。模型收到这个结果后下一轮不会再尝试执行这个命令而是可能换一种方式或者告诉你它无法完成。这里有个细节check_deny_list只对block.name bash生效。也就是说只有 bash 工具调用才会走硬拒绝检查。其他工具比如write_file不会经过DENY_LIST而是直接进入规则匹配。这是合理的因为DENY_LIST里的模式都是 shell 命令相关的。第二个场景规则匹配后用户放行。假设模型生成了一个 bash 工具调用命令是rm -rf ./tmp_cache。这个命令不包含sudo不命中DENY_LIST。但它包含rm会命中PERMISSION_RULES里的第二条规则返回Potentially destructive command。然后进入ask_user()程序暂停打印Potentially destructive command Tool: bash({command: rm -rf ./tmp_cache}) Allow? [y/N]如果你输入yask_user返回allowcheck_permission返回 Truehandler 执行删除./tmp_cache。如果你输入n或者直接回车返回denycheck_permission返回 False追加Permission denied.handler 不执行。实测下来这个交互式审批在调试阶段很有用但如果你要跑自动化任务input()会阻塞。这时候可以把ask_user改成读环境变量或者配置文件里的白名单比如设置AUTO_APPROVErm -rf ./tmp_cache命中白名单就自动放行。还有一个验证点只读操作应该直接放行。比如模型调用read_file读取工作区内的README.md。这个调用不经过DENY_LIST因为不是 bash也不命中任何PERMISSION_RULES因为read_file不在规则的工具列表里所以check_permission直接返回 Truehandler 执行返回文件内容。你可以手动构造一个测试from types import SimpleNamespace block SimpleNamespace( nameread_file, input{path: README.md}, idtest_1, ) print(check_permission(block)) # 应该输出 True再构造一个写工作区外的测试block SimpleNamespace( namewrite_file, input{path: /etc/hosts, content: test}, idtest_2, ) print(check_permission(block)) # 会触发 ask_user输入 n 后输出 False这两个测试能帮你确认权限管线的分支逻辑是否正确。5. 常见报错排查401、local proxy failed、reading choices、OAuth权限管线跑起来之后你可能会遇到一些报错。这一节整理几个高频问题对照真实报错给出排查方向。第一个是 401 Unauthorized。这个通常和权限管线无关而是 API Key 配置问题。报错长这样anthropic.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}排查步骤先确认TAOTOKEN_API_KEY环境变量是否设置正确有没有多余的空格或换行。然后确认 key 没有过期或被撤销。如果用的是.env文件确认加载顺序有时候 shell 里已经有一个旧的TAOTOKEN_API_KEY.env里的没覆盖上。可以在代码里打印os.environ.get(TAOTOKEN_API_KEY)[:8]看看前几位对不对。第二个是 local proxy failed。这个报错通常出现在你本地有代理设置但代理不可用的时候APIConnectionError: Connection error: local proxy failed排查方向检查HTTP_PROXY、HTTPS_PROXY、ALL_PROXY这些环境变量是否设置了一个不可用的代理。如果有先 unset 掉再跑。另外检查~/.anthropic/config或者项目里的配置文件有没有写死代理地址。TaoToken 的 API 地址是直连的不需要额外代理配置。第三个是 reading choices 相关报错。这个通常出现在流式响应或者响应格式不符合预期的时候KeyError: choices或者TypeError: NoneType object is not subscriptable (reading choices)这个报错的根源是你用的 SDK 或者代码期望的是 OpenAI 风格的响应有choices字段但实际拿到的是 Anthropic 风格的响应有content字段或者反过来。learn-claude-code 用的是 Anthropic SDK响应里是response.content不是response.choices。如果你在代码里混用了两种风格的解析逻辑就会出这个错。排查步骤确认你用的 client 是anthropic.Anthropic不是openai.OpenAI。确认base_url指向的是兼容 Anthropic 接口的地址。如果你在 TaoToken 上用的是 OpenAI 兼容接口那响应格式会不一样需要换解析方式。第四个是 OAuth 相关报错。如果你用的是 Claude Code 或者其他带 OAuth 登录的工具可能会遇到OAuth token expired或者Failed to refresh OAuth token这个和 API Key 模式不同。OAuth 模式需要定期刷新 token如果刷新失败就会报这个错。排查方向确认你的 OAuth 配置没有过期重新登录一次。如果你同时配置了 API Key 和 OAuth确认代码走的是哪条路径有时候环境变量里的 API Key 会覆盖 OAuth 配置导致预期外的行为。如果你在配置 Claude Code 接入 TaoToken 时遇到问题可以参考接入文档里的 Claude Code 部分https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc。文档里有 Base URL、Key、Model ID 的完整填法。还有一个和权限管线直接相关的报错Permission denied.出现在tool_result里但模型没有正确处理。这不是程序报错而是模型行为问题。有些模型收到Permission denied.之后会反复尝试同一个工具调用陷入循环。这时候你需要在 system prompt 里加一句If a tool call returns Permission denied., do not retry the same call. Try a different approach or ask the user for guidance.另外如果你发现check_permission总是返回 True检查一下block.name和block.input的取值。有些 SDK 版本里block.input可能是字符串而不是 dict这时候block.input.get(command, )会报AttributeError。可以在check_permission开头加一个类型检查if not isinstance(block.input, dict): return False这样至少不会因为格式问题导致权限检查被绕过。6. 把权限管线用起来从调试到长期运行s03 的 Permission Pipeline 代码量不大但系统意义很大。从这一节开始Agent 不再是模型提出工具调用程序立刻执行而是变成了模型提出工具调用运行时先判断权限再决定是否执行。这一节最值得记住的核心思想是不要信任模型要信任运行时边界。模型可以提出任何工具调用但真正能不能落地必须交给 harness 层判断。安全读取可以直接通过风险操作需要用户确认绝对危险的命令应该在 handler 执行前就被拦住。这也是 s03 相对于 s02 的最大升级s02 解决的是模型如何调用多个工具s03 解决的是模型调用工具之前系统要不要允许它执行。从这里开始Agent 的工具系统不再只是一个 dispatch map而开始具备最基础的权限治理能力。如果你要把这套权限管线用到长期运行的任务里有几个实用建议。第一把ask_user改成可配置的审批策略比如支持总是允许、总是拒绝、仅本次允许三种选项避免每次都要手动输入。第二把权限决策写进日志记录每次工具调用的名称、参数、决策结果和原因方便事后审计。第三定期 reviewPERMISSION_RULES随着 agent 能力扩展可能需要加新的规则也可能需要放宽某些规则。如果你在调试权限管线时需要反复发请求验证模型行为可以用模型对话页面快速测试https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel_chat。对于长期跑 agent 任务的场景Coding Plan 能提供更稳定的调用额度https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding_plan。最后一步把check_permission的返回值从 bool 改成更细的结构比如返回{decision: allow | deny | ask, reason: ...}这样在 agent loop 里可以根据不同决策做不同处理而不是简单地追加Permission denied.。这个改动不大但能让权限管线更灵活。
返回列表