Webhook
无需任何轮询。在任意创建或后处理任务上传入 callbackUrl,资产完成的瞬间,PicoBerry 就会向您的服务器 POST 一个带签名的事件。
在任务中启用
在任意生成请求中添加两个可选字段 — from-text、from-image、images、重构网格、纹理、动画:
| 参数 | 说明 |
|---|---|
| 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 让处理程序保持幂等。