Webhook 回调配置

Webhook 概述

Webhook 允许裕普智汇在特定事件发生时主动通知您的服务器，无需频繁轮询 API。当任务完成、状态变更等事件发生时，系统会自动向您配置的 URL 发送 POST 请求。

支持的事件类型

配置步骤

签名验证

为确保安全性，每个请求都包含 X-Yupu-Signature 头，用于验证请求确实来自裕普智汇。

签名算法：HMAC-SHA256
签名格式：timestamp + "." + payload 的 HMAC-SHA256 签名

验证步骤：
1. 从 Header 中提取 X-Yupu-Timestamp 和 X-Yupu-Signature
2. 计算当前时间与 timestamp 的差值，确保在 5 分钟内
3. 构造签名内容：timestamp + "." + request_body
4. 使用密钥计算 HMAC-SHA256，与 signature 比对

import hmac
import hashlib

def verify_signature(payload: bytes, timestamp: str, signature: str, secret: str):
    # 检查时间戳
    import time
    if abs(time.time() - int(timestamp)) > 300:
        return False

    # 计算签名
    content = f"{timestamp}.{payload.decode()}"
    expected = hmac.new(
        secret.encode(),
        content.encode(),
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(expected, signature)

import crypto from 'crypto';

function verifySignature(payload, timestamp, signature, secret) {
    // 检查时间戳
    if (Math.abs(Date.now() / 1000 - parseInt(timestamp)) > 300) {
        return false;
    }

    // 计算签名
    const content = `${timestamp}.${payload}`;
    const expected = crypto
        .createHmac('sha256', secret)
        .update(content)
        .digest('hex');

    return crypto.timingSafeEqual(
        Buffer.from(signature),
        Buffer.from(expected)
    );
}

重试机制

如果您的服务器在收到请求后返回非 2xx 状态码，或在 10 秒内未响应，系统会认为投递失败。失败后会按照以下策略重试：5秒后、30秒后、2分钟后、10分钟后、1小时后。最多重试 5 次，仍失败则标记为投递失败。

完整示例

Python Flask 示例：

from flask import Flask, request, jsonify
import hmac
import hashlib
import time

app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret"

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    # 获取 Header
    timestamp = request.headers.get('X-Yupu-Timestamp')
    signature = request.headers.get('X-Yupu-Signature')
    event_type = request.headers.get('X-Yupu-Event')

    # 验证签名
    if not verify_signature(request.data, timestamp, signature):
        return jsonify({"error": "Invalid signature"}), 401

    payload = request.json

    # 处理事件
    if event_type == 'task.completed':
        print(f"任务完成: {payload['task_id']}")
    elif event_type == 'report.generated':
        print(f"报告生成: {payload['report_id']}")

    return '', 200

def verify_signature(payload, timestamp, signature, secret):
    if abs(time.time() - int(timestamp)) > 300:
        return False
    content = f"{timestamp}.{payload.decode()}"
    expected = hmac.new(secret.encode(), content.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

常见问题