投递

Webhook

无需任何轮询。在任意创建或后处理任务上传入 callbackUrl,资产完成的瞬间,PicoBerry 就会向您的服务器 POST 一个带签名的事件。

在任务中启用

任意生成请求中添加两个可选字段 — from-textfrom-imageimages重构网格纹理动画:

参数说明
callbackUrl可选
string
用于 POST 完成资产的公开 https:// URL。私有/内部 IP 将被拒绝。最多 2,048 个字符。
webhookSecret可选
string
HMAC 签名密钥。设置后,投递会带上可供验证的 X-PB-Signature 请求头。强烈推荐。
带回调创建
curl -X POST https://api.picoberry.ai/v1/models/from-text \
  -H "Authorization: Bearer pb_live_xxx" -H "Content-Type: application/json" \
  -d '{"prompt":"a stylized treasure chest","engine":"tripo",
       "callbackUrl":"https://example.com/webhooks/picoberry",
       "webhookSecret":"whsec_your_secret"}'
requests.post(f"{BASE}/v1/models/from-text", headers=headers, json={
    "prompt": "a stylized treasure chest", "engine": "tripo",
    "callbackUrl": "https://example.com/webhooks/picoberry",
    "webhookSecret": "whsec_your_secret"})
await fetch(`${BASE}/v1/models/from-text`, { method: "POST",
  headers: { ...headers, "Content-Type": "application/json" },
  body: JSON.stringify({ prompt: "a stylized treasure chest", engine: "tripo",
    callbackUrl: "https://example.com/webhooks/picoberry",
    webhookSecret: "whsec_your_secret" }) });

事件

当任务达到终止状态时,PicoBerry 会投递一个事件:

asset.succeeded 资产已完成 — files 已就绪asset.failed 任务失败 — 积分已退回

投递

请求体中的 data 与从 GET /v1/assets/{id} 获取的对象逐字节完全相同。

POST <your callbackUrl>
X-PB-Event: asset.succeeded
X-PB-Delivery-Id: 7f3a1b2c-…         # idempotency key — dedupe on this
X-PB-Signature: t=1785920000,v1=<hex>   # present when webhookSecret was set
Content-Type: application/json

{
  "event": "asset.succeeded",
  "deliveryId": "7f3a1b2c-…",
  "createdAt": "2026-08-05T09:12:00.000Z",
  "data": { /* identical to GET /v1/assets/{id} — id, taskStatus, files, … */ }
}

验证签名

v1 是以 webhookSecret 为密钥,对字符串 "<t>.<raw-request-body>" 计算出的 HMAC-SHA256。请基于原始请求体(JSON 解析之前)重新计算,以恒定时间进行比较,并拒绝过期的时间戳(> 5 分钟)以防止重放攻击。使用 X-PB-Delivery-Id 去重。

Node.js (express)
const crypto = require("crypto");

function verify(header, rawBody, secret) {
  const parts = Object.fromEntries(header.split(",").map(p => p.split("=")));
  const expected = crypto.createHmac("sha256", secret)
    .update(`${parts.t}.${rawBody}`).digest("hex");
  return crypto.timingSafeEqual(Buffer.from(parts.v1), Buffer.from(expected));
}
Python (flask)
import hmac, hashlib, time

def verify(header, raw_body, secret):
    parts = dict(p.split("=") for p in header.split(","))
    if abs(time.time() - int(parts["t"])) > 300:
        return False
    expected = hmac.new(secret.encode(), f"{parts['t']}.{raw_body}".encode(),
                        hashlib.sha256).hexdigest()
    return hmac.compare_digest(parts["v1"], expected)
快速响应 请在几秒内返回 2xx,并将繁重的工作放到异步处理。非 2xx 响应和超时会以指数退避重试,因此请使用 X-PB-Delivery-Id 让处理程序保持幂等。