Webhook

Webhook 会在你的账户中发生某件事时，由 Sakuranbo AI 向你的服务器发送一个 HTTP 请求。使用 Webhook，你可以无需轮询就得知图像或视频任务已经完成，在充值到账时记录充值，并在余额不足时收到警告。

推送遵循 Standard Webhooks 规范，并且每一次推送都带有签名。你的服务器必须先验证签名，才能信任请求。详见验证签名。

端点可以订阅以下事件：job.succeeded, job.failed, job.cancelled, deposit.credited, balance.low。webhook.test 也会被发送，但仅在你请求测试事件时发送，你无法订阅它。

• job.succeeded：图像或视频任务变为 completed。data 是任务，与 GET /v1/jobs/{id} 返回的相同，outputs 中带有签名 URL。
• job.failed：任务变为 failed 或 expired。data 是任务，data.status 说明是这两者中的哪一个。对于 failed 的任务，data.error 说明原因。对于 expired 的任务，data.error 为 null。模型提供方因内容拒绝的任务也属于这种情况，此时 data.error.code 为 content_policy_violation。
• job.cancelled：任务变为 cancelled。data 是任务。
• deposit.credited：一笔充值已计入你的余额。data 包含金额、日期和你的余额。
• balance.low：你的可用余额低于你在控制台设置中设置的提醒额度。data 包含可用余额和提醒额度。
• webhook.test：你请求的测试事件。

任务状态详见任务，余额详见余额与 API 密钥限额。

每次推送的 JSON 请求体包含四个字段：

• id：事件的 ID。它同时也是 webhook-id 请求头的值。
• type：事件类型。
• timestamp：事件发生的时间，为 UTC 的 ISO 8601 时间。
• data：事件的内容，取决于类型。

请忽略你不认识的事件类型和字段。Sakuranbo AI 可能会在不另行通知的情况下新增事件和字段。

请在控制台的设置中设置提醒额度。每当可用余额从提醒额度或更高降到其以下时，就会发送一次 balance.low 事件。只有在余额回到提醒额度或更高之后，才会再发送下一次事件。扣费、充值余额到期或向下调整都可能触发它，单纯的预留则不会。未设置提醒额度时不会发送任何事件。

端点是接收事件的 URL。你可以在控制台中创建，也可以通过 API 创建。

在控制台中，打开 Webhook 页面。你可以创建端点、编辑端点、停止和恢复端点、发送测试事件、轮换签名密钥以及查看推送记录。

通过 API 创建时，调用 POST /v1/webhook-endpoints。

curl -X POST "https://api.sakuranbo.ai/v1/webhook-endpoints" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $SAKURANBO_API_KEY" \
  -d '{
  "url": "https://example.com/webhooks/cherry",
  "events": [
    "job.succeeded",
    "job.failed",
    "job.cancelled"
  ]
}'

const response = await fetch("https://api.sakuranbo.ai/v1/webhook-endpoints", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: `Bearer ${process.env.SAKURANBO_API_KEY}`,
  },
  body: JSON.stringify({
    "url": "https://example.com/webhooks/cherry",
    "events": [
      "job.succeeded",
      "job.failed",
      "job.cancelled"
    ]
  }),
});
const data = await response.json();
console.log(data);

import os

import requests

response = requests.post(
    "https://api.sakuranbo.ai/v1/webhook-endpoints",
    headers={
        "Authorization": f"Bearer {os.environ['SAKURANBO_API_KEY']}",
    },
    json={
        "url": "https://example.com/webhooks/cherry",
        "events": [
            "job.succeeded",
            "job.failed",
            "job.cancelled",
        ],
    },
)
print(response.json())

请求体包含以下字段：

如果省略 events，端点会订阅你创建它时存在的所有事件类型。之后新增的事件类型不会自动加入该端点。要接收它，请调用 PATCH /v1/webhook-endpoints/{id}，并在 events 中给出完整的事件类型列表，包括新增的那一个。events 会替换整个列表。

响应包含端点的 ID 和签名 secret。签名密钥以 whsec_ 开头。完整的签名密钥只会出现在这个请求的响应和轮换签名密钥的响应中。之后你无法再次读取，因此请立即保存。如果丢失，请轮换签名密钥。

URL 必须遵守以下规则：

• 使用 https 和默认端口。
• 主机是公开的 DNS 名称。IP 地址、localhost、不含点的名称和内部名称都会被拒绝。带有用户名或密码的 URL 也会被拒绝。
• 可以带有路径和查询字符串。

违反规则的 URL 会返回 400 和 invalid_request。error.param 为 url，error.details.reason 说明违反了哪条规则。如需在自己的机器上测试，请通过公开的 HTTPS 隧道对外暴露。

