视频和图片都是异步任务:提交后拿到任务 ID,过一会儿才出结果。不想反复查询的话,可以让我们在任务结束时把结果推送到你的服务器,余额不足时也可以提醒你。
适合用在哪些场景、怎么三步接好:看《Webhook 用法》教程 →
两种接收方式
- 账户默认地址:在控制台「设置 → Webhook」填写推送地址、勾选事件。之后所有通过 API 提交的任务,结束时都会按勾选推送。 打开 Webhook 设置 →
- 单次请求带 callback_url:在提交视频或图片任务时加一个 callback_url 字段,这一单结束后(成功或失败)只推送到这个地址,不受账户勾选影响。
callback_url 必须是公网可访问的 http(s) 地址,最长 2000 个字符;也接受 callBackUrl 这种写法。两种方式用的是同一把签名密钥,在「设置 → Webhook」里查看。
curl https://nezhagate.com/v1/videos/generations -H 'Authorization: Bearer YOUR_API_KEY' -H 'Content-Type: application/json' -d '{"model": "seedance-2.5", "prompt": "a paper boat drifting down a rainy street", "duration": 5, "callback_url": "https://example.com/webhooks/nezhagate"}'事件
| type | 什么时候发 | data 里是什么 |
|---|---|---|
video.completed | 视频生成成功 | 和查询任务接口(GET /v1/videos/jobs/{id}、GET /v1/images/jobs/{id})返回的 JSON 完全一样 |
video.failed | 视频生成失败(已全额退款) | |
image.completed | 图片生成成功 | |
image.failed | 图片生成失败(已全额退款) | |
balance.low | 余额跌破你设置的积分数(每次跌破只发一次,充值回到线上后重新生效) | balance_credits(当前余额,积分)和 threshold_credits(你设置的提醒线) |
webhook.test | 在设置页点「发送测试事件」 | 一句说明文字,用来确认能收到、能验签 |
请求格式
我们向你的地址发一个 POST 请求,正文是 JSON,带这几个请求头:
webhook-id | 事件 ID。同一事件重试时不变,用它去重。 |
webhook-timestamp | 发送时间(Unix 秒)。每次发送都会更新。 |
webhook-signature | 签名:v1, 后面跟 Base64 编码的 HMAC-SHA256。 |
Content-Type | 固定为 application/json。 |
正文示例
视频生成成功:
{
"id": "evt_5f0c3d2e8a9b4c71d2e3f405",
"object": "event",
"type": "video.completed",
"created_at": 1791200000,
"data": {
"id": "img_3f9a...c2",
"object": "video.generation.job",
"status": "succeeded",
"created": 1791199400,
"model": "seedance-2.5",
"data": [{ "url": "https://img.nezhagate.com/i/9f86d081a8....mp4" }],
"usage": { "clips": 1, "model": "seedance-2.5" }
}
}生成失败时 data 里是 error(和查询接口一样),扣款已全额退回:
{
"id": "evt_8b1e47c09d3a2f6e5b7c1d90",
"object": "event",
"type": "image.failed",
"created_at": 1791200120,
"data": {
"id": "img_7c21...e4",
"object": "image.generation.job",
"status": "failed",
"created": 1791200050,
"model": "gpt-image-2",
"error": { "message": "...", "type": "upstream_error", "code": "upstream_error" }
}
}{
"id": "evt_1a2b3c4d5e6f708192a3b4c5",
"object": "event",
"type": "balance.low",
"created_at": 1791203000,
"data": { "balance_credits": 412.5, "threshold_credits": 500 }
}验证签名
我们按 Standard Webhooks 规范签名(OpenAI 的 Webhook 用的也是这个规范)。验证步骤:
- 取出请求头 webhook-id、webhook-timestamp、webhook-signature,以及原始请求正文(不要先解析成 JSON 再序列化)。
- 拼接待签内容:webhook-id + "." + webhook-timestamp + "." + 原始正文。
- 密钥是 whsec_ 后面那段,先做 Base64 解码,再用它计算 HMAC-SHA256,结果做 Base64 编码。
- webhook-signature 可能带多个签名(用空格分隔,各自以 v1, 开头),任意一个和你算的相同即通过;比较时用常量时间比较。
- 建议拒绝时间戳和当前时间相差超过 5 分钟的请求,防止重放。
import base64, hashlib, hmac, time
from flask import Flask, abort, request
SECRET = "whsec_..." # Settings -> Webhook -> Signing secret
KEY = base64.b64decode(SECRET[len("whsec_"):])
def verify(headers, body: bytes) -> bool:
msg_id = headers.get("webhook-id", "")
ts = headers.get("webhook-timestamp", "")
if not msg_id or not ts.isdigit() or abs(time.time() - int(ts)) > 300:
return False
signed = f"{msg_id}.{ts}.".encode() + body
expected = base64.b64encode(hmac.new(KEY, signed, hashlib.sha256).digest()).decode()
return any(sig.startswith("v1,") and hmac.compare_digest(sig[3:], expected)
for sig in headers.get("webhook-signature", "").split())
app = Flask(__name__)
@app.post("/webhooks/nezhagate")
def webhook():
if not verify(request.headers, request.get_data()):
abort(401)
event = request.get_json()
if event["type"] == "video.completed":
print(event["data"]["id"], event["data"]["data"][0]["url"])
return "", 204import crypto from "node:crypto";
import express from "express";
const SECRET = "whsec_..."; // Settings -> Webhook -> Signing secret
const KEY = Buffer.from(SECRET.slice("whsec_".length), "base64");
const app = express();
app.post("/webhooks/nezhagate", express.raw({ type: "application/json" }), (req, res) => {
const id = req.header("webhook-id"), ts = req.header("webhook-timestamp");
if (!id || !ts || Math.abs(Date.now() / 1000 - Number(ts)) > 300) return res.sendStatus(401);
const expected = Buffer.from(
crypto.createHmac("sha256", KEY).update(`${id}.${ts}.`).update(req.body).digest("base64"));
const ok = (req.header("webhook-signature") || "").split(" ").some((sig) => {
const got = Buffer.from(sig.slice(3));
return sig.startsWith("v1,") && got.length === expected.length && crypto.timingSafeEqual(got, expected);
});
if (!ok) return res.sendStatus(401);
const event = JSON.parse(req.body.toString("utf8"));
console.log(event.type, event.data.id);
res.sendStatus(204);
});
app.listen(3000);也可以直接用各语言的 Standard Webhooks 官方库(Python:pip install standardwebhooks;Node:npm install standardwebhooks)。
响应与重试
- 10 秒内返回任意 2xx 状态码即算送达,返回内容不重要。请先回 2xx,再慢慢处理。
- 没收到 2xx(超时、连接失败、3xx、4xx、5xx)会自动重试:1 分钟、5 分钟、30 分钟、2 小时、6 小时后各一次,之后标记为失败。
- 同一事件可能送达不止一次,也不保证先后顺序:用 webhook-id 去重,以 data.status 为准。
- 最近 30 天的推送记录在「设置 → Webhook」里,可以看到每次的结果,也可以手动重发。
安全
- 只推送到公网地址:内网、本机和保留地址一律拒绝。
- 不跟随跳转:对方返回 3xx 视为失败。
- 签名密钥请像 API Key 一样保密;泄露了就在设置页重置,旧密钥立即失效。
测试
在「设置 → Webhook」保存地址后,点「发送测试事件」,我们会立刻发一个 webhook.test 事件,并显示对方返回的状态码和耗时。