Webhook 回调

除了轮询状态接口，你还可以在提交任务时提供回调 URL，在生成完成时自动接收通知。

工作原理

提交时附带回调 URL：在提交任务时加入 callback_url
接收通知：任务完成（成功或失败）时，PoYo AI 会向你的 URL 发送 POST 请求
处理结果：你的服务器会收到完整的任务数据，包括生成的文件

回调请求

任务完成时，PoYo AI 会向你的回调 URL 发送 POST 请求。回调请求体与状态接口的任务数据结构相同，但不包含外层的 code 和 data 包装。

请求头

Content-Type: application/json
X-Webhook-Timestamp: 1783680000
X-Webhook-Signature: base64_hmac_sha256_signature

请求体

{
  "task_id": "task-unified-1757165031-uyujaw3d",
  "status": "finished",
  "files": [
    {
      "file_url": "https://storage.poyo.ai/generated/image-abc123.jpg",
      "file_type": "image"
    }
  ],
  "created_time": "2025-11-12T10:30:00",
  "error_message": null
}

失败任务回调

{
  "task_id": "task-unified-1757165031-uyujaw3d",
  "status": "failed",
  "files": [],
  "created_time": "2025-11-12T10:30:00",
  "error_message": "The prompt violates our content policy"
}

签名验证

PoYo 使用 HMAC-SHA256 对每个 Webhook 请求签名。现有接收端可以忽略这些请求头，但生产系统应当验证签名。获取你的 Webhook HMAC 密钥：

GET /api/api-keys/webhook-secret

轮换你的 Webhook HMAC 密钥：

POST /api/api-keys/webhook-secret/rotate

签名规则：

base64(HMAC-SHA256(task_id + "." + timestamp, webhook_hmac_key))

其中：

task_id 是回调请求体中顶层的 task_id
timestamp 是 X-Webhook-Timestamp 的值
webhook_hmac_key 由 GET /api/api-keys/webhook-secret 返回

Python 验证

import base64
import hashlib
import hmac
import time


def verify_poyo_webhook(payload, headers, webhook_hmac_key):
    task_id = payload.get("task_id")
    timestamp = headers.get("X-Webhook-Timestamp")
    received_signature = headers.get("X-Webhook-Signature")

    if not task_id or not timestamp or not received_signature:
        return False

    if abs(int(time.time()) - int(timestamp)) > 300:
        return False

    expected_signature = base64.b64encode(
        hmac.new(
            webhook_hmac_key.encode("utf-8"),
            f"{task_id}.{timestamp}".encode("utf-8"),
            hashlib.sha256,
        ).digest()
    ).decode("utf-8")

    return hmac.compare_digest(expected_signature, received_signature)

Node.js 验证

const crypto = require("crypto");

function verifyPoyoWebhook(payload, headers, webhookHmacKey) {
  const taskId = payload.task_id;
  const timestamp = headers["x-webhook-timestamp"];
  const receivedSignature = headers["x-webhook-signature"];

  if (!taskId || !timestamp || !receivedSignature) return false;

  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - Number(timestamp)) > 300) return false;

  const expectedSignature = crypto
    .createHmac("sha256", webhookHmacKey)
    .update(`${taskId}.${timestamp}`)
    .digest("base64");

  if (expectedSignature.length !== receivedSignature.length) return false;

  return crypto.timingSafeEqual(
    Buffer.from(expectedSignature),
    Buffer.from(receivedSignature)
  );
}

要求

你的回调接口必须满足以下要求：

仅支持 HTTPS：必须使用 HTTPS 协议（不支持 HTTP）
URL 最大长度：2048 个字符
响应超时：必须在 30 秒内响应
成功响应：应返回 HTTP 200-299 状态码
禁止内网 IP：不能使用私有/内网 IP 地址（例如 192.168.x.x、10.x.x.x）
公网可访问：必须能从互联网公开访问

重试策略

如果你的回调接口未响应或返回错误：

重试次数：最多自动重试 5 次
重试间隔：指数退避，从约 60 秒开始，最长 10 分钟
最终失败：所有重试均失败后，不再继续重试

如果所有重试均失败，你仍可以通过轮询状态接口获取结果。

实现示例

Python (Flask)

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook/generation-complete', methods=['POST'])
def handle_generation_complete():
    # Parse the callback payload
    data = request.json
    task_id = data.get('task_id')
    status = data.get('status')

    if status == 'finished':
        # Task completed successfully
        files = data.get('files', [])
        for file in files:
            print(f"Generated file: {file['file_url']}")
            # Download and process the file
            # save_file(file['file_url'])

    elif status == 'failed':
        # Task failed
        error = data.get('error_message')
        print(f"Task {task_id} failed: {error}")
        # Handle the failure

    # Return 200 to acknowledge receipt
    return jsonify({"received": True}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=443, ssl_context='adhoc')

Node.js (Express)

