AI Agent 技能分享｜Tool Calling 的超时、重试、幂等和权限控制

Tool Calling 的入门示例通常只有十几行：声明一个函数，挂到 Agent 上，看到模型成功调用就结束了。真正接业务接口时，麻烦往往从第一次超时开始。

假设 Agent 正在创建售后工单。请求发出去以后客户端超时了，我们并不知道工单到底有没有创建成功。这时直接重试，可能多出一张工单；不重试，用户又可能一直等不到结果。再加上越权、无效参数和下游限流，原来的十几行很快就不够用了。

下面就拿这个场景拆一下超时、重试、幂等和权限。代码使用 OpenAI Agents SDK 与httpx。

从那段最简单的 Tool 开始

最简单的 Tool 往往长这样：

@function_toolasyncdefcreate_ticket(order_id:str,reason:str)->str:returnawaitrequest_order_api(order_id,reason)

这段代码当然能跑，只是没有处理这些情况：

• 下游接口 20 秒没有响应，Agent 一直等待；
• 请求已经成功，下游响应却在网络中丢失；
• 运行框架、网关和业务代码分别重试，最终创建三张工单；
• 用户没有售后权限，却通过自然语言让 Agent 调用了接口；
• 模型填入不存在的订单号或超长原因；
• Tool 把内部异常、Token 或数据库地址原样返回给模型。

补齐以后，调用大致会变成下面这样：

模型生成参数 ↓ Schema 校验 ↓ 身份与业务权限校验 ↓ 生成稳定的幂等键 ↓ 带超时地请求下游 ↓ 仅对可重试错误退避重试 ↓ 审计并返回受控结果

这几个概念别混在一起

1. 超时

超时的作用是给一次调用设置时间边界。通常要区分：

• 连接超时：多久无法建立连接就放弃；
• 读取超时：连接成功后多久收不到响应就放弃；
• Tool 总超时：整个工具执行最多允许多长时间；
• Agent 总时限：包含模型推理和多个工具调用的总时限。

这些时间要放在同一个预算里算。Tool 总超时只有 10 秒，HTTP 读取超时却设成 30 秒，外层一取消，内层请求很可能还没来得及正常收尾。

2. 重试

适合重试的是那些过一会儿可能自行恢复的故障，比如：

• 连接被重置；
• HTTP 408、429；
• HTTP 500、502、503、504；
• 下游明确返回可以稍后重试的错误码。

参数错误、无权限、订单不存在，重试十次也不会变好。遇到这类错误应尽快返回，别用重试掩盖问题。

3. 幂等

幂等表示同一个业务请求执行多次，最终效果与执行一次相同。

第一次：创建工单 T20260813001 第二次：返回已有工单 T20260813001 第三次：仍返回 T20260813001

超时以后是否再发一次，是重试策略的问题；再发一次会不会重复扣款，则要靠幂等保证。这两件事经常被放在一起说，实际上谁也替代不了谁。创建、扣款、退款、发消息、修改状态，只要有副作用就得想清楚重复请求会怎样。

4. 权限控制

Prompt 里写一句“没有权限时不要调用”只能减少误操作。有人绕过 Agent 直接调 Tool，或者模型判断错了，后端照样要能拦住。

把售后 Tool 补完整

安装依赖

pipinstallopenai-agents httpx pydantic

身份放在运行上下文里

用户身份、租户和角色从登录态或访问令牌中解析，再放进运行上下文。不要把这些字段做成 Tool 参数，否则模型也能填写。

fromdataclassesimportdataclass@dataclassclassAppContext:request_id:struser_id:strtenant_id:strroles:set[str]access_token:str

模型能看到的参数只有订单号和售后原因，看不到也无法修改tenant_id、user_id和roles。

权限检查和退避重试

