OpenMontage 中的 BFL API Webhook 集成：从轮询走向事件驱动的 FLUX 图像生成回调

OpenMontage 把 AI 编码助手变成一个完整视频制作工作室，其中图像生成环节高度依赖 BFL（Black Forest Labs）的 FLUX API。本文以仓库中的 webhook-integration.md 参考文档为主体，完整讲清生产环境如何用 Webhook 替代轮询来接收生成结果：如何在请求中挂载 webhook_url / webhook_secret、如何解析成功与失败回调负载、如何用 HMAC-SHA256 校验签名，以及幂等、重试兜底与健康度监控等实战要点。读完后，你既能复制出一套可直接运行的 Flask / Express 回调处理器，也能对照仓库自带的 Python 客户端源码理解每个参数在底层如何被拼装与校验。

为什么生产环境要用 Webhook 替代轮询

BFL FLUX API 是异步生成模型：所有请求先返回一个 polling_url，客户端需要反复轮询它直到状态变为 Ready。这一模式在脚本、CLI、本地开发和单次请求场景下简单且通用，但在高并发、服务器到服务器（server-to-server）的生产负载下，持续轮询会带来重复的 API 调用与无谓的计算浪费。

Webhook 正是为生产负载设计的替代方案。原文明确列出它相对轮询的四点优势：

• 减少 API 调用——不再有反复的轮询请求；
• 即时通知——生成完成的瞬间即可得知，而不是轮询间隔内才发现；
• 更好的资源效率——不把算力浪费在轮询上；
• 可扩展的架构——事件驱动设计。

这一点与技能主文档的取舍建议一致：在 SKILL.md 的 “Polling vs Webhooks” 小节中，官方给出的判断标准是——

方式	适用场景
轮询（Polling）	脚本、CLI 工具、本地开发、单次请求、简单集成
Webhook	生产应用、高并发、服务器到服务器、需要即时通知

并且给出了一条明确的演进路径：“从轮询开始——它更简单且到处都能跑；当你需要规模化或想要事件驱动架构时，再切换到 Webhook。” 这也正是本文后续 “混合模式” 一章要落地的做法。

提交带 Webhook 的生成请求

启用 Webhook 的关键在于：在生成请求体里额外带上 webhook_url，可选地再带上 webhook_secret。这两个字段在端点文档中被列为所有请求的通用参数——见 endpoints.md 的 “Common Request Parameters” 表，其中 webhook_url（“URL for async notification”）与 webhook_secret（“Secret for webhook signature”）均为非必填（No）。

以 FLUX.2 [pro] 端点为例，带 Webhook 的请求如下：

curl -X POST "https://api.bfl.ai/v1/flux-2-pro" \
  -H "x-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A beautiful sunset over mountains",
    "webhook_url": "https://your-server.com/api/bfl-webhook",
    "webhook_secret": "your-secret-key-here"
  }'

请求头 x-key 携带 API 密钥，是所有 BFL 请求的鉴权方式；BFL_API_KEY 环境变量需在调用前设置（可用 echo $BFL_API_KEY 快速检查）。

从仓库自带的生产级 Python 客户端 python-client.py 可以看到这两个参数在底层是如何被拼装的：BFLClient.generate() 的签名就包含 webhook_url: str = None 与 webhook_secret: str = None，方法体内只有当二者非空时才写入请求体：

if webhook_url:
    payload["webhook_url"] = webhook_url
if webhook_secret:
    payload["webhook_secret"] = webhook_secret

这说明 webhook_secret 是可选但强烈建议提供的——提供它，BFL 才会对回调负载做签名，你才能在回调端校验来源（见下一节）。同一客户端还支持 region 参数选择地域端点（global / eu / us），对应端点表见 SKILL.md 的 “Base Endpoints”：https://api.bfl.ai（默认，自动故障切换）、https://api.eu.bfl.ai（GDPR）、https://api.us.bfl.ai（美国数据驻留）。

解析 Webhook 回调负载

生成完成后，BFL 会向你的 webhook_url 发送一个 POST 请求。回调负载有两种形态：成功与失败。

成功（status = "Ready"）：

{
  "id": "gen_abc123xyz",
  "status": "Ready",
  "result": {
    "sample": "https://bfldeliveryprod.blob.core.windows.net/results/...",
    "prompt": "...",
    "seed": 1234567890
  },
  "timestamp": "2025-01-15T10:30:00Z"
}

失败（status = "Error"）：

