Agent 工程体系 · 第 14/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

Agent 写操作确认：参数预览、确认令牌、过期和防重放

Agent 调用只读工具时，错误通常表现为“答案不准确”；调用写工具时，错误可能直接变成退款、发邮件、删数据、改权限或提交订单。因此，写操作不能被实现成“模型产生函数调用，服务端立即执行”这一条直线。

写操作确认是一个受约束的两阶段协议：

Agent 先提出一个待执行的写意图；
系统解析、校验并生成可供人检查的参数预览；
用户明确确认这一次、这一组参数、这一个目标；
系统签发与具体操作绑定的确认令牌；
执行端在令牌未过期、未被使用、调用者匹配、参数摘要匹配的条件下执行一次；
执行结果通过幂等请求键和回执关联起来。

这里的“确认”不是 UI 上的一个按钮，而是对一次具体操作的授权证明。按钮只是产生确认请求的界面；真正的安全边界必须由服务端验证。

OpenAI 的 Function Calling 文档将工具调用描述为一个多步循环：模型请求调用工具，应用侧执行代码，再把工具输出传回模型；模型返回的工具调用本身只是请求，不是外部系统已经执行的事实。(developers.openai.com) MCP 也把工具定义为服务端提供的能力，并要求客户端通过 tools/call 请求调用；在敏感操作场景下，MCP 安全建议包括向用户展示工具输入并请求确认。(modelcontextprotocol.io)

一、先区分四件容易混淆的事

1. 工具调用不是写入

工具调用是 Agent 或模型向应用发送的结构化请求，例如：

{
  "tool": "refund_order",
  "arguments": {
    "order_id": "ord_1001",
    "amount": 19900,
    "currency": "CNY",
    "reason": "商品破损"
  }
}

这段数据表达的是：

“建议调用 refund_order，参数是这些值。”

它不表达：

“退款已经完成。”

应用侧必须自行决定是否允许执行。OpenAI 的工具调用流程明确要求应用接收模型的 tool call 后，在应用侧执行代码，再向模型发送 tool output。(developers.openai.com)

写操作是会改变外部状态的工具调用，包括但不限于：

创建、修改或删除数据库记录；
退款、转账、扣款、下单；
发送邮件、短信、站内信；
修改权限、密钥或配置；
发布内容、合并代码、触发部署；
代表用户向第三方系统提交申请。

读操作通常返回已有状态；写操作会改变未来状态。确认机制只应该围绕后者建立，但不能假设所有工具名称都能可靠表达副作用。update_user、sync_data、execute_sql 都可能隐藏写入，因此工具注册表应显式记录副作用等级，而不是仅凭名称判断。

side_effect = none       # 纯计算或只读
side_effect = reversible  # 可撤销，但已经产生外部影响
side_effect = irreversible # 不可逆或高代价
side_effect = external    # 对外发送、扣款、发布等

这是应用内部的策略字段，不是 OpenAI Function Calling 或 MCP 自动提供的安全保证。MCP 工具可以带 annotations 描述工具行为，但规范同时要求客户端把工具注解视为不可信信息，除非工具来自可信服务端。(modelcontextprotocol.io)

2. 参数校验不是人工确认

参数校验回答：

参数是否满足类型、格式、范围和业务约束？

例如：

amount 是整数
amount > 0
currency = CNY
order_id 存在
订单属于当前用户
退款金额不超过可退金额

人工确认回答：

有权限的用户是否看到了这一次具体操作，并明确同意执行？

二者不能互相替代：

通过参数校验，不代表用户看过参数；
用户点击确认，不代表参数仍然有效；
UI 显示了参数，不代表服务端执行时使用的参数没有被篡改；
确认了一个订单，不代表可以把令牌改成另一个订单。

因此，正确顺序是：

模型输出
  → 结构解析
  → 参数规范化
  → JSON Schema 校验
  → 业务授权与状态校验
  → 生成参数预览
  → 用户确认
  → 服务端验证确认令牌
  → 再次校验业务状态
  → 幂等执行

OpenAI 的 strict: true 可以让函数调用更可靠地遵循函数参数 Schema；其要求包括对象设置 additionalProperties: false，并将属性标记为必填，想表达可选值时使用包含 null 的类型。(developers.openai.com) 但严格 Schema 只约束结构，不会替应用判断“这个用户能否退款”或“现在是否允许退款”。

3. 确认令牌不是登录令牌

登录令牌证明：

调用者是谁，或者调用者具有什么会话身份。

确认令牌证明：

这个身份曾经确认过某一个具体的操作摘要，而且该确认仍在有效期内。

确认令牌必须至少绑定：

subject       用户或主体身份
audience      哪个应用、租户或服务可使用
operation     哪个工具或操作
resource      哪个资源，例如订单 ID
args_digest   参数规范化后的摘要
policy        使用的确认策略版本
issued_at     签发时间
expires_at    过期时间
token_id      令牌唯一标识
request_key   幂等请求键