const express = require('express');
const app = express();

app.use(express.json());

app.post('/webhook/generation-complete', (req, res) => {
  const { task_id, status, files, error_message } = req.body;

  if (status === 'finished') {
    // Task completed successfully
    console.log(`Task ${task_id} completed`);
    files.forEach(file => {
      console.log(`Generated file: ${file.file_url}`);
      // Download and process the file
    });
  } else if (status === 'failed') {
    // Task failed
    console.log(`Task ${task_id} failed: ${error_message}`);
    // Handle the failure
  }

  // Return 200 to acknowledge receipt
  res.status(200).json({ received: true });
});

const https = require('https');
const fs = require('fs');

const options = {
  key: fs.readFileSync('private-key.pem'),
  cert: fs.readFileSync('certificate.pem')
};

https.createServer(options, app).listen(443, () => {
  console.log('Webhook server listening on port 443');
});

Go

package main

import (
    "encoding/json"
    "fmt"
    "net/http"
)

type TaskData struct {
    TaskID       string      `json:"task_id"`
    Status       string      `json:"status"`
    Files        []MediaFile `json:"files"`
    CreatedTime  string      `json:"created_time"`
    Progress     int         `json:"progress"`
    ErrorMessage *string     `json:"error_message"`
}

type MediaFile struct {
    FileURL  string `json:"file_url"`
    FileType string `json:"file_type"`
}

func handleWebhook(w http.ResponseWriter, r *http.Request) {
    var taskData TaskData

    if err := json.NewDecoder(r.Body).Decode(&taskData); err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    if taskData.Status == "finished" {
        // Task completed successfully
        fmt.Printf("Task %s completed\n", taskData.TaskID)
        for _, file := range taskData.Files {
            fmt.Printf("Generated file: %s\n", file.FileURL)
            // Download and process the file
        }
    } else if taskData.Status == "failed" {
        // Task failed
        fmt.Printf("Task %s failed: %s\n", taskData.TaskID, *taskData.ErrorMessage)
        // Handle the failure
    }

    // Return 200 to acknowledge receipt
    w.WriteHeader(http.StatusOK)
    json.NewEncoder(w).Encode(map[string]bool{"received": true})
}

func main() {
    http.HandleFunc("/webhook/generation-complete", handleWebhook)

    // Use HTTPS
    err := http.ListenAndServeTLS(":443", "cert.pem", "key.pem", nil)
    if err != nil {
        panic(err)
    }
}

安全最佳实践

验证请求：处理回调前先校验 X-Webhook-Timestamp 和 X-Webhook-Signature。

幂等性：将 Webhook 处理程序设计为幂等，以应对重复投递

异步处理：异步处理回调并尽快返回 200，避免超时

日志记录：记录所有 Webhook 请求，便于调试和监控

测试

使用 ngrok 进行本地测试

开发期间，你可以使用 ngrok 暴露本地服务器：

# Start your local webhook server on port 3000
node webhook-server.js

# In another terminal, start ngrok
ngrok http 3000

# Use the HTTPS URL provided by ngrok as your callback_url
# Example: https://abc123.ngrok.io/webhook/generation-complete

Webhook 与轮询对比

方式	适用场景	优点	缺点
Webhook	生产系统	实时通知，无轮询开销	需要公网接口，调试较难
轮询	开发、简单集成	易于实现，无需服务器	延迟更高，消耗更多资源

对于处理大量请求的生产系统，建议使用 Webhook，以减少 API 调用并即时获得通知。

后续步骤

任务状态

了解如何轮询任务状态

图像模型

浏览所有图像模型

视频模型

浏览所有视频模型

快速开始

返回快速开始指南

概览

任务管理

文件管理

账户管理

安全与最佳实践

Webhook 回调

Webhook 回调

工作原理

回调请求

请求头

请求体

失败任务回调

签名验证

Python 验证

Node.js 验证

要求

重试策略

实现示例

Python (Flask)

Node.js (Express)

Go

安全最佳实践

测试

使用 ngrok 进行本地测试

Webhook 与轮询对比

后续步骤

任务状态

图像模型

视频模型

快速开始

​Webhook 回调

​工作原理

​回调请求

​请求头

​请求体

​失败任务回调

​签名验证

​Python 验证

​Node.js 验证

​要求

​重试策略

​实现示例

​Python (Flask)

​Node.js (Express)

​Go

​安全最佳实践

​测试

​使用 ngrok 进行本地测试

​Webhook 与轮询对比

​后续步骤

任务状态

图像模型

视频模型

快速开始

Webhook 回调

工作原理

回调请求

请求头

请求体

失败任务回调

签名验证

Python 验证

Node.js 验证

要求

重试策略

实现示例

Python (Flask)

Node.js (Express)

Go

安全最佳实践

测试

使用 ngrok 进行本地测试

Webhook 与轮询对比

后续步骤