{
  "id": "gen_abc123xyz",
  "status": "Error",
  "error": "content_policy_violation",
  "message": "The prompt violated content policy",
  "timestamp": "2025-01-15T10:30:00Z"
}

解析时有两个仓库层面反复强调的关键点：

1. 结果 URL 十分钟过期。 SKILL.md 用加粗标出 “Important: Image URLs Expire in 10 Minutes”，并强调“生成完成后立即下载图片——不要保存或缓存 URL 本身”。这一点在 polling-patterns.md 的 “URL Expiration” 一节也以 “Critical:” 再次强调。因此回调处理器里必须立即下载 result.sample 指向的图片，而非仅记录 URL。
2. 错误码要与恢复策略对应。 失败负载中的 error 字段取值，可以在 error-handling.md 的 “Common Errors and Solutions / Generation Failures” 中找到对照：content_policy_violation（安全策略拦截）、generation_timeout（生成超时）、internal_error（服务端问题）、invalid_image（输入图无法处理）。该文档还给出 classify_error 的分流逻辑——content_policy_violation 与 invalid_image 属于不可重试错误，而 generation_timeout、internal_error 属于可重试错误，这对回调端“是否重试/如何降级”的决策至关重要。

安全：用 HMAC-SHA256 校验签名

只要提供了 webhook_secret，BFL 就会用 HMAC-SHA256 对负载签名，并放在请求头里：

X-BFL-Signature: sha256=<hex-encoded-signature>

签名格式固定为 sha256= 前缀加十六进制摘要。校验的核心是：用你的 webhook_secret 对原始请求体重新计算 HMAC-SHA256，再与请求头提供的摘要做恒定时间比较（避免时序侧信道）。

Python 校验函数（来自原文档）：

import hmac
import hashlib

def verify_webhook_signature(payload, signature, secret):
    """Verify the webhook came from BFL."""
    if not signature or not signature.startswith('sha256='):
        return False

    expected_signature = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()

    provided_signature = signature[7:]  # Remove 'sha256=' prefix

    return hmac.compare_digest(expected_signature, provided_signature)

这里有两个容易踩坑的细节，仓库源码里也做了同样处理：

• signature[7:] 去掉 sha256= 前缀（前缀恰为 7 个字符）；
• 用 hmac.compare_digest 做比较，而非 ==。

对照 python-client.py 中的独立函数 verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool，逻辑完全一致，只是它把 payload 显式标注为 bytes 类型——这一点值得注意：签名必须对原始请求体字节计算，因此在框架里要拿到未经解析的原始 body（Python 端是 request.data，Node 端要用 express.raw），而不能先 JSON.parse 再重序列化，否则会因空白/键序差异导致摘要不匹配。

带校验的 Flask 处理器（原文档完整实现）：

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

app = Flask(__name__)
WEBHOOK_SECRET = "your-secret-key-here"

@app.route('/api/bfl-webhook', methods=['POST'])
def handle_webhook():
    # Verify signature
    signature = request.headers.get('X-BFL-Signature')
    if not verify_webhook_signature(request.data, signature, WEBHOOK_SECRET):
        return jsonify({'error': 'Invalid signature'}), 401

    data = request.json

    if data['status'] == 'Ready':
        handle_completion(data)
    elif data['status'] == 'Error':
        handle_failure(data)

    return jsonify({'status': 'received'}), 200

def handle_completion(data):
    generation_id = data['id']
    result_url = data['result']['sample']

    # Download image immediately (URL expires in 10 min)
    image_data = requests.get(result_url).content

    # Store to your storage
    store_image(generation_id, image_data)

    # Update your database
    update_generation_status(generation_id, 'completed')

    # Notify your application/users
    notify_completion(generation_id)

def handle_failure(data):
    generation_id = data['id']
    error = data.get('error', 'unknown')

    # Log the failure
    log_generation_failure(generation_id, error)

    # Update your database
    update_generation_status(generation_id, 'failed', error)

    # Maybe retry or notify
    handle_generation_error(generation_id, error)

注意处理器里 requests.get(result_url) 立即下载图片这一行，正是 “URL 十分钟过期” 约束的直接体现。

Express.js 处理器（原文档完整实现）：

const express = require('express');
const crypto = require('crypto');
const axios = require('axios');

const app = express();
app.use(express.raw({ type: 'application/json' }));

const WEBHOOK_SECRET = 'your-secret-key-here';