如果令牌只包含“用户已确认退款”，攻击者就可能把它用于另一个订单或另一个金额。确认的对象必须是操作实例，而不是一个抽象动作。

4. 过期和防重放解决不同问题

过期解决时间边界：

用户确认后过了太久，这次确认是否仍然有效？

防重放解决使用次数和重复提交：

同一个有效确认是否可以被重复发送？

一个过期令牌可能从未被使用；一个未过期令牌也可能被重复使用。因此二者必须同时实现：

可执行 = 签名有效
      ∧ 身份匹配
      ∧ 参数摘要匹配
      ∧ 当前时间 < expires_at
      ∧ token_id 尚未消费
      ∧ request_key 的执行状态允许继续
      ∧ 当前业务状态仍满足约束

二、参数预览：用户确认的到底是什么

1. 参数预览的定义

参数预览是系统将即将执行的写操作转换成用户可理解、可核对的表示。它至少应包含：

操作名称和人类可读说明；
目标资源；
每个关键参数的实际值；
金额、单位、时区、时间范围等解释；
预计影响；
不可逆性或外部发送范围；
触发该操作的上下文；
预览生成时间和有效期。

例如，不能只显示：

即将调用 refund_order，是否确认？

因为用户无法判断订单、金额和原因。

更合适的预览是：

操作：为订单 ord_1001 退款
对象：杭州云栖路 88 号，订单尾号 1001
金额：人民币 199.00 元
原因：商品破损
影响：退款将原路退回，不再重复执行
预览时间：2026-09-01 14:20:00 +08:00
有效期至：2026-09-01 14:25:00 +08:00

这里的“可理解”不是把值改写得更好看，而是不能在改写过程中改变语义。例如金额应该同时保留机器值和展示值：

{
  "amount_minor": 19900,
  "currency": "CNY",
  "display_amount": "¥199.00"
}

display_amount 只能用于界面；执行端必须使用经过校验的 amount_minor 和 currency，不能从展示字符串反解析金额。

2. 预览必须来自规范化参数

设模型原始参数为 $a$ ，规范化函数为 $N$ ，业务校验函数为 $V$ ，则预览应基于：

a^{*} = N(a)

并且只有在：

V(a^{*}, u, s) = \text{true}

时才能生成待确认操作。

其中：

$a$ ：模型产生的原始参数；
$N$ ：类型转换、默认值填充、字段排序、单位明确化等规范化过程；
$a^{*}$ ：真正准备执行的参数；
$u$ ：当前用户身份和权限；
$s$ ：生成预览时观察到的业务状态；
$V$ ：参数和业务规则校验。

例如，用户说“退款 199 元”，模型可能输出：

{"amount": 199}

系统不能直接假设这是 199 元人民币。规范化过程应要求货币明确：

{
  "amount_minor": 19900,
  "currency": "CNY"
}

如果货币无法从上下文确定，就应该进入澄清流程，而不是进入确认流程。确认不能替代澄清：用户确认“199”并不能补足单位缺失。

3. 摘要必须基于确定性序列化

为了让确认令牌绑定具体参数，需要计算参数摘要：

d = H(\text{canonical}(a^{*}))

其中：

$H$ ：密码学哈希函数，例如 SHA-256；
canonical：确定性序列化；
$d$ ：参数摘要。

普通 JSON 文本不适合直接摘要，因为以下两个对象语义相同但文本不同：

{"order_id":"ord_1001","amount_minor":19900}

{"amount_minor":19900,"order_id":"ord_1001"}

工程上应采用明确的规范化规则，例如：

对象键按字典序排序；
数字使用固定表示；
字符串使用 UTF-8；
不允许未定义的额外字段；
时间统一为带时区或 UTC 的 RFC 3339；
金额使用最小货币单位整数；
不把展示字段纳入执行参数。

Python 示例：

import hashlib
import json

def canonical_json(value: dict) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")

def args_digest(args: dict) -> str:
    return hashlib.sha256(canonical_json(args)).hexdigest()

args1 = {
    "order_id": "ord_1001",
    "amount_minor": 19900,
    "currency": "CNY",
}

args2 = {
    "currency": "CNY",
    "amount_minor": 19900,
    "order_id": "ord_1001",
}

assert args_digest(args1) == args_digest(args2)
print(args_digest(args1))

预期结果是两个参数对象产生相同摘要。这里的前提是两个对象已经通过同一套规范化规则；如果一个对象中的 199 被解释为元、另一个对象中的 199 被解释为分，那么它们在规范化后必须不同。

4. 预览和执行必须使用同一个参数快照

一个常见错误是：

