第二章的执行器前面,再放什么?
用户要求先读 main.py,并在运行前先询问。第二章的 execute_tool 已经会查名称、检查参数、执行函数,但没有接收宿主决定的位置。思想篇已经区分了参数合法和获得许可,现在把这个区别写进代码。
这次不改循环的消息与停止规则。给原来的 execute(call) 换一个入口:先保存调用副本,取得权限决定,只有 allow 才交给第二章分发器;拒绝则直接返回关联的未执行反馈。
正式实现由本册独立编写,位于公开仓库的 examples/s03-permission/。三个函数分别负责:
permission_for检查请求和宿主规则,只产生allow / ask / deny与理由。authorize_call只在ask时调用宿主,取得最终允许或拒绝。execute_guarded保存副本并守住真实执行入口,返回与原 ID 对应的结果。
下面展示从公开源文件按符号提取的完整函数。代码断点和图共用同一个游标。浏览器中的宿主回答、文件和进程是可展开查看的教学替身;默认目录中的公开 CLI 才会真实读文件,获准后真实运行 Python。
permission_for:规则只做判断
先确定本章的最小策略:读取工具放行,运行工具询问,禁止项和没有权限规则的工具拒绝。将禁止项作为宿主传入的 blocked,而不是允许模型在参数里填写一个“已经批准”的字段。
权限判断之前,仍要验证工具名和参数。为复用第二章已有的检查,本次把它提取为 validate_call:返回错误字符串或 None。第二章的执行器和第三章的权限入口使用同一个检查函数,字段集合与类型规则保持原样。
真实执行轨迹回放;宿主回答、文件与进程使用公开教学替身,不发请求、不启动文件。
def permission_for(call, tools, blocked=()):
"""只判断,不提问、不执行。没有策略的工具默认拒绝。"""
error = validate_call(call, tools)
if error:
return {"decision": "deny", "reason": error}
name = call["name"]
if name in blocked:
return {"decision": "deny", "reason": "宿主策略禁止这个工具。"}
if name == "read_file":
return {"decision": "allow", "reason": "允许读取;目录约束仍由读取函数检查。"}
if name == "run_main":
return {"decision": "ask", "reason": "运行 main.py 前需要用户允许。"}
return {"decision": "deny", "reason": "这个工具尚未配置权限策略。"}点击行号设置断点,再点“调试”;也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。
from copy import deepcopy
_scenario = 'read'
_executions = []
_approvals = []
def _read_file(path):
_executions.append("read_file")
return "print(6 * 7)\n"
def _run_main():
_executions.append("run_main")
return "exit_code=0\n42"
tools = {"read_file": {"parameters": {"path": str}, "run": _read_file},
"run_main": {"parameters": {}, "run": _run_main}}
blocked = ("run_main",) if _scenario == "blocked" else ()
call = {"type": "tool_use", "id": "run_1", "name": "run_main", "input": {}}
if _scenario == "read":
call.update(id="read_1", name="read_file", input={"path": "main.py"})
if _scenario == "invalid":
call["input"] = {"approved": True}
def ask_user(shown, reason):
_approvals.append(deepcopy(shown))
if _scenario == "mutated":
shown.update(name="unknown", input={"path": "changed"})
if _scenario == "not_boolean": return "yes"
return _scenario in ("approved", "mutated")
def validate_call(call, tools):
"""本课小型参数契约;供分发与后续权限入口共用。"""
name, arguments = call.get("name"), call.get("input")
if not isinstance(name, str) or name not in tools:
return "未知工具。"
parameters = tools[name]["parameters"]
if not isinstance(arguments, dict) or set(arguments) != set(parameters):
return "参数字段不匹配。"
if any(type(arguments[key]) is not kind for key, kind in parameters.items()):
return "参数类型不匹配。"
return None
def execute_tool(call, tools):
"""校验与执行;每个返回值保留原调用 ID。"""
result = {"type": "tool_result", "tool_use_id": call["id"]}
error = validate_call(call, tools)
if error:
return {**result, "content": error, "is_error": True}
try:
output = tools[call["name"]]["run"](**call["input"])
return {**result, "content": output, "is_error": False}
except (ValueError, OSError) as error:
return {**result, "content": str(error), "is_error": True}
def authorize_call(call, tools, ask_user, blocked=()):
"""只有 ask 才调用宿主;必须收到布尔 True 才允许。"""
verdict = permission_for(call, tools, blocked)
if verdict["decision"] != "ask":
return verdict
# 宿主看到调用副本;不能通过修改显示对象改变后续执行参数。
answer = ask_user(deepcopy(call), verdict["reason"])
if answer is True:
return {"decision": "allow", "reason": "用户允许本次调用。"}
return {"decision": "deny", "reason": "本次调用未获用户允许。"}
# 调用入口
outcome = permission_for(call, tools, blocked)
print(outcome["decision"])- 校验名称和参数。
- 优先检查宿主禁止项。
- 读取返回allow;运行返回ask;无策略的工具拒绝。
先看“运行询问”,在返回 ask 的行停下。这里还没有宿主调用,也没有程序执行。再切到“策略禁止”:它会先于运行询问规则返回 deny。
“参数无效”在 run_main 的空参数里加入 approved=True,也会被拒绝。这个字段来自请求,不能给请求本身授权。新注册的工具即使能够通过参数检查,没有对应策略也会返回 deny;不会因为开发者忘了配置就默认放行。
authorize_call:只有 ask 才询问宿主
permission_for 返回的 allow 和 deny 可以直接使用,只有 ask 缺少宿主回答。authorize_call 把这条分支单独收在一个函数里,避免每个工具各写一套询问逻辑。
ask_user 由宿主注入。本章要求它返回真正的布尔值;模型文本、工具参数或字符串 "yes" 不能被顺手转换成允许。
真实执行轨迹回放;宿主回答、文件与进程使用公开教学替身,不发请求、不启动文件。
def authorize_call(call, tools, ask_user, blocked=()):
"""只有 ask 才调用宿主;必须收到布尔 True 才允许。"""
verdict = permission_for(call, tools, blocked)
if verdict["decision"] != "ask":
return verdict
# 宿主看到调用副本;不能通过修改显示对象改变后续执行参数。
answer = ask_user(deepcopy(call), verdict["reason"])
if answer is True:
return {"decision": "allow", "reason": "用户允许本次调用。"}
return {"decision": "deny", "reason": "本次调用未获用户允许。"}点击行号设置断点,再点“调试”;也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。
from copy import deepcopy
_scenario = 'approved'
_executions = []
_approvals = []
def _read_file(path):
_executions.append("read_file")
return "print(6 * 7)\n"
def _run_main():
_executions.append("run_main")
return "exit_code=0\n42"
tools = {"read_file": {"parameters": {"path": str}, "run": _read_file},
"run_main": {"parameters": {}, "run": _run_main}}
blocked = ("run_main",) if _scenario == "blocked" else ()
call = {"type": "tool_use", "id": "run_1", "name": "run_main", "input": {}}
if _scenario == "read":
call.update(id="read_1", name="read_file", input={"path": "main.py"})
if _scenario == "invalid":
call["input"] = {"approved": True}
def ask_user(shown, reason):
_approvals.append(deepcopy(shown))
if _scenario == "mutated":
shown.update(name="unknown", input={"path": "changed"})
if _scenario == "not_boolean": return "yes"
return _scenario in ("approved", "mutated")
def validate_call(call, tools):
"""本课小型参数契约;供分发与后续权限入口共用。"""
name, arguments = call.get("name"), call.get("input")
if not isinstance(name, str) or name not in tools:
return "未知工具。"
parameters = tools[name]["parameters"]
if not isinstance(arguments, dict) or set(arguments) != set(parameters):
return "参数字段不匹配。"
if any(type(arguments[key]) is not kind for key, kind in parameters.items()):
return "参数类型不匹配。"
return None
def execute_tool(call, tools):
"""校验与执行;每个返回值保留原调用 ID。"""
result = {"type": "tool_result", "tool_use_id": call["id"]}
error = validate_call(call, tools)
if error:
return {**result, "content": error, "is_error": True}
try:
output = tools[call["name"]]["run"](**call["input"])
return {**result, "content": output, "is_error": False}
except (ValueError, OSError) as error:
return {**result, "content": str(error), "is_error": True}
def permission_for(call, tools, blocked=()):
"""只判断,不提问、不执行。没有策略的工具默认拒绝。"""
error = validate_call(call, tools)
if error:
return {"decision": "deny", "reason": error}
name = call["name"]
if name in blocked:
return {"decision": "deny", "reason": "宿主策略禁止这个工具。"}
if name == "read_file":
return {"decision": "allow", "reason": "允许读取;目录约束仍由读取函数检查。"}
if name == "run_main":
return {"decision": "ask", "reason": "运行 main.py 前需要用户允许。"}
return {"decision": "deny", "reason": "这个工具尚未配置权限策略。"}
# 调用入口
outcome = authorize_call(call, tools, ask_user, blocked)
print("实际执行:" + ("、".join(_executions) or "无"))- 取得策略结果。
- allow/deny直接返回,不询问。
- ask才把调用副本交给宿主。
- 仅布尔True返回allow,其余返回deny。
在 answer = ask_user(...) 处停下,图会指向宿主确认。这行执行完以后才取得回答。选“用户允许”时返回 allow,选“用户拒绝”或“无有效回答”时返回 deny。
切到“策略禁止”,宿主确认这一行不会命中;返回的仍是策略的禁止结果。这个分支没有再问一次,也就不会用一次确认覆盖禁止项。
公开 CLI 的真实询问函数如下:
def terminal_approval(call, reason):
print(f"等待允许:{call['id']} · {call['name']} · {call['input']}\n{reason}")
if not sys.stdin.isatty():
print("非交互输入,按未获允许处理")
return False
try:
return input("允许这一次?[y/N] ").strip().lower() == "y"
except EOFError:
return False终端会显示工具名、参数、调用 ID 和理由。交互终端里输入 y 才允许;非交互环境和输入结束按未获允许处理。ask_user 调用期间同步等待,循环不会偷偷再调用模型。这是单进程的同步交互,没有持久化待审批任务或跨进程恢复。
execute_guarded:把决定落实到同一份调用上
最终执行入口只接受 allow。它先对请求做 deepcopy,用这份 snapshot 完成权限判断和实际执行。authorize_call 给宿主显示的又是一个副本,因此确认界面修改显示对象,不会把另一个工具或参数塞进真正的执行请求。
副本只隔离调用数据。宿主代码与工具注册表仍是可信的应用代码;这个设计不把任意插件代码变成安全的确认组件。
真实执行轨迹回放;宿主回答、文件与进程使用公开教学替身,不发请求、不启动文件。
def execute_guarded(call, tools, ask_user, blocked=()):
"""验证、显示与执行使用同一份调用快照;拒绝结果不冒充执行结果。"""
snapshot = deepcopy(call)
verdict = authorize_call(snapshot, tools, ask_user, blocked)
if verdict["decision"] != "allow":
return {
"type": "tool_result", "tool_use_id": snapshot["id"],
"content": "未执行:" + verdict["reason"], "is_error": True,
}
return execute_tool(snapshot, tools)点击行号设置断点,再点“调试”;也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。
from copy import deepcopy
_scenario = 'approved'
_executions = []
_approvals = []
def _read_file(path):
_executions.append("read_file")
return "print(6 * 7)\n"
def _run_main():
_executions.append("run_main")
return "exit_code=0\n42"
tools = {"read_file": {"parameters": {"path": str}, "run": _read_file},
"run_main": {"parameters": {}, "run": _run_main}}
blocked = ("run_main",) if _scenario == "blocked" else ()
call = {"type": "tool_use", "id": "run_1", "name": "run_main", "input": {}}
if _scenario == "read":
call.update(id="read_1", name="read_file", input={"path": "main.py"})
if _scenario == "invalid":
call["input"] = {"approved": True}
def ask_user(shown, reason):
_approvals.append(deepcopy(shown))
if _scenario == "mutated":
shown.update(name="unknown", input={"path": "changed"})
if _scenario == "not_boolean": return "yes"
return _scenario in ("approved", "mutated")
def validate_call(call, tools):
"""本课小型参数契约;供分发与后续权限入口共用。"""
name, arguments = call.get("name"), call.get("input")
if not isinstance(name, str) or name not in tools:
return "未知工具。"
parameters = tools[name]["parameters"]
if not isinstance(arguments, dict) or set(arguments) != set(parameters):
return "参数字段不匹配。"
if any(type(arguments[key]) is not kind for key, kind in parameters.items()):
return "参数类型不匹配。"
return None
def execute_tool(call, tools):
"""校验与执行;每个返回值保留原调用 ID。"""
result = {"type": "tool_result", "tool_use_id": call["id"]}
error = validate_call(call, tools)
if error:
return {**result, "content": error, "is_error": True}
try:
output = tools[call["name"]]["run"](**call["input"])
return {**result, "content": output, "is_error": False}
except (ValueError, OSError) as error:
return {**result, "content": str(error), "is_error": True}
def permission_for(call, tools, blocked=()):
"""只判断,不提问、不执行。没有策略的工具默认拒绝。"""
error = validate_call(call, tools)
if error:
return {"decision": "deny", "reason": error}
name = call["name"]
if name in blocked:
return {"decision": "deny", "reason": "宿主策略禁止这个工具。"}
if name == "read_file":
return {"decision": "allow", "reason": "允许读取;目录约束仍由读取函数检查。"}
if name == "run_main":
return {"decision": "ask", "reason": "运行 main.py 前需要用户允许。"}
return {"decision": "deny", "reason": "这个工具尚未配置权限策略。"}
def authorize_call(call, tools, ask_user, blocked=()):
"""只有 ask 才调用宿主;必须收到布尔 True 才允许。"""
verdict = permission_for(call, tools, blocked)
if verdict["decision"] != "ask":
return verdict
# 宿主看到调用副本;不能通过修改显示对象改变后续执行参数。
answer = ask_user(deepcopy(call), verdict["reason"])
if answer is True:
return {"decision": "allow", "reason": "用户允许本次调用。"}
return {"decision": "deny", "reason": "本次调用未获用户允许。"}
# 调用入口
outcome = execute_guarded(call, tools, ask_user, blocked)
print("实际执行:" + ("、".join(_executions) or "无"))- 保存调用副本。
- 以副本完成权限判断。
- 拒绝直接返回关联的未执行反馈。
- 只有allow才把同一份副本交给实际分发器。
先选“拒绝运行”,在最终的 return execute_tool(...) 设置断点。这条执行行不会命中,函数从旁路返回 is_error=True、原调用 ID 和“未执行”的理由。选“允许运行”后,重新在这一行设置断点,才会停在实际执行前;此时许可已经取得,但高亮行还没有执行。
“修改显示副本”输入会在宿主替身里修改显示对象,再返回允许。继续到执行行,真正的 snapshot 仍为 run_main({})。图和变量展示的是这份实际快照,而不是另一套手动步骤。
如果真实程序获准后自己报错,返回的是执行错误;如果权限拒绝,返回的是未执行反馈。宿主确认函数异常时,异常继续向上传递,执行行不会被调用,不能把接口故障当成默认允许。
接回循环,只替换执行依赖
装配时继续使用第二章的 build_tools(workspace),原来的读取和进程实现也继续使用。新入口只替换循环中的执行依赖:
tools = build_tools(workspace)
execute = lambda call: execute_guarded(call, tools, ask_user, blocked)
outcome = run_loop(ask, execute, max_calls=4, question=QUESTION)这一段省略了 CLI、模型客户端和终端输入的组装;完整入口为 user_code/practice.py。ask 仍接收完整消息记录,返回模型响应。它无权自己生成宿主允许结果。
下载公开仓库后,Python 3.10+ 标准库即可运行离线入口。先固定一次拒绝,再固定一次允许,便于对照:
python3 examples/s03-permission/user_code/practice.py --decision deny
python3 examples/s03-permission/user_code/practice.py --decision allow
python3 examples/s03-permission/user_code/verify_practice.py这两个 --decision 是教学运行的固定宿主回答,不是模型参数,也不是永久权限配置;一次回答被消费后,后续询问默认拒绝。不带它时,交互终端会真正询问用户。模型默认仍为固定响应,文件读取是真实的,程序只在获得允许后真实启动。
两条路径都有三次模型判断,但第二次判断后的证据不同:
| 宿主决定 | 第二次工具结果 | 最终能报告什么 |
|---|---|---|
| 拒绝 | run_1,is_error=True,未执行 | 已读文件,本次没有运行 |
| 允许 | run_1,退出码为 0,输出 42 | 实际程序输出 42 |
本课测试用一个会写入临时标记文件的程序核对“是否启动过”:允许时标记存在;用户拒绝、策略禁止、参数错误和确认接口异常时,标记都不存在。它证明拒绝确实阻止了执行,不只是界面换了一句提示。
再故意触发两条边界
运行 --decision allow --block-run。宿主禁止项先命中,固定的允许回答不会被消费,程序也不会启动。开发者应先查规则顺序,不要为了消除提示而移动到执行之后再判断。
再用 --max-calls 1 --decision allow。第一轮只完成读取,随后达到调用上限,尚未请求运行,也没有发生确认或进程执行。权限允许不取消预算;反复请求被拒绝的工具也仍然会被调用上限截住。
如果需要真实模型,按公开 README 配置 SDK 和凭据后显式使用 --live。本章已验证离线模型、真实文件与进程、权限拒绝和 SDK 模拟 HTTP;没有在线模型验证。许可层没有限制已启动程序的操作系统权限,也没有锁定等待期间可能变化的文件内容。
至此,用户的“运行前问我”有了一个确切位置:从 ask 到宿主回答,再到 execute_tool。只要沿着这三处断点追踪,就能判断请求是还在等待、已经被拒绝,还是确实进入了工具实现。
资料核对于 2026-10-18:课程 Permission提供执行前的权限判断思路;官方权限文档说明模型指令与宿主强制规则的区别。本册的规则、同步回调和调用快照由本册自行实现。