一个账户最多可以有 10 个端点。请先删除一个，再添加新的。

Sakuranbo AI 把每个事件作为 POST 请求发送，请求体为该 JSON，并带有以下请求头：

• webhook-id：事件的 ID。
• webhook-timestamp：本次尝试的时间，单位为 Unix 秒。
• webhook-signature：签名。详见验证签名。

你的服务器按以下规则应答：

• 在 15 秒内返回 2xx 状态，表示事件已收到。请先应答，再做耗时的处理。
• Sakuranbo AI 不会跟随重定向。3xx 状态视为失败。
• 410 Gone 状态会使端点停止。端点停止期间发生的事件不会排队，恢复端点也不会重新发送它们。要恢复，请调用 PATCH /v1/webhook-endpoints/{id} 并把 enabled 设为 true，或使用控制台。然后手动重试失败的推送。
• 其他任何状态、超时或连接失败都视为失败。Sakuranbo AI 会再次尝试，在大约 31 小时内总共最多尝试 8 次。最后一次尝试之后，该推送变为 failed。

编写接收端时请记住以下事实：

• 同一个事件可能多次到达。同一个事件的每一次尝试都有相同的 webhook-id 和相同的请求体，因此请保存已处理的 ID，并跳过重复的事件。
• 事件的顺序无法保证。请使用 timestamp 和 data 中的状态。
• 对于你不处理的事件类型，也请返回 2xx，这样它们就不会被重试。

每次推送都用端点的签名密钥签名。请验证签名，以确认请求是由 Sakuranbo AI 发送的，并且没有人改动过。所用的方案是 Standard Webhooks 版本 v1，使用 HMAC-SHA256。

1. 读取 webhook-id、webhook-timestamp 和 webhook-signature 请求头。如果缺少任何一个，就拒绝该请求。
2. 检查时间戳与当前时间相差不超过 5 分钟，无论是在过去还是在未来。这样可以阻止重放的请求。
3. 生成密钥。从签名密钥中去掉 whsec_ 前缀，并把其余部分按 base64 解码。密钥是解码后的字节，而不是签名密钥的文本。
4. 生成待签名内容：ID 请求头的值、一个点、时间戳请求头的值、一个点，以及原始请求体。
5. 用该密钥计算待签名内容的 HMAC-SHA256，并把结果编码为标准 base64。
6. 签名请求头包含一个或多个以空格分隔的条目。每个条目由版本、一个逗号和签名组成。跳过版本不是 v1 的条目。只要有任何一个条目与你的结果匹配，该请求就有效。请以恒定时间比较。
7. 只有在签名有效之后，才解析 JSON。

请对你收到的原始字节进行验证。Web 框架如果先解析 JSON 再重新序列化，可能会改变字节，导致签名不匹配。请在任何解析器运行之前读取原始请求体。

TypeScript：

import { createHmac, timingSafeEqual } from "node:crypto";

/**
 * Verifies a Sakuranbo AI webhook (Standard Webhooks) and returns the parsed event.
 * rawBody must be the request body exactly as received, before any JSON parsing.
 */
export function verifyWebhook(
  secret: string, // "whsec_..." as shown when the endpoint was created or its secret rotated
  rawBody: string,
  headers: Record<string, string | undefined>, // header names in lower case
  toleranceSeconds = 300,
) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = headers["webhook-signature"];
  if (!id || !timestamp || !signatures) throw new Error("missing webhook headers");

  const sentAt = Number(timestamp);
  const now = Math.floor(Date.now() / 1000);
  if (!Number.isInteger(sentAt) || Math.abs(now - sentAt) > toleranceSeconds) {
    throw new Error("timestamp outside the tolerance");
  }

  const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
  const expected = createHmac("sha256", key).update(`${id}.${timestamp}.${rawBody}`).digest();

  // During a secret rotation the header holds two signatures, separated by a space.
  for (const entry of signatures.split(" ")) {
    const [version, signature] = entry.split(",");
    if (version !== "v1" || !signature) continue;
    const received = Buffer.from(signature, "base64");
    if (received.length === expected.length && timingSafeEqual(received, expected)) {
      return JSON.parse(rawBody);
    }
  }
  throw new Error("no matching signature");
}

Python：

import base64
import hashlib
import hmac
import json
import time