模型生成退款金额 199 元；
UI 显示退款 199 元；
用户点击确认；
服务端重新调用模型，让模型“生成最终参数”；
用第二次生成的参数执行。

这会导致用户确认的是 $a_1$ ，执行的却是 $a_2$ 。只要：

a_1 \neq a_2

确认就不应成立。

正确做法是保存不可变的待确认快照：

proposal_id = prop_abc
canonical_args = {"amount_minor":19900,...}
args_digest = sha256(...)

确认只引用 proposal_id，执行端从服务端存储取回原始规范化参数，不能相信浏览器回传的金额、订单号或工具名。

三、确认令牌的结构和签发

1. 两种常见令牌形式

确认令牌可以采用两类设计。

不透明令牌

令牌本身是随机字符串，例如：

ctk_7b4d...random...

服务端数据库保存完整内容：

token_id → subject, operation, args_digest, expires_at, status

优点是易于撤销、内容不泄露、便于修改内部字段。缺点是执行端需要访问令牌存储。

自包含签名令牌

令牌包含载荷并使用服务端密钥签名，例如类似 JWT 的结构：

{
  "iss": "approval-service",
  "aud": "refund-service",
  "sub": "user_42",
  "jti": "tok_abc",
  "op": "refund_order",
  "resource": "ord_1001",
  "args_digest": "sha256:...",
  "iat": 1788243600,
  "exp": 1788243900
}

签名可以防止字段被修改，但签名本身不能防重放。服务端仍需持久化 jti 或其他唯一标识，记录它是否已经被消费。

因此，实际生产系统常使用：

签名载荷 + 一次性消费记录

或直接使用：

高熵随机令牌 + 服务端状态

不要把“不可伪造”误认为“不可重复使用”。

2. 令牌必须绑定主体和受众

确认令牌应绑定当前用户、租户和执行服务：

sub = user_42
tenant = tenant_hz
aud = refund-service

执行端至少检查：

token.sub == 当前认证用户
token.tenant == 当前租户
token.aud == 当前工具服务

否则可能出现跨用户、跨租户或跨服务误用。

MCP 的工具调用是 JSON-RPC 请求，工具输入由 inputSchema 描述；MCP 对需要跨调用保存的状态建议使用显式句柄，而不是依赖连接隐式状态，并要求服务端在每次调用时重新验证授权。(modelcontextprotocol.io) 这对确认令牌同样适用：proposal_id 或 token_id 是显式状态标识，但不是天然的授权证明。

3. 签发确认令牌的条件

设待确认提案为 $p$ ，当前用户为 $u$ ，当前业务状态为 $s$ ，则签发条件可以形式化为：

\text{Issue}(p,u,s) \iff \text{ValidSchema}(p.args) \land \text{Authorized}(u,p) \land \text{Previewed}(p) \land \text{ExplicitAccept}(u,p) \land \text{Fresh}(p,s)

其中 Fresh 表示预览所依据的状态仍然足够新。例如退款提案生成时订单可退金额是 199 元，但等待五分钟后订单已经被人工退款 100 元，那么原提案不应直接签发或执行 199 元退款。

确认令牌的载荷可以这样生成：

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import secrets

@dataclass(frozen=True)
class Proposal:
    proposal_id: str
    subject: str
    tenant: str
    operation: str
    resource_id: str
    args: dict
    digest: str
    expires_at: datetime
    state_version: int

def create_proposal(subject: str, tenant: str, operation: str,
                    resource_id: str, args: dict,
                    state_version: int) -> Proposal:
    now = datetime.now(timezone.utc)
    return Proposal(
        proposal_id="prop_" + secrets.token_urlsafe(18),
        subject=subject,
        tenant=tenant,
        operation=operation,
        resource_id=resource_id,
        args=args,
        digest=args_digest(args),
        expires_at=now + timedelta(minutes=5),
        state_version=state_version,
    )

这段代码只生成提案，不执行写操作。前置条件是 args 已经完成类型、权限和业务校验；state_version 必须来自读取订单状态时的版本号，而不是由客户端自行填写。

四、过期：不只是检查一个时间戳

1. 过期时间的含义

令牌有效区间通常定义为：

iat \leq now < exp

采用右开区间的好处是边界明确：当 now == exp 时已经过期。

不要使用客户端时间判断过期。浏览器时间可以被修改，移动设备时间可能漂移，跨服务器也可能存在时钟偏差。过期判断应由签发服务或执行服务使用可信服务器时间完成。

2. 预览过期、令牌过期和业务状态过期

这三种过期应分开：

preview_expired  = 预览展示时间过长
token_expired    = 确认令牌超过 exp
state_stale      = 订单、库存、余额等业务状态发生变化

即使令牌没有过期，业务状态也可能失效：

14:20:00  可退金额 = 199 元，版本 = 7
14:20:10  用户确认
14:20:20  另一个渠道退款 100 元，版本 = 8
14:20:25  Agent 执行原提案

