> ## Documentation Index
> Fetch the complete documentation index at: https://docs.poyo.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Webhook 回调

> 任务完成时自动接收通知

# Webhook 回调

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

## 工作原理

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

## 回调请求

任务完成时，PoYo AI 会向你的回调 URL 发送 POST 请求。回调请求体与[状态接口](https://docs.poyo.ai/zh/api-manual/task-management/status.md)的任务数据结构相同，但不包含外层的 `code` 和 `data` 包装。

### 请求头

```http theme={"dark"}
Content-Type: application/json
X-Webhook-Timestamp: 1783680000
X-Webhook-Signature: base64_hmac_sha256_signature
```

### 请求体

```json theme={"dark"}
{
  "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
}
```

### 失败任务回调

```json theme={"dark"}
{
  "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 密钥：

```bash theme={"dark"}
GET /api/api-keys/webhook-secret
```

轮换你的 Webhook HMAC 密钥：

```bash theme={"dark"}
POST /api/api-keys/webhook-secret/rotate
```

签名规则：

```text theme={"dark"}
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 验证

```python theme={"dark"}
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 验证

```javascript theme={"dark"}
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 分钟
* **最终失败**：所有重试均失败后，不再继续重试

> [!WARNING]
>
> 如果所有重试均失败，你仍可以通过轮询[状态接口](https://docs.poyo.ai/zh/api-manual/task-management/status.md)获取结果。

## 实现示例

### Python (Flask)

```python theme={"dark"}
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)

```javascript theme={"dark"}
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

```go theme={"dark"}
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)
    }
}
```

## 安全最佳实践

> [!TIP]
>
> **验证请求**：处理回调前先校验 `X-Webhook-Timestamp` 和 `X-Webhook-Signature`。

> [!TIP]
>
> **幂等性**：将 Webhook 处理程序设计为幂等，以应对重复投递

> [!TIP]
>
> **异步处理**：异步处理回调并尽快返回 200，避免超时

> [!TIP]
>
> **日志记录**：记录所有 Webhook 请求，便于调试和监控

## 测试

### 使用 ngrok 进行本地测试

开发期间，你可以使用 [ngrok](https://ngrok.com) 暴露本地服务器：

```bash theme={"dark"}
# 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** | 生产系统 | 实时通知，无轮询开销 | 需要公网接口，调试较难 |
| **轮询** | 开发、简单集成 | 易于实现，无需服务器 | 延迟更高，消耗更多资源 |

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

## 后续步骤

> **[任务状态](https://docs.poyo.ai/zh/api-manual/task-management/status.md)**
>
> 了解如何轮询任务状态

> **[图像模型](https://docs.poyo.ai/zh/api-manual/image-series/overview.md)**
>
> 浏览所有图像模型

> **[视频模型](https://docs.poyo.ai/zh/api-manual/video-series/overview.md)**
>
> 浏览所有视频模型

> **[快速开始](https://docs.poyo.ai/zh/index.md)**
>
> 返回快速开始指南


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.