工具鉴权、审批绑定与幂等账本
网络超时之后,调用方不知道发信是失败还是已经成功。重试模型会再次提出同一动作,所以“调用 ID 不同”不代表业务意图不同。幂等键应对应业务动作,并绑定租户、参数和版本;相同键配不同参数必须拒绝。
在事务内保存结果
以下使用 SQLite 内存数据库展示本地业务写入与幂等账本的同一事务。它没有实际发送邮件,也不宣称解决远程 API 的 exactly-once。完整保存运行:
python
import sqlite3, json, hashlib
c = sqlite3.connect(":memory:")
c.executescript("CREATE TABLE actions(tenant TEXT, key TEXT, digest TEXT, result TEXT, PRIMARY KEY(tenant,key)); CREATE TABLE drafts(id INTEGER PRIMARY KEY, body TEXT);")
def digest(payload):
return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
def create_draft(tenant, key, payload, approval):
if tenant != "a" or set(payload) != {"body"} or not isinstance(payload["body"], str):
raise PermissionError("not allowed")
d = digest(payload)
if approval != (tenant, key, d):
raise PermissionError("approval does not bind action")
with c:
row = c.execute("SELECT digest,result FROM actions WHERE tenant=? AND key=?", (tenant,key)).fetchone()
if row:
if row[0] != d: raise ValueError("key reused with changed payload")
return row[1]
cur = c.execute("INSERT INTO drafts(body) VALUES (?)", (payload["body"],))
result = str(cur.lastrowid)
c.execute("INSERT INTO actions VALUES (?,?,?,?)", (tenant,key,d,result))
return result
p = {"body":"hello"}
permit = ("a", "job-7", digest(p))
assert create_draft("a","job-7",p,permit) == create_draft("a","job-7",p,permit)
assert c.execute("SELECT count(*) FROM drafts").fetchone()[0] == 1
print("one draft")先鉴权再查账本,避免撤权后仍从历史结果泄露数据。审批绑定内容摘要,因此改正文后旧审批不能沿用。生产审批还需批准人、有效期、资源版本以及服务端防伪存储;本例 tuple 只是契约,不是可由客户端自签的凭据。
远程副作用的未知状态
数据库与外部邮件服务不能共用本地事务。常见做法是事务内写 outbox,worker 发送时使用供应商幂等键;超时先按键查状态,再决定重发。供应商不支持幂等和查询时,只能进入 unknown 并人工核对,不能承诺绝不重复。取消后也要查已发生的效果,补偿不是回到“什么都没发生”:撤销通知本身可能被用户看到。
工具包装器限制超时、结果大小和错误码。URL 工具需防内网访问与重定向绕过,路径工具需按解析后的真实路径检查目录范围,代码工具使用独立沙箱与资源限额。模型描述和 schema 帮助选择工具,服务端用户、租户与业务状态校验才授予能力。
练习
在正文后尝试使用原 key 和原审批把 body 改为 bye,应 PermissionError,draft 数量仍为 1;再为新内容重新审批但沿用原 key,应 ValueError。