function verifySignature(payload, signature, secret) {
  if (!signature || !signature.startsWith('sha256=')) {
    return false;
  }

  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(payload)
    .digest('hex');

  const providedSignature = signature.slice(7);

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

app.post('/api/bfl-webhook', async (req, res) => {
  const signature = req.headers['x-bfl-signature'];

  if (!verifySignature(req.body, signature, WEBHOOK_SECRET)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  const data = JSON.parse(req.body);

  if (data.status === 'Ready') {
    // Download image (URL expires in 10 min)
    const imageResponse = await axios.get(data.result.sample, {
      responseType: 'arraybuffer'
    });

    // Store the image
    await storeImage(data.id, imageResponse.data);
  }

  res.json({ status: 'received' });
});

Node 版本用 express.raw({ type: 'application/json' }) 把 body 保留为原始字节（req.body 为 Buffer），再 JSON.parse(req.body)——这与 “签名必须基于原始 body” 的要求严格对应；校验用的是 crypto.timingSafeEqual，与 Python 的 hmac.compare_digest 同为恒定时间比较。

回调端点的基本要求与重试策略

Webhook 端点必须满足一组硬性约束，原文档在 “Requirements” 小节逐一列出：

必须使用 HTTPS。 生产环境中 Webhook URL 必须走 HTTPS，BFL 不会向 HTTP 端点发送回调。

响应要求：

• 用 2xx 状态码确认收到（acknowledge receipt）；
• 在 30 秒内响应；
• 保持 handler 快速——把重活卸到异步任务/队列里去做。

重试策略。 BFL 会对投递失败的 Webhook 进行重试：

尝试	延迟
第 1 次重试	1 秒
第 2 次重试	5 秒
第 3 次重试	30 秒

三次都失败后，该 Webhook 就会被放弃。原文档给出了一条关键兜底建议：“After 3 failed attempts, the webhook is abandoned. Fall back to polling if critical.”（三次失败后 Webhook 被放弃；对关键任务要能回退到轮询。）这正是下一节 “混合模式” 的出发点。

幂等性与去重

因为存在重试，回调可能被重复投递，端点必须幂等。原文档用 Redis 的 SET NX（不存在才写入）加 TTL 来实现去重：

from functools import lru_cache
import redis

redis_client = redis.Redis()

def is_duplicate_webhook(generation_id):
    """Check if we've already processed this webhook."""
    key = f"webhook:processed:{generation_id}"

    # Try to set with NX (only if not exists)
    was_set = redis_client.set(key, "1", nx=True, ex=3600)  # 1 hour TTL

    return not was_set  # If we couldn't set it, it's a duplicate

@app.route('/api/bfl-webhook', methods=['POST'])
def handle_webhook():
    # ... signature verification ...

    data = request.json
    generation_id = data['id']

    if is_duplicate_webhook(generation_id):
        return jsonify({'status': 'already_processed'}), 200

    # Process webhook...

要点：以 generation_id（即负载里的 id）作为去重键；ex=3600 给键设置 1 小时 TTL 防止内存无限增长；命中重复时仍返回 200，避免触发 BFL 再次重试。这与仓库工具层对 “幂等键” 的普遍关注是一致的——例如 flux_image.py 中声明了 idempotency_key_fields，从源码结构看，整个工具体系都把“同一输入只产生一次副作用”当作一等公民。

混合模式：Webhook 为主，轮询兜底

最稳健的做法是把 Webhook 和轮询结合起来：正常情况走事件驱动的 Webhook，超时未收到回调时回退到轮询 polling_url。原文档的 HybridClient 给出了完整骨架：

class HybridClient:
    def __init__(self, api_key, webhook_url, webhook_secret):
        self.api_key = api_key
        self.webhook_url = webhook_url
        self.webhook_secret = webhook_secret
        self.pending = {}  # Track pending generations

    def generate(self, prompt, timeout=300):
        """Generate with webhook, fall back to polling."""
        response = self._submit(prompt)
        generation_id = response['id']
        polling_url = response['polling_url']

        # Wait for webhook (with timeout)
        result = self._wait_for_webhook(generation_id, timeout=timeout)

        if result is None:
            # Webhook didn't arrive, fall back to polling
            result = self._poll(polling_url, timeout=60)

        return result

    def _submit(self, prompt):
        return requests.post(
            "https://api.bfl.ai/v1/flux-2-pro",
            headers={"x-key": self.api_key},
            json={
                "prompt": prompt,
                "webhook_url": self.webhook_url,
                "webhook_secret": self.webhook_secret
            }
        ).json()

    def receive_webhook(self, data):
        """Called by webhook handler."""
        generation_id = data['id']
        if generation_id in self.pending:
            self.pending[generation_id].set_result(data)

这个设计正好衔接了文档里的两条线：_submit 返回的 polling_url 是异步生成的标配字段（endpoints.md 的 “Polling Endpoint / Get Result” 定义了 GET /v1/get_result?id={generation_id}），而 _poll 的实现可以复用 polling-patterns.md 中给出的固定间隔、指数退避（含 jitter）或自适应三种策略——该文档强烈建议“实现指数退避 + jitter，避免多请求同时轮询时的惊群效应（thundering herd）”。receive_webhook 则由 Webhook 处理器在成功校验签名后被调用，用 id 关联回 pending 中的等待句柄，实现“谁提交了、谁来收”。

监控 Webhook 健康度

事件驱动不等于“发了就完事”，生产环境仍需跟踪 Webhook 本身的健康度。原文档给出一个轻量的 WebhookMetrics：

import time

class WebhookMetrics:
    def __init__(self):
        self.received = 0
        self.processed = 0
        self.failed = 0
        self.avg_latency = 0

    def record_webhook(self, generation_id, submit_time):
        self.received += 1
        latency = time.time() - submit_time
        self.avg_latency = (self.avg_latency * (self.received - 1) + latency) / self.received

    def record_success(self):
        self.processed += 1

    def record_failure(self):
        self.failed += 1

    def get_stats(self):
        return {
            "received": self.received,
            "processed": self.processed,
            "failed": self.failed,
            "success_rate": self.processed / max(self.received, 1),
            "avg_latency_seconds": self.avg_latency
        }

它统计四项指标：收到数、处理成功数、失败数、平均延迟（从提交到收到回调的端到端时长），并导出 success_rate。结合上一节的重试策略，这套指标能帮助你及时判断“Webhook 通道是否退化”，并在必要时触发回退到轮询。

与仓库实现的衔接（源码佐证）

为了把本文放回 OpenMontage 的实际工程语境，有两点值得从源码层面确认：

其一，BFL 的官方 API（api.bfl.ai）集成知识被沉淀为一整套 Agent 技能，入口是 SKILL.md（name: bfl-api，作者 Black Forest Labs）。它的 References 明确把本文对应的文档单列为 “references/webhook-integration.md - Webhook setup and security”，并与 endpoints.md、polling-patterns.md、rate-limiting.md、error-handling.md 并列，构成 “端点 / 轮询 / 限流 / 错误 / Webhook” 的完整知识闭环。技能还附带三份可直接运行的客户端示例：curl-examples.sh、python-client.py、typescript-client.ts。其中 Python 客户端同时内置了 verify_webhook_signature（签名校验）与 BFLClient.generate(webhook_url=..., webhook_secret=...)（带回调的提交），是本文各小节的权威落地版本。

其二，需要澄清 FLUX 模型在仓库里的两条不同接入路径，以免混淆。技能文档描述的是 BFL 原生 API（api.bfl.ai，异步、polling_url/Webhook）；而工具层 tools/graphics/flux_image.py 中的 FluxImage 工具则是通过 fal.ai（https://fal.run/fal-ai/{model}，读取 FAL_KEY）做同步调用，其 execution_mode = ExecutionMode.SYNC，在 execute() 里一次性 POST、直接取回 images[0].url 并下载。从源码结构看，该工具把 "bfl-api" 写进了自己的 agent_skills 列表，意味着它把 FLUX 相关的提示词与集成知识交由 bfl-api 技能提供。因此，Webhook / 事件驱动这套模式适用于你直接对接 BFL 原生 API 的生产集成场景；而仓库内置的 flux_image 工具走的是另一条 fal.ai 同步通道。两者面向同一批 FLUX 模型，但在鉴权（x-key vs Authorization: Key ...）、密钥（BFL_API_KEY vs FAL_KEY）、返回形态（异步 polling_url vs 同步 images）上各不相同——选择哪一条，取决于你实际接入的是 BFL 官方 API 还是 fal.ai 托管端点。

参考文件

• Webhook 集成参考文档（本文主体）
• bfl-api 技能主文档
• 端点文档
• 轮询模式
• 错误处理
• 限流
• Python 客户端（含签名校验与带回调提交）
• TypeScript 客户端
• cURL 示例
• 仓库内置 FLUX 工具（fal.ai 同步通道）