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

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

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

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

从那段最简单的 Tool 开始

最简单的 Tool 往往长这样：

@function_tool
async def create_ticket(order_id: str, reason: str) -> str:
    return await request_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 补完整

安装依赖

pip install openai-agents httpx pydantic

身份放在运行上下文里

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

from dataclasses import dataclass


@dataclass
class AppContext:
    request_id: str
    user_id: str
    tenant_id: str
    roles: set[str]
    access_token: str

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

权限检查和退避重试

import asyncio
import hashlib
import logging
import random
from typing import Annotated

import httpx
from agents import RunContextWrapper, function_tool
from pydantic import Field


logger = logging.getLogger("agent-tools")
RETRYABLE_STATUS = {408, 429, 500, 502, 503, 504}


class ToolBusinessError(Exception):
    """可以安全转换成用户提示的业务异常。"""


def require_role(context: AppContext, role: str) -> None:
    if role not in context.roles:
        raise ToolBusinessError("当前用户没有创建售后工单的权限")


def make_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}"
    )
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


async def post_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)

    async with httpx.AsyncClient(timeout=timeout) as client:
        for attempt in range(1, max_attempts + 1):
            try:
                response = await client.post(url, json=json, headers=headers)

                if response.status_code not in RETRYABLE_STATUS:
                    response.raise_for_status()
                    return response.json()

                if attempt == max_attempts:
                    response.raise_for_status()

                # 优先尊重下游 Retry-After，示例只处理秒数格式。
                retry_after = response.headers.get("Retry-After")
                if retry_after and retry_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)

                await asyncio.sleep(delay)

            except (httpx.ConnectError, httpx.ReadTimeout) as exc:
                if attempt == max_attempts:
                    raise exc

                delay = min(0.5 * (2 ** (attempt - 1)), 4.0)
                delay += random.uniform(0, 0.2)
                await asyncio.sleep(delay)

    raise RuntimeError("unreachable")

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

Tool 本体

@function_tool(
    timeout=12.0,
    timeout_behavior="error_as_result",
)
async def create_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 = await post_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"],
        }

    except ToolBusinessError:
        raise
    except httpx.HTTPStatusError as exc:
        # 不把下游响应体直接暴露给模型，其中可能包含内部信息。
        logger.warning(
            "tool failed request_id=%s status=%s",
            context.request_id,
            exc.response.status_code,
        )
        raise ToolBusinessError("售后服务暂时无法完成请求")
    except Exception:
        logger.exception(
            "tool crashed request_id=%s",
            context.request_id,
        )
        raise ToolBusinessError("工具执行失败，请稍后重试")

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

请求头

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

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

CREATE TABLE dbo.ApiIdempotency
(
    IdempotencyKey varchar(64) NOT NULL,
    OperationName varchar(100) NOT NULL,
    RequestHash char(64) NOT NULL,
    Status varchar(20) NOT NULL,
    ResponseBody nvarchar(max) NULL,
    CreatedAt datetime2 NOT NULL
        CONSTRAINT DF_ApiIdempotency_CreatedAt DEFAULT SYSUTCDATETIME(),
    ExpiresAt datetime2 NOT NULL,
    CONSTRAINT PK_ApiIdempotency PRIMARY KEY (IdempotencyKey)
);
GO

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

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

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

幂等键

常用的做法有两种：

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

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

调用可自动重试

操作	示例	是否可自动重试	前提
纯读取	查询订单状态	通常可以	没有副作用
幂等写入	设置订单备注为指定内容	可以谨慎重试	服务端语义幂等
非幂等创建	创建工单	默认不可以	除非实现 Idempotency-Key
资金操作	退款、扣款	极其谨慎	强幂等、审计、人工审批
外部通知	发短信、邮件	谨慎	消息去重或业务唯一键

不要只按 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 暂停下来，等人工审批后由另一个进程接着跑。