importasyncioimporthashlibimportloggingimportrandomfromtypingimportAnnotatedimporthttpxfromagentsimportRunContextWrapper,function_toolfrompydanticimportField logger=logging.getLogger("agent-tools")RETRYABLE_STATUS={408,429,500,502,503,504}classToolBusinessError(Exception):"""可以安全转换成用户提示的业务异常。"""defrequire_role(context:AppContext,role:str)->None:ifrolenotincontext.roles:raiseToolBusinessError("当前用户没有创建售后工单的权限")defmake_idempotency_key(context:AppContext,order_id:str)->str:# request_id 在一次用户请求的所有重试中必须保持不变。raw=(f"{context.tenant_id}:"f"{context.user_id}:"f"{context.request_id}:"f"create_after_sale_ticket:"f"{order_id}")returnhashlib.sha256(raw.encode("utf-8")).hexdigest()asyncdefpost_with_retry(url:str,*,json:dict,headers:dict,max_attempts:int=3,)->dict:timeout=httpx.Timeout(connect=2.0,read=5.0,write=3.0,pool=2.0)asyncwithhttpx.AsyncClient(timeout=timeout)asclient:forattemptinrange(1,max_attempts+1):try:response=awaitclient.post(url,json=json,headers=headers)ifresponse.status_codenotinRETRYABLE_STATUS:response.raise_for_status()returnresponse.json()ifattempt==max_attempts:response.raise_for_status()# 优先尊重下游 Retry-After，示例只处理秒数格式。retry_after=response.headers.get("Retry-After")ifretry_afterandretry_after.isdigit():delay=min(float(retry_after),5.0)else:delay=min(0.5*(2**(attempt-1)),4.0)delay+=random.uniform(0,0.2)awaitasyncio.sleep(delay)except(httpx.ConnectError,httpx.ReadTimeout)asexc:ifattempt==max_attempts:raiseexc delay=min(0.5*(2**(attempt-1)),4.0)delay+=random.uniform(0,0.2)awaitasyncio.sleep(delay)raiseRuntimeError("unreachable")

等待时间不是固定值，而是随着重试次数增加，并混入一点随机量。这样多个实例不会在同一时刻再次冲向刚恢复的下游服务。

Tool 本体

@function_tool(timeout=12.0,timeout_behavior="error_as_result",)asyncdefcreate_after_sale_ticket(ctx:RunContextWrapper[AppContext],order_id:Annotated[str,Field(min_length=6,max_length=32,pattern=r"^[A-Za-z0-9_-]+$"),],reason:Annotated[str,Field(min_length=5,max_length=500)],)->dict:"""为当前用户有权访问的订单创建售后工单。"""context=ctx.context require_role(context,"after_sale:create")idempotency_key=make_idempotency_key(context,order_id)headers={"Authorization":f"Bearer{context.access_token}","X-Tenant-Id":context.tenant_id,"X-Request-Id":context.request_id,"Idempotency-Key":idempotency_key,}try:result=awaitpost_with_retry("https://order-api.internal/api/after-sale/tickets",json={"orderId":order_id,"reason":reason},headers=headers,)logger.info("tool=create_after_sale_ticket request_id=%s user_id=%s ""tenant_id=%s order_id=%s ticket_id=%s",context.request_id,context.user_id,context.tenant_id,order_id,result.get("ticketId"),)return{"success":True,"ticket_id":result["ticketId"],"status":result["status"],}exceptToolBusinessError:raiseexcepthttpx.HTTPStatusErrorasexc:# 不把下游响应体直接暴露给模型，其中可能包含内部信息。logger.warning("tool failed request_id=%s status=%s",context.request_id,exc.response.status_code,)raiseToolBusinessError("售后服务暂时无法完成请求")exceptException:logger.exception("tool crashed request_id=%s",context.request_id,)raiseToolBusinessError("工具执行失败，请稍后重试")

OpenAI Agents SDK 的异步函数工具可以直接设置超时。error_as_result会把超时作为工具结果交回模型，Agent 还能组织一句正常的用户提示；希望一超时就终止整次运行时，再换成raise_exception。

请求头

Idempotency-Key只是双方约定的标识。下游接口如果没有保存和检查它，这个请求头就是摆设。

SQL Server 可以建立一张幂等记录表：

CREATETABLEdbo.ApiIdempotency(IdempotencyKeyvarchar(64)NOTNULL,OperationNamevarchar(100)NOTNULL,RequestHashchar(64)NOTNULL,Statusvarchar(20)NOTNULL,ResponseBody nvarchar(max)NULL,CreatedAt datetime2NOTNULLCONSTRAINTDF_ApiIdempotency_CreatedAtDEFAULTSYSUTCDATETIME(),ExpiresAt datetime2NOTNULL,CONSTRAINTPK_ApiIdempotencyPRIMARYKEY(IdempotencyKey));GO

