开发指南

快速指南带你跑通第一个请求。这一页讲的是：当这段代码需要在生产里长期稳定运行时，有什么不一样。

两种调用形态

BeatAPI 只有两种请求生命周期，而几乎所有集成事故都源于把其中一种当成了另一种。

文本是同步的：POST /v1/chat/completions、/v1/responses、/v1/messages 或 Gemini 兼容端点直接返回结果。其余接口只是接收任务并返回一个编号。

异步任务的生命周期

1. 提交 —— POST /v1/images/tasks（或 videos、effects、video-analysis、music-video、ecommerce-video）。返回 201 表示已受理，不表示已完成。
2. 先存编号 —— 在启动任何后台流程之前，先把 data.id 与 data.request_id 落库。响应之后、写库之前崩溃一次，就会留下一个你已经付了钱却再也找不到的任务。
3. 等待 —— 轮询 GET /v1/tasks/{task_id}，或注册回调。
4. 取结果 —— 状态为 succeeded 时读 data.output.media；为 failed 时读 data.error_code 与退款字段。

轮询

1
import time
2
import random
3
import requests
4
5
TERMINAL = {"succeeded", "failed"}
6
7
def wait_for_task(api_key, task_id, timeout=900):
8
    url = f"https://api.beatapi.io/v1/tasks/{task_id}"
9
    headers = {"Authorization": f"Bearer {api_key}"}
10
    deadline = time.time() + timeout
11
    delay = 5.0
12
13
    while time.time() < deadline:
14
        response = requests.get(url, headers=headers, timeout=30)
15
16
        if response.status_code == 429:
17
            time.sleep(float(response.headers.get("Retry-After", 10)))
18
            continue
19
        response.raise_for_status()
20
21
        task = response.json()["data"]
22
        if task["status"] in TERMINAL:
23
            return task
24
25
        # 加抖动，避免一批 worker 步调一致地同时打过来
26
        time.sleep(delay + random.uniform(0, delay * 0.3))
27
        delay = min(delay * 1.5, 30.0)
28
29
    raise TimeoutError(f"任务 {task_id} 在 {timeout} 秒内未完成")

这个循环有三点是关键：

• 按终态停止，而不是按经过的时间猜。任务一旦终结就不会再变。
• 退避并加抖动。 任务接口对每把密钥的上限是每分钟 120 次请求；仅仅几个并发任务用固定 1 秒轮询就能自己把额度耗光。
• 把 429 当作「等一下」而不是「失败」，按 Retry-After 等待。

用回调代替轮询

用 POST /v1/webhooks 注册一个端点，BeatAPI 会把终态推送过去，从而完全省掉轮询循环，代价是你需要一个可被公网访问的 HTTPS 端点。

先按原始字节校验签名，再解析 JSON —— 重新序列化会改变字节，签名就对不上了。详见 Webhook 回调。

即使配了回调，也请保留一条低频的对账轮询。推送可能失败；一个 processing 状态远超该模型正常耗时的任务，值得直接去查一下。

错误处理

每次失败都会带一个稳定的 error.code 和一个 request_id。请把 request_id 记进日志 —— 它能在支持侧精确定位到那一次调用。

1
{
2
  "error": {
3
    "code": "rate_limit_exceeded",
4
    "message": "Too many requests.",
5
    "request_id": "req_abc123",
6
    "retry_after_seconds": 12
7
  }
8
}

完整的 error.code 清单见快速指南。

1
import requests
2
3
response = requests.post(
4
    "https://api.beatapi.io/v1/images/tasks",
5
    headers={"Authorization": f"Bearer {api_key}"},
6
    json={"model": "nano-banana", "prompt": "a cute panda"},
7
    timeout=30,
8
)
9
10
if not response.ok:
11
    error = response.json().get("error", {})
12
    # request_id 是事后唯一能定位这次调用的东西
13
    log.error("beatapi failed", code=error.get("code"), request_id=error.get("request_id"))
14
    if response.status_code == 402:
15
        alert_billing()
16
    elif response.status_code == 429:
17
        schedule_retry(after=error.get("retry_after_seconds", 10))

两种限流

两条限制各自独立，失败方式也不同。

• 请求频率 —— 每把密钥每分钟的请求数上限，返回 429 并带 Retry-After。额度随累计充值提升；当前值与下一档位在控制台和 GET /api/user/self 中都能看到。
• 并发 —— 同时处于处理中的任务数上限，返回 user_concurrency_exceeded。唯一的办法是等在跑的任务结束；立刻重试只会白白消耗频率额度。

GET /v1/usage 会同时给出 concurrency.limit 与 concurrency.active。

不会白花钱的重试

• 只重试 429 与 5xx。400、401、402、403、404、409 一律不重试。
• 指数退避 + 抖动，并设上限。
• 每次创建任务都带 Idempotency-Key，让重复提交不可能重复扣费。
• 提交失败但已经拿到编号时，先查询再决定是否重新提交。
• 限制总尝试次数。一个永远失败又永远重试的请求，是一场会计费的故障。

成本控制

• 给文本调用设输出上限。 输出是贵的那一半，max_tokens 是最简单的手段。
• 让模型与步骤匹配。 分类和路由用不着最强的档位。
• 盯住会变价的模型。 GPT-5.6 与 Grok 系列在上下文超过阈值后按更高档位计费；DeepSeek 系列与混元 hy3 在公布的高峰窗口内按更高倍率计费。长时间运行的智能体或定时批处理，越线时是没有任何提示的。 每次响应的用量记录会显示实际适用的档位。
• 一个组件一把密钥。 这样 GET /v1/usage 的 by_api_key 不需要你做任何埋点就能归集成本。
• 对账退款。 失败的任务会退款；如果你的账本只记了预扣而不读退款，数字迟早和账单对不上。

上生产前的检查清单

• 使用准确的 https://api.beatapi.io 源地址。
• 长期密钥只放在服务端。浏览器只拿实时会话的短期 client_secret，绝不拿 API 密钥。
• 启动后台流程前先存 data.id 与 request_id。
• 轮询加抖动，在终态或需要决策的状态停止。
• 按原始字节校验回调签名。
• 记录预扣、结算、退款、错误与最终产物地址。
• 给 402 配告警 —— 它会一次性中止账户上所有付费操作。