如果执行端只检查令牌时间，就可能超额退款。执行前应使用乐观并发控制：

UPDATE orders
SET refunded_minor = refunded_minor + :amount,
    version = version + 1
WHERE order_id = :order_id
  AND version = :expected_version
  AND refunded_minor + :amount <= payable_minor;

预期结果：

更新行数为 1：状态版本仍匹配，且业务约束成立；
更新行数为 0：状态已变化、订单不存在或退款金额超限，必须停止执行并重新预览。

这比“先查询，再无条件更新”安全，因为后者在查询和更新之间存在竞态窗口。

3. 过期后的恢复路径

过期不能简单返回“失败”。系统需要区分：

EXPIRED            用户需要重新确认
STATE_CHANGED      需要重新读取状态并生成新预览
ALREADY_CONSUMED   不能再次执行，应查询原执行结果
REVOKED            提案被撤销，不应自动重试

重新确认时不能只延长旧令牌的 exp。如果业务状态或参数已经改变，应生成新的 proposal_id、新的摘要和新的确认事件。

五、防重放：一次确认只能授权一次执行意图

1. 什么是重放

重放是攻击者或故障重试逻辑再次提交一份以前有效的确认请求。

典型来源包括：

用户连续点击两次确认；
前端超时后自动重试；
网关重试请求；
Worker 执行成功但响应丢失，随后重新消费消息；
攻击者截获有效令牌；
Agent 在收到超时后重复发出同一个工具调用；
多个 Agent 分支同时使用同一提案。

如果退款接口没有幂等和令牌消费控制，结果可能是：

第一次请求：退款 199 元，成功
第二次请求：退款 199 元，成功

2. 原子消费令牌

令牌必须在执行前被一次性占用，并且占用动作要和检查原子化。伪代码如下：

def consume_token(token_id: str, subject: str, digest: str) -> str:
    """
    返回 execution_id。
    若令牌不存在、身份不匹配、摘要不匹配、已过期或已消费，则抛出异常。
    """
    with db.transaction():
        token = db.query_one_for_update(
            "SELECT * FROM approval_tokens WHERE token_id = %s",
            [token_id],
        )

        if token is None:
            raise TokenError("unknown_token")

        if token.subject != subject:
            raise TokenError("subject_mismatch")

        if token.args_digest != digest:
            raise TokenError("args_mismatch")

        if token.expires_at <= db.server_now():
            raise TokenError("expired")

        if token.status != "issued":
            raise TokenError("already_consumed")

        execution_id = "exec_" + secrets.token_urlsafe(18)

        db.execute(
            """
            UPDATE approval_tokens
            SET status = 'consumed',
                consumed_at = CURRENT_TIMESTAMP,
                execution_id = %s
            WHERE token_id = %s
              AND status = 'issued'
            """,
            [execution_id, token_id],
        )

        return execution_id

SELECT ... FOR UPDATE 或等价的原子条件更新保证两个并发请求不会都观察到 status = issued。如果数据库更新影响行数不是 1，第二个请求必须得到 already_consumed，不能继续调用下游系统。

不过，先消费令牌再调用下游仍存在一个故障窗口：

令牌已消费
进程在调用支付系统前崩溃

此时不能简单把令牌状态改回 issued，否则可能在“下游其实已经成功”的情况下重复扣款。解决办法是引入执行记录和恢复查询。

3. 令牌消费和幂等请求键必须同时存在

确认令牌控制“用户是否确认过”。

幂等请求键控制“同一个执行意图在下游只产生一个逻辑结果”。

可以将两者表示为：

approval_token_id = tok_abc
request_key       = refund:tenant_hz:ord_1001:proposal_123
execution_id      = exec_xyz

建议在数据库中设置唯一约束：

CREATE TABLE action_executions (
    execution_id       TEXT PRIMARY KEY,
    request_key        TEXT NOT NULL UNIQUE,
    token_id           TEXT NOT NULL UNIQUE,
    operation          TEXT NOT NULL,
    args_digest        TEXT NOT NULL,
    status             TEXT NOT NULL,
    provider_ref       TEXT,
    result_json        JSONB,
    created_at         TIMESTAMP NOT NULL,
    updated_at         TIMESTAMP NOT NULL
);

状态可以是：

prepared
running
succeeded
failed_retryable
failed_terminal
unknown

其中 unknown 很重要：

请求已发给支付服务
本地没有收到响应

此时系统不能把它当成失败并重新发起另一笔退款，而应使用同一个 request_key 查询下游状态，或者进入人工处理。

4. 防重放的完整条件

一次执行请求 $x$ 被接受的条件可以写成：

