给 Agent 挂工具，最容易被低估的不是提示词，是失败处理。人写的后端代码调一个支付接口，调用方是确定的、次数是确定的、语义是确定的。Agent 不是：它可能把同一个工具在一轮里调两次，可能在超时之后自己再试一遍，可能一次任务跑二十分钟中间重试了四次，而你在日志里只看到一条"天气查询失败"。

这篇讲的是一套能直接抄的写法，分三件事：超时怎么切、重试该给谁、幂等键放哪一层。

先说结论

三条规则，后面展开为什么：

一是超时至少分两层，连接超时和读取超时必须分开设，再在外面套一个总时限。只有一个"超时 30 秒"的参数，等于把"对方没响应"和"对方在慢慢吐数据"混成一件事，排查时完全看不出是哪一头出了问题。

二是只对幂等的方法自动重试。GET 可以随便重试，POST 要看接口语义。第三方接口里大量 POST 其实不幂等（下单、提交工单、发消息），重试一次就是重复一次业务副作用。

三是幂等键不要放在工具内部生成，要由编排层按"这次任务 + 这个工具 + 这组参数"算出来，一路传下去。工具自己每次调用都生成新 key，重试等于没做幂等。

超时切三层

第一层是连接超时（connect timeout），管 TCP 握手和 TLS 协商。这一层超时通常意味着网络不通、DNS 慢、对端挂了，一般设 3 到 5 秒就够，超过这个数大概率不是慢，是断。

第二层是读取超时（read timeout），指连接建立之后，两次收到数据之间的最大间隔。它管的是"对方接了但不回"，或者"对方回了一半卡住"。这一层要看接口性质：普通查询接口 10 到 30 秒，流式接口要按 token 间隔算，不能拿总时长当超时。

第三层是总时限（deadline），从发起调用开始算的硬上限，包含重试等待。这一层最容易被忘，但没有它，重试和退避会把单次工具调用拖到几分钟，Agent 那边早就超时放弃了，你这边还在傻等。

import asyncio
import httpx

CONNECT_TIMEOUT = 3.0
READ_TIMEOUT = 12.0
TOTAL_DEADLINE = 20.0

async def call_weather(city: str, client: httpx.AsyncClient) -> dict:
    timeout = httpx.Timeout(READ_TIMEOUT, connect=CONNECT_TIMEOUT)
    try:
        async with asyncio.timeout(TOTAL_DEADLINE):
            resp = await client.get("/v1/weather", params={"city": city}, timeout=timeout)
            resp.raise_for_status()
            return resp.json()
    except (asyncio.TimeoutError, httpx.TimeoutException) as exc:
        raise ToolTimeout(city=city, stage="total") from exc

IDEMPOTENT_METHODS = {"GET", "HEAD", "PUT", "DELETE", "OPTIONS"}

def should_retry(method: str, status: int | None, attempt: int) -> bool:
    if attempt >= MAX_ATTEMPTS:
        return False
    if status is None:                    # 网络层异常，连接都没成
        return True
    if status in (429, 502, 503, 504):    # 明确的"过一会儿再来"
        return True
    if 500 <= status < 600:
        return method.upper() in IDEMPOTENT_METHODS
    return False

import random

def backoff_delay(attempt: int, base: float = 0.8, cap: float = 12.0) -> float:
    raw = min(cap, base * (2 ** attempt))
    return raw * random.uniform(0.5, 1.0)

import hashlib
import json

def idempotency_key(task_id: str, tool: str, args: dict) -> str:
    canonical = json.dumps(args, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    digest = hashlib.sha256(f"{task_id}|{tool}|{canonical}".encode()).hexdigest()
    return f"{task_id[:8]}-{digest[:24]}"