回调
在请求里设置 callback_url,每次状态变化我们主动推给你,不用轮询。
握手
一个 URL 第一次被使用时,我们先发一个 challenge:
{ "challenge": "9f2c…" }
三秒内把这个值原样回显:
{ "challenge": "9f2c…" }
握手成功前不会推送任何通知。这一步证明这个 URL 是你的、并且是活的 —— 它防止这套 API 被当成向别人发流量的放大器。
通知内容
每次推送的 body 与查询接口返回的一致:
{
"task": {
"id": "424010985738629",
"status": "succeeded",
"content": { "url": "https://cdn.h3.studio/…" },
"usage": { "total_seconds": 5, "output_seconds": 5, "input_seconds": 0, "input_image_count": 0 }
}
}
校验签名
每次 POST 带两个头:
| 头 | 含义 |
|---|---|
X-H3-Timestamp | 签名时的 Unix 秒 |
X-H3-Signature | sha256= 加上对 "{timestamp}." + body 的 HMAC-SHA256 |
import hmac, hashlib
def verify(secret: str, timestamp: str, body: bytes, signature: str) -> bool:
expected = "sha256=" + hmac.new(
secret.encode(), f"{timestamp}.".encode() + body, hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, signature)
用常数时间函数比较,不要用 ==。时间戳超过几分钟的请求直接拒绝,
这是防止旧请求被重放的关键。
重试与重复
非 2xx 或超时会按指数退避重试:5 秒、10 秒、20 秒、40 秒、80 秒、160 秒,然后放弃。
投递按 (任务, 状态) 去重,所以一个任务到达 succeeded 只会通知一次。
即便如此,请把处理函数写成幂等的:
你的服务器已经落库、但我们还没收到 200 就断网了 —— 这种情况从我们这边看,
和落库之前就失败完全无法区分。
只要事件已经可靠记下就立刻返回 2xx,慢活放到之后做 —— 在响应前先转码的处理函数会超时,然后收到同一个事件。