\text{Accept}(x) \iff \begin{cases} \text{VerifySignature}(x.token) \\ \land\ \text{MatchSubject}(x) \\ \land\ \text{MatchAudience}(x) \\ \land\ \text{MatchOperation}(x) \\ \land\ \text{MatchDigest}(x) \\ \land\ \text{NotExpired}(x.token) \\ \land\ \text{AtomicallyConsume}(x.token) \\ \land\ \text{CreateOrReuseExecution}(x.request\_key) \end{cases}

若 request_key 已存在：

状态为 succeeded：返回原成功回执；
状态为 running：返回处理中，不再次调用下游；
状态为 unknown：查询或升级，不自动复制请求；
状态为 failed_retryable：使用同一请求键重试；
参数摘要不同：拒绝，说明同一个请求键被错误复用。

这解释了为什么“给接口加一个随机 request ID”还不够：如果重试时每次生成新的 ID，服务端无法判断它们是不是同一个逻辑操作。

六、推荐状态机：提案、确认、执行和回执分开

一个完整的写操作状态机可以设计为：

stateDiagram-v2
    [*] --> Proposed: 模型产生写意图
    Proposed --> Rejected: Schema/权限/业务校验失败
    Proposed --> AwaitingConfirmation: 生成参数预览
    AwaitingConfirmation --> Declined: 用户拒绝
    AwaitingConfirmation --> Cancelled: 用户取消
    AwaitingConfirmation --> Expired: 超过预览有效期
    AwaitingConfirmation --> Confirmed: 用户明确确认
    Confirmed --> Revoked: 策略或管理员撤销
    Confirmed --> Executing: 原子消费确认令牌
    Executing --> Succeeded: 下游成功且写入回执
    Executing --> FailedRetryable: 网络/临时故障
    Executing --> FailedTerminal: 参数或业务永久失败
    Executing --> Unknown: 下游结果未知
    FailedRetryable --> Executing: 使用同一请求键恢复
    Unknown --> Executing: 查询下游状态后恢复
    Succeeded --> [*]
    Declined --> [*]
    Cancelled --> [*]
    Expired --> [*]
    Rejected --> [*]
    Revoked --> [*]
    FailedTerminal --> [*]

几个状态不能合并：

Declined 是用户明确拒绝；
Cancelled 是用户没有作出明确决定；
Expired 是系统撤销了等待中的确认资格；
FailedTerminal 是已经尝试执行，但业务永久拒绝；
Unknown 是外部世界可能已经改变，本地不能确定。

如果把所有非成功状态都返回成 false，Agent 可能把用户拒绝误判成临时失败并自动重试，或者把结果未知误判成未执行而重复扣款。

七、并发路径：确认按钮不是串行锁

1. 双击确认

两个 HTTP 请求同时到达：

请求 A：读取 token.status = issued
请求 B：读取 token.status = issued
请求 A：执行退款
请求 B：执行退款

如果读取和更新不是原子的，双击就会导致双执行。

正确顺序应该是：

请求 A：原子地将 issued → consumed，成功
请求 B：原子地将 issued → consumed，失败
请求 A：执行或恢复 execution_id
请求 B：返回 already_consumed 或同一 execution_id 的状态

2. 两个 Agent 分支

并行工具调用会放大问题。OpenAI 文档说明，在支持的模型和配置下，模型可能在同一轮产生多个函数调用；可以通过 parallel_tool_calls: false 将行为限制为零个或一个工具调用。(developers.openai.com)

但关闭并行调用只解决模型输出层面的并行，不解决：

两个不同 Agent 实例同时执行；
用户在两个浏览器标签页点击确认；
消息队列重复投递；
网关重试；
另一个系统直接调用同一写接口。

因此，最终一致性必须由服务端的令牌消费、请求键唯一约束和业务行版本控制保证，而不能依赖模型只发一次调用。

3. 批量确认的边界

批量操作不能默认把一个确认视为无限范围授权。例如：

确认“删除这 1000 条日志”

必须明确：

resource_set_digest = H(排序后的资源 ID 集合)
count = 1000
filter = 固定后的查询条件

如果确认后集合发生变化，执行端应拒绝继续使用旧令牌。否则用户确认的是集合 $S_1$ ，执行的可能是扩展后的集合 $S_2$ ，而：

S_1 \subset S_2

这会形成权限范围扩大攻击。

更安全的选择是将批量操作拆成：

固定资源 ID 列表；
服务端生成批次快照；
令牌绑定批次摘要；
执行按批次逐项使用同一 request_key 派生键；
失败项独立回执，不自动扩大范围。

八、一个端到端实现骨架

下面的示例用纯 Python 展示核心生命周期，不依赖具体 Web 框架或支付 SDK。它不是完整生产服务，但每一步都对应真实的协议边界。

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import hashlib
import json
import secrets


def canonical_json(value: dict) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def digest_args(value: dict) -> str:
    return hashlib.sha256(canonical_json(value)).hexdigest()