接口收到请求后，可以按这个顺序处理：

收到请求 ↓ 计算请求体 RequestHash ↓ 不存在 Idempotency-Key → 建立 PROCESSING 记录并执行业务 已存在且 RequestHash 不同 → 返回 409，拒绝“一键多用” 已存在且状态 SUCCESS → 直接返回上次保存的结果 已存在且状态 PROCESSING → 返回 409/202，提示处理中

业务写入和幂等状态更新要放在可靠的事务边界里。并发请求可能同时发现“记录不存在”，所以还得靠唯一索引裁决，不能只写一个无锁的“先查再插入”。

幂等键

常用的做法有两种：

1. 调用方生成：同一次业务意图的所有重试复用同一个键；
2. 服务端根据稳定业务键生成：例如租户 + 订单 + 操作类型 + 退款批次。

最容易犯的错是每次重试都uuid4()：请求看起来有幂等键，实际上每次都不一样。只用order_id也太粗，同一订单以后再发起一次合法售后，可能被旧记录永久挡住。

调用可自动重试

不要只按 GET、POST 判断。HTTP 方法是线索，业务动作能否安全重放才是决定因素。

权限多检查

先减少工具可见范围

普通客服 Agent 根本不应该看到财务退款工具。减少工具数量也能降低模型选错工具的概率。

Tool 执行前再验角色

像示例中的require_role一样，执行前根据可信上下文检查角色或 Scope。不要使用模型传入的role。

下游还要验具体业务对象

有after_sale:create权限，不代表可以操作任何订单。订单服务仍然需要校验：

当前 tenant_id 是否拥有该订单？ 当前 user_id 是否能访问该组织/门店的订单？ 订单当前状态是否允许创建售后？ 金额是否超过该用户的授权额度？

这些判断只有订单服务掌握完整数据，放在 Agent 侧并不可靠。

几个很容易踩的坑

错误 1：所有异常都重试

401、403、参数校验失败、业务规则不满足都不是网络抖动。重试不会让它们变成功。

错误 2：多层无限叠加重试

如果 Agent SDK 重试 3 次、Tool 重试 3 次、网关再重试 3 次，最坏可能放大为 27 次请求。每一层都要明确重试责任，并设置总时间预算。

错误 3：超时后假设操作一定失败

超时只表示调用方没有及时拿到结果，不代表下游没有执行成功。对于写操作，超时后应使用同一个幂等键查询或重试。

错误 4：把授权规则全写进 Prompt

Prompt 可以帮助模型做正确选择，但不能抵抗越权调用、代码缺陷或恶意客户端。

错误 5：将完整异常返回给模型

堆栈、SQL、内部 URL、请求头和响应体都可能包含敏感信息。日志中保留排障信息，模型只接收稳定的业务错误码和简短说明。

多补的几组故障测试

正常用例跑通以后，可以直接人为制造故障：让下游延迟到超过读取超时，连续返回两次 503，或者在业务已经写入后断开连接。观察 Agent 最终调用了几次、总耗时有没有超出预算，以及重试是否始终复用同一个幂等键。

幂等接口至少再测两组并发请求：同一个 Key、相同请求体应该拿到同一份结果；同一个 Key、不同请求体必须返回冲突。权限测试则不要经过对话界面，直接使用无角色、错租户的上下文调用 Tool，确认后端确实会拒绝。

最后看日志。重试次数、每次等待时间、下游状态码和幂等命中情况都应该查得到，异常响应体、Authorization 请求头则不该出现在日志里。

最后

网络超时、重复请求和参数选错都不是罕见事故，而是正常运行时迟早会碰到的情况。把 Tool 当成普通后端接口来做就好：输入要校验，调用要有时间预算，写操作要幂等，权限要在服务端落地，日志也得能串起整次请求。

模型只是这条链路里一个新的调用方。后端原本该守的边界，并不会因为接入 Agent 而消失。

下一篇再往前走一步：高风险工具不立即执行，先把 Agent 暂停下来，等人工审批后由另一个进程接着跑。