任务完成回调（Webhook）

  • 任务完成或失败时，由 OAIRoute 主动推送结果
  • 免去轮询，长耗时任务尤其划算
  • 推送失败会按退避策略重试

认证

string必填

所有接口均使用 Bearer Token 进行认证。

创建任务时同样需要在请求头里带上 Key。

Seedance 系列的接口需要在 ark 站 单独获取 Key，与 api 站的 Key 不通用。

请求时在请求头中添加：

Authorization: Bearer YOUR_ARK_KEY

配置方式

在创建任务时传入 callback_url，任务进入终态后就会向这个地址发起 POST 请求。

{
  "model": "seedance-2.5",
  "prompt": "海面日落，镜头缓慢前推",
  "callback_url": "https://your-server.com/hooks/oairoute"
}

callback_url 必须是公网可达的 HTTP(S) 地址。本地开发可以先用内网穿透工具暴露端口，或改用轮询。

回调请求参数

string

任务 ID，与轮询接口里的 id 一致。

string

任务状态。completed 成功完成，failed 失败，cancelled 用户取消。

number

本次任务扣费金额。

number

本次任务扣费积分。

object

任务结果。仅 completed 时出现。

result 的元素字段 · 2 项
array

图像任务结果数组，元素含 url 与 expires_at。

array

视频任务结果数组。

object

错误详情。仅 failed 时出现。

integer

任务创建时间戳。

integer

任务完成时间戳。

响应约定

接收端返回 2xx 即视为推送成功。返回其他状态码，或 10 秒内没有响应，会触发重试。

接收端需要具备幂等能力：同一个 task_id 可能被推送多次。建议按 task_id 做去重，处理过就直接返回 200。

重试策略

次数间隔说明
第 1 次立即初始推送
第 2 次30 秒后上次失败
第 3 次2 分钟后上次失败
第 4 次10 分钟后上次失败
第 5 次30 分钟后最后一次，仍失败则放弃

重试期间任务状态已经落库，随时可以用任务状态接口补查。

回调示例

{
  "id": "task_01KA040M0HP1GJWBJYZMKX1XS1",
  "status": "completed",
  "cost": 0.48,
  "credits_cost": 4.8,
  "result": {
    "videos": [
      {
        "url": "https://cdn.oairoute.com/v/task_01KA040M0HP1GJWBJYZMKX1XS1.mp4",
        "expires_at": 1763174708
      }
    ]
  },
  "created": 1763088289,
  "completed": 1763088332
}

一个最小接收端

from flask import Flask, request

app = Flask(__name__)
seen = set()

@app.post("/hooks/oairoute")
def hook():
    payload = request.get_json(force=True)
    task_id = payload.get("id")

    if task_id in seen:          # 幂等去重
        return {"ok": True}, 200
    seen.add(task_id)

    if payload.get("status") == "completed":
        for item in payload["result"].get("videos", []):
            download(item["url"])  # 立刻落盘，别依赖 CDN 长期保存
    else:
        log_error(payload.get("error"))

    return {"ok": True}, 200

使用示例

提交任务时带上回调地址

在生成请求里传 callback_url，任务状态变化时平台会 POST 到这个地址。比轮询省请求，也更及时。

处理完成回调

回调体里的 status 决定走哪个分支。产物 URL 会过期，收到就下载并落盘，不要只存链接。

@app.post("/hooks/oairoute")
def hook():
    payload = request.get_json(force=True)

    if payload.get("status") != "completed":
        log_error(payload.get("error"))
        return {"ok": True}, 200

    for item in payload["result"].get("videos", []):
        save_locally(item["url"])

    return {"ok": True}, 200

用任务 ID 做幂等

网络抖动会导致同一条回调被投递多次。用 id 去重，避免重复下载或重复入库。

import redis
seen = redis.Redis()

@app.post("/hooks/oairoute")
def hook():
    payload = request.get_json(force=True)
    task_id = payload.get("id")

    # 只在第一次见到时写入，返回 True 说明是新的
    if not seen.set("hook:" + task_id, "1", nx=True, ex=86400):
        return {"ok": True}, 200

    handle(payload)
    return {"ok": True}, 200

本地联调

开发阶段用 ngrok 之类把本地端口暴露出去，直接把公网地址填进 callback_url，不用先部署。

# 终端一：暴露本地 8000 端口
ngrok http 8000

# 终端二：把打印出的公网地址填进回调
curl --request POST https://oairoute-docs.ngrok-free.app/hooks/oairoute