@dataclass
class Order:
    order_id: str
    user_id: str
    payable_minor: int
    refunded_minor: int
    version: int


@dataclass
class Proposal:
    proposal_id: str
    user_id: str
    operation: str
    args: dict
    args_digest: str
    expected_order_version: int
    expires_at: datetime
    status: str = "awaiting_confirmation"


@dataclass
class Execution:
    execution_id: str
    request_key: str
    status: str
    result: dict | None = None


class DemoService:
    def __init__(self):
        self.orders = {
            "ord_1001": Order(
                order_id="ord_1001",
                user_id="user_42",
                payable_minor=19900,
                refunded_minor=0,
                version=7,
            )
        }
        self.proposals: dict[str, Proposal] = {}
        self.tokens: dict[str, dict] = {}
        self.executions: dict[str, Execution] = {}

    def propose_refund(
        self,
        user_id: str,
        order_id: str,
        amount_minor: int,
        currency: str,
        reason: str,
    ) -> Proposal:
        order = self.orders.get(order_id)

        if order is None or order.user_id != user_id:
            raise ValueError("order_not_found_or_not_owned")

        if currency != "CNY":
            raise ValueError("unsupported_currency")

        if amount_minor <= 0:
            raise ValueError("amount_must_be_positive")

        refundable = order.payable_minor - order.refunded_minor
        if amount_minor > refundable:
            raise ValueError("amount_exceeds_refundable_balance")

        args = {
            "order_id": order_id,
            "amount_minor": amount_minor,
            "currency": currency,
            "reason": reason,
        }

        proposal = Proposal(
            proposal_id="prop_" + secrets.token_urlsafe(12),
            user_id=user_id,
            operation="refund_order",
            args=args,
            args_digest=digest_args(args),
            expected_order_version=order.version,
            expires_at=datetime.now(timezone.utc) + timedelta(minutes=5),
        )
        self.proposals[proposal.proposal_id] = proposal
        return proposal

    def confirm(self, user_id: str, proposal_id: str) -> str:
        proposal = self.proposals.get(proposal_id)

        if proposal is None:
            raise ValueError("proposal_not_found")

        if proposal.user_id != user_id:
            raise ValueError("subject_mismatch")

        if proposal.status != "awaiting_confirmation":
            raise ValueError("proposal_not_confirmable")

        if datetime.now(timezone.utc) >= proposal.expires_at:
            proposal.status = "expired"
            raise ValueError("proposal_expired")

        token_id = "ctk_" + secrets.token_urlsafe(18)
        self.tokens[token_id] = {
            "token_id": token_id,
            "proposal_id": proposal_id,
            "user_id": user_id,
            "operation": proposal.operation,
            "args_digest": proposal.args_digest,
            "expires_at": proposal.expires_at,
            "status": "issued",
        }
        proposal.status = "confirmed"
        return token_id

    def execute_refund(self, user_id: str, token_id: str) -> Execution:
        token = self.tokens.get(token_id)

        if token is None:
            raise ValueError("unknown_token")

        if token["user_id"] != user_id:
            raise ValueError("subject_mismatch")

        if token["status"] != "issued":
            # 实际系统应根据 token 记录返回原 execution_id
            raise ValueError("already_consumed")

        if datetime.now(timezone.utc) >= token["expires_at"]:
            token["status"] = "expired"
            raise ValueError("token_expired")

        proposal = self.proposals[token["proposal_id"]]
        if digest_args(proposal.args) != token["args_digest"]:
            raise ValueError("args_digest_mismatch")

        # 令牌消费应在数据库事务中完成。
        token["status"] = "consumed"

        args = proposal.args
        request_key = (
            f"refund:{user_id}:{args['order_id']}:{proposal.proposal_id}"
        )

        existing = self.executions.get(request_key)
        if existing is not None:
            return existing

        execution = Execution(
            execution_id="exec_" + secrets.token_urlsafe(12),
            request_key=request_key,
            status="running",
        )
        self.executions[request_key] = execution

        order = self.orders[args["order_id"]]

        # 模拟乐观并发控制。
        if order.version != proposal.expected_order_version:
            execution.status = "failed_terminal"
            execution.result = {"code": "order_state_changed"}
            return execution

        refundable = order.payable_minor - order.refunded_minor
        if args["amount_minor"] > refundable:
            execution.status = "failed_terminal"
            execution.result = {"code": "amount_exceeds_refundable_balance"}
            return execution

        # 真实系统中，这里必须把 provider request key 设置为 request_key。
        order.refunded_minor += args["amount_minor"]
        order.version += 1

        execution.status = "succeeded"
        execution.result = {
            "order_id": order.order_id,
            "refunded_minor": args["amount_minor"],
            "currency": args["currency"],
        }
        return execution