def verify_webhook(secret: str, raw_body: bytes, headers: dict, tolerance_seconds: int = 300):
    """Verify a Sakuranbo AI webhook (Standard Webhooks) and return the parsed event.

    raw_body must be the request body exactly as received, before any JSON parsing.
    """
    headers = {name.lower(): value for name, value in headers.items()}
    message_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not (message_id and timestamp and signatures):
        raise ValueError("missing webhook headers")

    try:
        sent_at = int(timestamp)
    except ValueError:
        raise ValueError("invalid timestamp") from None
    if abs(int(time.time()) - sent_at) > tolerance_seconds:
        raise ValueError("timestamp outside the tolerance")

    if secret.startswith("whsec_"):
        secret = secret[len("whsec_"):]
    key = base64.b64decode(secret)
    signed = f"{message_id}.{timestamp}.".encode() + raw_body
    expected = hmac.new(key, signed, hashlib.sha256).digest()

    # During a secret rotation the header holds two signatures, separated by a space.
    for entry in signatures.split(" "):
        version, _, signature = entry.partition(",")
        if version != "v1" or not signature:
            continue
        if hmac.compare_digest(base64.b64decode(signature), expected):
            return json.loads(raw_body)
    raise ValueError("no matching signature")

Standard Webhooks 的库会做同样的检查。如果你更愿意使用库而不是自己写代码，请使用 npm 或 PyPI 上的 standardwebhooks 包。

如果你丢失了签名密钥，或签名密钥可能已经泄露，请轮换它。使用控制台或 POST /v1/webhook-endpoints/{id}/rotate-secret。响应包含新的签名密钥，这是你唯一一次可以读取它的机会。

轮换之后的 24 小时内，签名请求头包含两个签名，新的在前，之前的在后。在此期间，持有任意一个签名密钥的接收端都能验证请求。请把你的服务器切换到新的签名密钥。这段时间结束后，旧的签名密钥就会失效。端点上的 previous_secret_expires_at 显示何时失效。

如果你在这段时间内再次轮换，第一次轮换之前的签名密钥会立即失效。只有最新的两个签名密钥会为推送签名。

POST /v1/webhook-endpoints/{id}/test 向一个端点发送 webhook.test 事件，无论该端点订阅了哪些事件。端点必须处于启用状态。响应是新的推送，状态为 pending。控制台里有一个执行同样操作的按钮。

curl -X POST "https://api.sakuranbo.ai/v1/webhook-endpoints/we_0123456789abcdef0123456789abcdef/test" \
  -H "Authorization: Bearer $SAKURANBO_API_KEY"

const response = await fetch("https://api.sakuranbo.ai/v1/webhook-endpoints/we_0123456789abcdef0123456789abcdef/test", {
  method: "POST",
  headers: {
    Authorization: `Bearer ${process.env.SAKURANBO_API_KEY}`,
  },
});
const data = await response.json();
console.log(data);

import os

import requests

response = requests.post(
    "https://api.sakuranbo.ai/v1/webhook-endpoints/we_0123456789abcdef0123456789abcdef/test",
    headers={
        "Authorization": f"Bearer {os.environ['SAKURANBO_API_KEY']}",
    },
)
print(response.json())

通过 GET /v1/webhook-endpoints/{id}/deliveries 和控制台可以查看端点的每一次推送，最新的在前。每一条包含状态、尝试次数、最近一次的 HTTP 状态码、下一次尝试的时间，以及你的服务器响应的开头部分。推送记录不包含事件的请求体。Sakuranbo AI 将推送记录保留 30 天。

如需再次发送某次推送，请调用 POST /v1/webhook-endpoints/{id}/deliveries/{deliveryId}/retry 或使用控制台。该推送会以相同的事件 ID 和相同的请求体发出，并开始新一轮尝试。你只能重试已经结束的推送，即 succeeded 或 failed，并且端点必须处于启用状态。

所有这些端点都需要 API 密钥。详见认证。

• POST /v1/webhook-endpoints 用于创建端点。
• GET /v1/webhook-endpoints 用于列出你的端点。
• GET /v1/webhook-endpoints/{id} 用于读取一个端点。
• PATCH /v1/webhook-endpoints/{id} 用于修改 URL、事件、说明，或端点是否启用。
• DELETE /v1/webhook-endpoints/{id} 用于删除端点及其推送记录。
• POST /v1/webhook-endpoints/{id}/test 用于发送测试事件。
• POST /v1/webhook-endpoints/{id}/rotate-secret 用于创建新的签名密钥。
• GET /v1/webhook-endpoints/{id}/deliveries 用于读取推送记录。
• POST /v1/webhook-endpoints/{id}/deliveries/{deliveryId}/retry 用于再次发送一次推送。

其他端点在你的账户中不存在该端点或推送时，会返回 404 和 not_found。对已停止的端点发送测试事件或重试，会返回 400 和 invalid_request。完整的错误代码列表请参阅错误。