调用过程：

service = DemoService()

proposal = service.propose_refund(
    user_id="user_42",
    order_id="ord_1001",
    amount_minor=19900,
    currency="CNY",
    reason="商品破损",
)

print(proposal.args)
# {
#   'order_id': 'ord_1001',
#   'amount_minor': 19900,
#   'currency': 'CNY',
#   'reason': '商品破损'
# }

token = service.confirm(
    user_id="user_42",
    proposal_id=proposal.proposal_id,
)

result = service.execute_refund(
    user_id="user_42",
    token_id=token,
)

print(result.status)
# succeeded

这个示例的关键点不在于数据结构，而在于生命周期：

propose_refund()
  → 固化参数和订单版本
confirm()
  → 用户确认提案，签发一次性令牌
execute_refund()
  → 校验令牌、摘要、过期时间和版本
  → 使用请求键创建执行记录
  → 执行一次业务写入
  → 返回结构化回执

示例仍有一个生产级限制：内存字典不能提供跨进程原子性。部署多个 Worker 后，必须将提案、令牌和执行记录放进支持事务和唯一约束的持久化存储中；下游支付或订单系统也必须接收稳定的幂等请求键。

九、与 Function Calling 和 MCP 的衔接

1. OpenAI Function Calling 层

可以将写工具暴露给模型，但工具实现不要直接执行高风险写入。模型产生的调用先进入 propose 路径：

{
  "type": "function",
  "function": {
    "name": "request_refund_confirmation",
    "description": "创建退款确认预览，不会立即退款",
    "strict": true,
    "parameters": {
      "type": "object",
      "properties": {
        "order_id": {
          "type": "string"
        },
        "amount_minor": {
          "type": "integer"
        },
        "currency": {
          "type": "string",
          "enum": ["CNY"]
        },
        "reason": {
          "type": "string"
        }
      },
      "required": [
        "order_id",
        "amount_minor",
        "currency",
        "reason"
      ],
      "additionalProperties": false
    }
  }
}

工具描述应明确：

该工具只创建人工确认提案，不执行退款。

这会降低模型误以为“调用成功就代表退款完成”的概率，但真正的禁止执行仍要在服务端实现。Schema 和描述是模型接口约束，不是权限边界。

如果一次写操作需要用户确认，应用可以让模型先收到如下工具结果：

{
  "proposal_id": "prop_abc",
  "status": "awaiting_confirmation",
  "requires_user_confirmation": true,
  "preview": {
    "operation": "退款",
    "order_id": "ord_1001",
    "amount": "¥199.00",
    "reason": "商品破损"
  }
}

模型随后可以向用户解释预览；用户确认通过应用自己的 UI 或会话协议完成，而不是把自然语言“确认”直接当成数据库写入条件。

2. MCP Tool 层

MCP 的 tools/call 请求包含工具名和参数；工具定义通过 inputSchema 描述输入，工具也可以通过 outputSchema 描述结构化输出。(modelcontextprotocol.io)

因此，MCP 服务可以把工具拆成两个显式操作：

request_refund_confirmation
  输入：订单、金额、原因
  输出：proposal_id、预览、expires_at

execute_confirmed_refund
  输入：proposal_id、confirmation_token
  输出：execution_id、status、receipt

拆分的意义是让协议层可观察：

第一个工具是“创建待确认状态”；
第二个工具是“消费确认并执行”；
两个工具的参数和权限不同；
审计日志能够区分“建议写入”和“实际写入”。

MCP 规范允许工具调用返回 InputRequiredResult，表示工具在完成前还需要额外输入；客户端重试时可以携带 inputResponses 和 requestState，且重试使用不同的 JSON-RPC id。(modelcontextprotocol.io) 这可以承载澄清或人工输入，但需要注意：

JSON-RPC 的 id 只关联协议请求与响应，不是业务幂等键，也不是确认令牌。

不能因为重试使用了不同的 JSON-RPC id，就把它们当成两次不同业务操作；业务层仍应携带稳定的 proposal_id 和 request_key。

3. MCP Elicitation 不是自动的写授权

MCP Elicitation 用于服务端通过客户端向用户请求额外信息，支持表单模式和 URL 模式。表单模式适合结构化非敏感输入；密码、API Key、访问令牌和支付凭证等敏感信息不得通过表单模式收集，应使用 URL 模式。(modelcontextprotocol.io)

Elicitation 的响应区分：

{"action": "accept"}
{"action": "decline"}
{"action": "cancel"}

其中：

accept 表示用户接受这次交互；
decline 表示用户明确拒绝；
cancel 表示用户关闭或中断，没有作出明确决定。

MCP 文档还特别说明，URL 模式中的 accept 只表示用户同意进行外部交互，不代表外部流程已经完成；客户端重试原请求后，服务端仍需根据状态判断是否完成。(modelcontextprotocol.io)

因此，Elicitation 的 accept 不能直接等价于：

approval_token = valid

应用仍需：

将用户响应绑定到具体工具调用和用户身份；
检查用户看到的字段是否与执行摘要一致；
生成或消费一次性确认令牌；
在执行端再次检查授权、过期和业务状态。

十、失败表现和诊断方法

1. 用户确认了，但服务端提示参数不一致

常见原因：

UI 使用展示金额生成摘要，执行端使用最小单位金额；
JSON 键顺序或数字格式不一致；
服务端在确认后重新补充了默认字段；
前端把 proposal_id 和参数一起提交，服务端错误地相信了前端参数；
模型第二次生成了不同参数。

诊断应记录：

proposal_id
token_id
expected_args_digest
received_args_digest
canonical_args_version
operation
subject

不要在日志中记录完整的密码、令牌、支付凭证或不必要的个人信息。摘要用于比较，不应被当成可逆数据。

2. 用户看到成功，但操作执行了两次

检查以下位置：

确认令牌消费是否原子化
token_id 是否有唯一约束
request_key 是否稳定
下游请求是否传递同一个幂等键
超时后是否错误生成了新 request_key
消息队列是否重复投递
多实例之间是否共享令牌存储

尤其要检查下游系统：本地数据库的幂等记录不能阻止已经发出的外部请求被重复处理，除非外部系统也支持相同的幂等语义或可查询状态。

3. 令牌过期后仍然执行成功

可能原因：

只在签发时检查过期，没有在执行时检查；
使用了客户端时间；
时区转换错误；
将 expires_at 当作本地时间保存；
执行任务进入队列时有效，真正消费时没有重新校验；
令牌已过期，但 Worker 使用了已经反序列化到内存中的旧状态。

过期检查必须位于真正产生副作用之前，而不是只放在创建提案或展示 UI 的接口中。

4. 用户只说“好的”，Agent 就执行了

自然语言确认存在歧义：

“好的，继续看看”
“可以，给我解释一下”
“这个金额有问题”
“先保留”

系统应定义明确的确认语义，并将确认动作绑定到当前 proposal_id。对于高风险操作，建议要求结构化事件：

{
  "action": "confirm",
  "proposal_id": "prop_abc",
  "ui_session_id": "ui_789"
}

如果用户修改了金额、订单或收款账户，这不是对旧提案的确认，而是对新参数的修改，必须重新校验并生成新预览。

十一、生产取舍：确认强度应随风险变化

不是所有写操作都需要相同的确认强度。可以定义风险函数：

R = f(v, i, r, e, u)

其中：

$v$ ：金额或影响范围；
$i$ ：不可逆程度；
$r$ ：权限敏感度；
$e$ ：外部传播范围；
$u$ ：用户和设备可信度。

低风险操作可以使用会话内一次点击确认；高风险操作可以要求：

显示完整目标和参数；
重新输入关键字段；
二次身份认证；
多人审批；
短有效期；
禁止批量扩大范围；
执行前重新读取状态；
执行后发送回执。

但确认强度不能通过隐藏参数来提高安全性。例如，把金额从预览中隐藏，要求用户只点击“同意”，并不会更安全，反而破坏了用户确认的证据质量。

十二、责任边界：谁决定、谁执行、谁回执

一个可审计的写操作至少应记录四类主体：

requester    谁提出了操作
approver     谁确认了操作
executor     哪个服务实际执行
provider     哪个外部系统接受了请求

以及四类数据：

requested_args
approved_args_digest
executed_args_digest
provider_result

理想情况下：

\text{requested\_digest} = \text{approved\_digest} = \text{executed\_digest}

如果不相等，系统必须拒绝或产生明确的人工升级事件，而不是继续执行。

最终回执不能只写：

成功

至少要包含：

{
  "execution_id": "exec_xyz",
  "request_key": "refund:tenant_hz:ord_1001:prop_abc",
  "operation": "refund_order",
  "status": "succeeded",
  "args_digest": "sha256:...",
  "provider_ref": "refund_9001",
  "completed_at": "2026-09-01T06:25:30Z"
}

回执中的 provider_ref 用于对账和恢复；execution_id 用于内部追踪；args_digest 用于证明执行参数与确认参数一致。

写操作确认的核心不是“让 Agent 先问一句再调用工具”，而是把一次外部改变拆成可验证的状态转换：

意图 ≠ 预览
预览 ≠ 确认
确认 ≠ 执行
执行 ≠ 成功
成功响应 ≠ 唯一执行

只有当参数快照、用户身份、确认令牌、过期时间、业务版本、幂等请求键和最终回执彼此关联时，系统才能回答最关键的问题：

谁在什么时间确认了哪组参数，系统是否只执行了一次，外部系统最终到底发生了什么。

系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。