BFL FLUX API 集成实战指南：端点选型、异步轮询与 Webhook 全链路解析

本指南基于 OpenMontage 仓库中的 BFL API 集成 Skill 文档编写，系统覆盖 API Key 配置、区域端点选型、FLUX.2/FLUX.1 模型端点与定价、异步轮询模式、速率限制处理、错误恢复及生产级 Webhook 集成的完整链路，帮助开发者从第一个 curl 请求一路推进到可上线的 Python/TypeScript 客户端实现。

---

一、快速了解 BFL API 的核心工作模型

BFL（Black Forest Labs）API 采用异步生成模型：所有图像生成请求都不是同步返回结果的，提交后返回一个 polling_url，客户端需轮询该 URL 直到状态变为 Ready，再立即下载结果图片。

整个核心流程只有三步：

1. POST 到模型端点
   └─> 响应: { "polling_url": "..." }

2. GET polling_url（重复直到完成）
   └─> 响应: { "status": "Pending" | "Ready" | "Error", ... }

3. 状态为 Ready 时，立即下载结果 URL
   └─> URL 有效期仅 10 分钟，必须立即下载

理解这个三步模型是后续所有内容的基石。

---

二、API Key 配置

2.1 验证 Key 是否已配置

每次使用 API 之前，先确认环境变量已设置：

echo $BFL_API_KEY

更严谨的验证方式：

[ -z "$BFL_API_KEY" ] && echo "Error: BFL_API_KEY not set" || echo "OK: Key configured"

若返回空或出现 "Not authenticated" 错误，需要执行以下获取流程。

2.2 获取 API Key

1. 访问 BFL Dashboard 的 Get Started 页面，点击 "Create Key"
2. 选择组织（若有多个组织需确认）
3. 复制生成的 Key（以 bfl_ 开头）

2.3 持久化保存到 .env

echo 'BFL_API_KEY=bfl_your_key_here' >> .env
echo '.env' >> .gitignore

在 Agent 工作流中，若 .env 已存在，可直接导出到当前会话：

grep BFL_API_KEY .env 2>/dev/null && export BFL_API_KEY=$(grep BFL_API_KEY .env | cut -d '=' -f2)

完整配置说明见 api-key-setup.md

---

三、认证方式

所有 BFL API 请求都必须携带 x-key Header：

x-key: YOUR_API_KEY

这是唯一认证方式，无需 OAuth 或 Bearer Token。

---

四、区域端点与端点选型

4.1 三大区域端点

区域	端点	适用场景
Global	https://api.bfl.ai	默认端点，支持自动故障转移
EU	https://api.eu.bfl.ai	GDPR 合规，欧盟数据驻留
US	https://api.us.bfl.ai	美国数据驻留

选型建议：无特殊合规要求时优先使用 Global 端点，它具备自动故障转移能力；有 EU 数据驻留要求时切换到 EU 端点。

4.2 模型端点总览

FLUX.2 系列（支持 T2I + I2I）

模型	路径	首 MP 价格	每 MP 价格	1MP T2I	1MP I2I	最佳适用场景
FLUX.2 [klein] 4B	/v1/flux-2-klein-4b	1.4c	0.1c	$0.014	$0.015	实时、大批量生成
FLUX.2 [klein] 9B	/v1/flux-2-klein-9b	1.5c	0.2c	$0.015	$0.017	质量/速度均衡
FLUX.2 [pro]	/v1/flux-2-pro	3c	1.5c	$0.03	$0.045	生产环境，快速交付
FLUX.2 [max]	/v1/flux-2-max	7c	3c	$0.07	$0.10	最高质量，支持 grounding search
FLUX.2 [flex]	/v1/flux-2-flex	5c	5c	$0.05	$0.10	文字排版，可调 steps/guidance
FLUX.2 [dev]	—	—	—	Free	Free	本地开发（非商用）

定价公式：(firstMP + (outputMP-1) × mpPrice) + (inputMP × mpPrice)，单位为美分。1 credit = $0.01 USD。FLUX.2 按实际输出分辨率计费，分辨率越高成本越高。

FLUX.1 系列

模型	路径	单张价格	最佳适用场景
FLUX.1 Kontext [pro]	/v1/flux-kontext	$0.04	带上下文的图像编辑
FLUX.1 Kontext [max]	/v1/flux-kontext-max	$0.08	最高质量编辑（速率限制为 6 并发，非 24）
FLUX1.1 [pro]	/v1/flux-pro-1.1	$0.04	标准 T2I，快速可靠
FLUX1.1 [pro] Ultra	/v1/flux-pro-1.1-ultra	$0.06	超高分辨率
FLUX1.1 [pro] Raw	/v1/flux-pro-1.1-raw	$0.06	纪实摄影风格
FLUX.1 Fill [pro]	/v1/flux-pro-1.0-fill	$0.05	局部重绘（Inpainting）

提示：所有 FLUX.2 模型均通过 input_image 参数支持图像编辑，无需单独的编辑端点。编辑场景优先推荐 FLUX.2 系列而非 FLUX.1 Kontext。

完整端点文档见 endpoints.md

---

五、请求参数详解

5.1 T2I（文本到图像）通用参数

参数	类型	必填	说明
prompt	string	是	文本描述，最多 32K tokens
width	integer	否	图像宽度，必须为 16 的倍数，总像素不超过 4MP
height	integer	否	图像高度，必须为 16 的倍数，总像素不超过 4MP
seed	integer	否	随机种子，用于复现结果
safety_tolerance	integer	否	0（严格）~ 5（宽松），默认值 2
output_format	string	否	"jpeg" 或 "png"，默认 "jpeg"
webhook_url	string	否	生成完成后的回调 URL
webhook_secret	string	否	Webhook 签名密钥

5.2 I2I（图像到图像）参数

参数	类型	必填	说明
prompt	string	是	编辑指令
input_image	string	是	优先传 URL，API 自动抓取；也支持 base64
input_image_2 ~ input_image_8	string	否	额外参考图（URL 或 base64）
width	integer	否	输出宽度
height	integer	否	输出高度

URL 优先原则：API 会自动抓取 URL 指向的图片，无需客户端下载再编码为 base64。当图片有可访问 URL 时，强烈建议直接使用 URL。

5.3 FLUX.2 [flex] 专属参数

参数	类型	默认值	说明
steps	integer	50	推理步数，范围 1~50
guidance	float	4.5	引导强度，范围 1.5~10

5.4 分辨率约束

• 最小尺寸：64 × 64 像素
• 最大总像素：4MP（即 width × height ≤ 4,000,000）
• 对齐要求：宽度和高度均须为 16 的倍数

宽高比	推荐分辨率	像素数
1:1（方形）	1024 × 1024	1.05 MP
16:9（宽屏）	1920 × 1080	2.07 MP
9:16（竖屏）	1080 × 1920	2.07 MP
4:3（经典）	1536 × 1152	1.77 MP
2:1（全景）	2048 × 1024	2.10 MP

---

六、快速上手：cURL 四步完整流程

6.1 第一步：提交生成请求

curl -s -X POST "https://api.bfl.ai/v1/flux-2-pro" \
  -H "x-key: $BFL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A serene mountain landscape at sunset",
    "width": 1024,
    "height": 1024
  }'

响应：

{
  "id": "abc123",
  "polling_url": "https://api.bfl.ai/v1/get_result?id=abc123"
}

6.2 第二步：轮询结果

curl -s "POLLING_URL" -H "x-key: $BFL_API_KEY"

当生成完成时，响应为：

{
  "status": "Ready",
  "result": {
    "sample": "https://bfldeliveryprod.blob.core.windows.net/results/...",
    "prompt": "A serene mountain landscape at sunset",
    "seed": 1234567890
  }
}

轮询过程中可能出现三种状态：

状态	含义	客户端操作
Pending	请求已排队/处理中	继续轮询
Ready	生成完成	立即下载 result.sample
Error	生成失败	处理错误信息

6.3 第三步：立即下载图片

curl -s -o output.png "IMAGE_URL"

⚠️ 关键：结果 URL 有效期仅 10 分钟，状态变为 Ready 后必须立即下载，切勿存储 URL 本身。

6.4 完整 bash 脚本示例

仓库提供了一份可直接运行的 cURL 脚本，涵盖提交、轮询、下载、I2I 编辑全流程：

# 设置变量
API_KEY="${BFL_API_KEY:-YOUR_API_KEY}"
BASE_URL="https://api.bfl.ai"

# 提交请求
RESPONSE=$(curl -s -X POST "${BASE_URL}/v1/flux-2-pro" \
  -H "x-key: ${API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A serene mountain landscape at golden hour, dramatic lighting",
    "width": 1024,
    "height": 1024
  }')

# 提取 polling_url
POLLING_URL=$(echo "${RESPONSE}" | grep -o '"polling_url":"[^"]*"' | cut -d'"' -f4)

# 轮询直到完成
while true; do
  RESULT=$(curl -s "${POLLING_URL}" -H "x-key: ${API_KEY}")
  STATUS=$(echo "${RESULT}" | grep -o '"status":"[^"]*"' | cut -d'"' -f4)
  echo "Status: ${STATUS}"
  if [ "${STATUS}" = "Ready" ]; then
    IMAGE_URL=$(echo "${RESULT}" | grep -o '"sample":"[^"]*"' | cut -d'"' -f4)
    break
  elif [ "${STATUS}" = "Error" ]; then
    echo "Generation failed!"
    exit 1
  fi
  sleep 2
done

# 下载图片
curl -s -o output.png "${IMAGE_URL}"

完整脚本见 curl-examples.sh

---

七、图像编辑与多参考图

7.1 单图编辑（I2I）

所有 FLUX.2 模型均支持 I2I 编辑，通过 input_image 参数传入：

curl -X POST "https://api.bfl.ai/v1/flux-2-pro" \
  -H "x-key: $BFL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "Change the background to a sunset",
    "input_image": "https://example.com/photo.jpg"
  }'

7.2 多参考图（Multi-Reference I2I）

FLUX.2 系列支持多张参考图输入，可用于元素组合、风格迁移、角色一致性等场景：

模型系列	最大参考图数
FLUX.2 [klein]	4 张
FLUX.2 [pro / max / flex]	8 张

参数命名规则：input_image、input_image_2、input_image_3 … input_image_8

提示词模式：在 prompt 中按编号引用图片：

• "The subject from image 1 in the environment from image 2"
• "Apply the style of image 2 to the scene in image 1"
• "The person from image 1 wearing the outfit from image 2, in the pose from image 3"

完整示例：

curl -s -X POST "https://api.bfl.ai/v1/flux-2-max" \
  -H "x-key: $BFL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "Person from image 1 wearing outfit from image 2 in setting from image 3",
    "input_image": "https://example.com/person.jpg",
    "input_image_2": "https://example.com/outfit.jpg",
    "input_image_3": "https://example.com/location.jpg"
  }'

多参考图的进阶用法（角色一致性、风格迁移、姿势引导）详见 multi-reference-editing.md

---

八、速率限制与并发控制

8.1 当前速率限制

端点类别	并发请求上限
标准端点（大多数模型）	24
flux-kontext-max	6

"并发请求"指已提交但尚未完成的 in-flight 请求数，而非 QPS。

8.2 响应头监控

每次响应包含以下速率限制相关 Header：

X-RateLimit-Limit: 24
X-RateLimit-Remaining: 23
X-RateLimit-Reset: 1640000000

8.3 HTTP 429 响应体

触发速率限制时返回：

{
  "error": "rate_limit_exceeded",
  "message": "Too many concurrent requests",
  "retry_after": 5
}

8.4 客户端并发控制策略

策略一：信号量限制（推荐）

from threading import Semaphore

class RateLimitedClient:
    def __init__(self, api_key, max_concurrent=24):
        self.api_key = api_key
        self.semaphore = Semaphore(max_concurrent)

    def generate(self, model, prompt, **kwargs):
        with self.semaphore:  # 达到上限时阻塞等待
            return self._submit_and_poll(model, prompt, **kwargs)

策略二：429 指数退避重试

def request_with_retry(endpoint, payload, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.post(endpoint, json=payload, headers=headers)
        if response.status_code == 429:
            retry_after = int(response.headers.get('Retry-After', 5))
            wait_time = retry_after * (2 ** attempt)
            time.sleep(wait_time)
            continue
        response.raise_for_status()
        return response
    raise Exception("Max retries exceeded due to rate limiting")

策略三：队列缓冲（高吞吐量场景）

from queue import Queue
from threading import Thread

class RequestQueue:
    def __init__(self, api_key, max_concurrent=24):
        self.queue = Queue()
        for _ in range(max_concurrent):
            worker = Thread(target=self._worker, daemon=True)
            worker.start()

    def submit(self, model, prompt, callback):
        self.queue.put({'model': model, 'prompt': prompt, 'callback': callback})

策略四：异步并发（Python asyncio + aiohttp）

import asyncio, aiohttp

class AsyncRateLimitedClient:
    def __init__(self, api_key, max_concurrent=24):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.headers = {"x-key": api_key}

    async def generate(self, model, prompt):
        async with self.semaphore:
            async with aiohttp.ClientSession() as session:
                # 提交 + 轮询
                ...

async def main():
    client = AsyncRateLimitedClient("your-api-key")
    prompts = [f"Image {i}" for i in range(50)]
    results = await asyncio.gather(*[client.generate("flux-2-pro", p) for p in prompts])

asyncio.run(main())

⚠️ 注意：flux-kontext-max 端点的并发上限为 6，而非 24。Python 客户端中的 RATE_LIMITS 字典已对此做了区分。

完整速率限制文档见 rate-limiting.md

---

九、轮询策略深入解析

9.1 三种轮询模式对比

策略	适用场景	特点
固定间隔	本地开发、脚本	实现简单，对服务器压力大
指数退避 + Jitter	生产环境推荐	降低并发压力，避免惊群
自适应轮询	状态敏感场景	根据 Pending 状态动态调整间隔

9.2 指数退避 + Jitter（推荐实现）

import time, random, requests

def poll_with_backoff(polling_url, headers, max_attempts=30):
    """指数退避 + 随机抖动，防止惊群效应。"""
    base_delay = 0.5   # 起始 500ms
    max_delay = 10.0   # 上限 10 秒

    for attempt in range(max_attempts):
        response = requests.get(polling_url, headers=headers)
        data = response.json()

        if data["status"] == "Ready":
            return data["result"]
        elif data["status"] == "Error":
            raise Exception(data.get("error", "Generation Error"))

        delay = min(base_delay * (2 ** attempt), max_delay)
        jitter = random.uniform(0, delay * 0.1)  # 10% 抖动
        time.sleep(delay + jitter)

    raise TimeoutError("Max polling attempts exceeded")

关键设计要点：

1. 必须有超时上限——永远不要无限轮询
2. Jitter 很重要——同时轮询多个请求时，固定间隔会导致所有客户端在同一时刻发出请求，Jitter 可打散这种同步
3. 处理所有状态值——包括未预期到的状态
4. 下载必须紧随 Ready——URL 10 分钟过期

9.3 完整 Python 客户端

仓库提供了一份生产级 Python 客户端，集成速率限制、重试、Webhook 支持、批量处理和异步操作：

import os, requests

from bfl_client import BFLClient

api_key = os.environ["BFL_API_KEY"]
client = BFLClient(api_key)

# 单张生成
result = client.generate(
    model="flux-2-pro",
    prompt="A serene mountain landscape at golden hour, dramatic lighting",
    width=1024,
    height=1024,
)
print(f"Image URL: {result.url}")

# 立即下载（URL 10 分钟过期）
client.download(result.url, "output.png")

# 批量生成（并发，受信号量保护）
results = client.generate_batch(
    "flux-2-pro",
    prompts=["Mountain", "Ocean", "Forest"],
)

客户端内部通过 _validate_dimensions 在发送请求前校验分辨率约束（16 倍数、4MP 上限、最小 64px），避免触发 400 错误。

完整 Python 客户端源码见 python-client.py

9.4 批量异步处理

import asyncio, aiohttp

async def generate_batch(session, client, prompts, model="flux-2-pro"):
    tasks = [submit_and_poll(session, client, model, p) for p in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

async def submit_and_poll(session, client, model, prompt):
    # 提交
    async with session.post(
        f"{client.base_url}/v1/{model}",
        headers=client.headers,
        json={"prompt": prompt}
    ) as resp:
        data = await resp.json()
        polling_url = data["polling_url"]

    # 轮询
    while True:
        async with session.get(polling_url, headers=client.headers) as resp:
            data = await resp.json()
            if data["status"] == "Ready":
                return data["result"]
            elif data["status"] == "Error":
                raise Exception(data.get("error"))
        await asyncio.sleep(2)

完整轮询模式文档见 polling-patterns.md

---

十、错误处理与重试

10.1 HTTP 状态码完整对照表

状态码	含义	原因	处理策略
200	OK	请求成功	正常处理
400	Bad Request	参数无效	检查请求格式，不可重试
401	Unauthorized	API Key 无效/缺失	验证凭证，不可重试
402	Payment Required	余额不足	充值，不可重试
403	Forbidden	权限不足	检查权限，不可重试
404	Not Found	端点不存在	验证 URL，不可重试
429	Too Many Requests	触发速率限制	指数退避重试，可重试
500	Internal Server Error	服务器内部错误	指数退避重试，可重试
502	Bad Gateway	网络问题	指数退避重试，可重试
503	Service Unavailable	服务暂时不可用	指数退避重试，可重试

10.2 错误响应体结构

{
  "error": "error_code",
  "message": "Human-readable description",
  "details": {
    "field": "specific field info"
  }
}

10.3 常见错误场景与响应

401 — 认证失败

{
  "error": "invalid_api_key",
  "message": "The provided API key is invalid or expired"
}

处理：验证 Key 是否以 bfl_ 开头，检查是否过期。

402 — 余额不足

{
  "error": "insufficient_credits",
  "message": "Your account does not have enough credits"
}

处理：记录日志并告警，可选择暂停批量操作。

400 — 参数校验失败

{
  "error": "validation_error",
  "message": "Invalid request parameters",
  "details": {
    "width": "Must be a multiple of 16",
    "prompt": "Cannot be empty"
  }
}

生成失败（轮询阶段）

{
  "status": "Error",
  "error": "content_policy_violation",
  "message": "The prompt violated content policy"
}

常见失败原因：

错误码	含义	是否可重试
content_policy_violation	内容策略违规	否
generation_timeout	生成超时	是
internal_error	服务端错误	是
invalid_image	输入图无法处理	否

10.4 重试分类策略

class RetryableError(Exception): pass
class NonRetryableError(Exception): pass

def classify_error(status_code, error_code):
    # 可重试
    if status_code in (429, 500, 502, 503):
        return RetryableError
    # 不可重试
    if status_code in (400, 401, 402, 403):
        return NonRetryableError
    # 生成失败码
    if error_code in ('generation_timeout', 'internal_error'):
        return RetryableError
    if error_code in ('content_policy_violation', 'invalid_image'):
        return NonRetryableError
    return RetryableError  # 默认可重试

10.5 熔断器模式（生产环境）

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.state = "closed"  # closed / open / half-open

    def can_proceed(self):
        if self.state == "open" and \
           time.time() - self.last_failure_time > self.reset_timeout:
            self.state = "half-open"
            return True
        return self.state != "open"

完整错误处理文档见 error-handling.md

---

十一、Webhook 生产级集成

11.1 Polling 与 Webhook 选型

方案	适用场景
Polling	脚本、CLI 工具、本地开发、单次请求、简单集成
Webhook	生产应用、高吞吐量、服务器到服务器、需要即时通知

建议：从 Polling 开始，它在任何环境都能工作；当需要扩展或事件驱动架构时再切换到 Webhook。

11.2 带 Webhook 的请求

在请求体中加入 webhook_url 和可选的 webhook_secret：

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"
  }'

11.3 Webhook 回调 Payload

生成成功：

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

生成失败：

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

11.4 HMAC-SHA256 签名验证

当提供 webhook_secret 时，BFL 会对 Payload 进行 HMAC-SHA256 签名，通过 X-BFL-Signature Header 传递：

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

Python 验证实现：

import hmac, hashlib

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    if not signature or not signature.startswith('sha256='):
        return False

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

    provided = signature[7:]  # 去掉 'sha256=' 前缀
    return hmac.compare_digest(expected, provided)

TypeScript 验证实现：

import * as crypto from "crypto";

function verifyWebhookSignature(payload: string, signature: string, secret: string): boolean {
  if (!signature || !signature.startsWith('sha256=')) return false;

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

  const provided = signature.slice(7);

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

11.5 Flask Webhook 处理器（含完整验证）

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

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

@app.route('/api/bfl-webhook', methods=['POST'])
def handle_webhook():
    # 1. 验证签名
    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':
        generation_id = data['id']
        result_url = data['result']['sample']

        # 2. 立即下载图片（URL 10 分钟过期）
        image_data = requests.get(result_url).content

        # 3. 存储 + 更新状态 + 通知
        store_image(generation_id, image_data)
        update_generation_status(generation_id, 'completed')
        notify_completion(generation_id)

    elif data['status'] == 'Error':
        log_generation_failure(data['id'], data.get('error', 'unknown'))
        update_generation_status(data['id'], 'failed', data.get('error'))

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

11.6 Express.js Webhook 处理器

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

const app = express();
app.use(express.raw({ type: 'application/json' }));  // 保留原始 body 用于签名验证

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

function verifySignature(payload, signature, secret) {
  if (!signature || !signature.startsWith('sha256=')) return false;
  const expected = crypto.createHmac('sha256', secret).update(payload).digest('hex');
  return crypto.timingSafeEqual(
    Buffer.from(expected),
    Buffer.from(signature.slice(7))
  );
}

app.post('/api/bfl-webhook', async (req, res) => {
  if (!verifySignature(req.body, req.headers['x-bfl-signature'], WEBHOOK_SECRET)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  const data = JSON.parse(req.body);
  if (data.status === 'Ready') {
    const imageResponse = await axios.get(data.result.sample, { responseType: 'arraybuffer' });
    await storeImage(data.id, imageResponse.data);
  }
  res.json({ status: 'received' });
});

11.7 Webhook 生产要求

HTTPS 强制：Webhook URL 必须使用 HTTPS，BFL 不会向 HTTP 端点发送回调。

响应要求：

• 必须返回 2xx 状态码确认接收
• 必须在 30 秒内响应
• 保持处理器轻量，重处理逻辑应异步 offload

重试策略：

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

3 次失败后 BFL 放弃投递，关键场景建议实现 Webhook + Polling 的混合方案作为兜底。

11.8 幂等性处理

由于 Webhook 可能重复投递，必须做幂等性处理：

import redis

redis_client = redis.Redis()

def is_duplicate_webhook(generation_id: str) -> bool:
    key = f"webhook:processed:{generation_id}"
    # NX: 仅当 key 不存在时设置；EX 3600: 1 小时 TTL
    was_set = redis_client.set(key, "1", nx=True, ex=3600)
    return not was_set  # 无法设置说明已处理过

11.9 混合方案（Webhook + Polling 兜底）

class HybridClient:
    def generate(self, prompt, timeout=300):
        response = self._submit(prompt)
        generation_id = response['id']
        polling_url = response['polling_url']

        # 等待 Webhook 回调（带超时）
        result = self._wait_for_webhook(generation_id, timeout=timeout)

        if result is None:
            # Webhook 未到达，降级为 Polling
            result = self._poll(polling_url, timeout=60)

        return result

完整 Webhook 集成文档见 webhook-integration.md

---

十二、TypeScript 客户端

仓库提供了一份功能完整的 TypeScript 客户端，结构与 Python 版对应：

import { BFLClient } from './bfl-client';

const client = new BFLClient(process.env.BFL_API_KEY!);

// 单张生成
const result = await client.generate(
  'flux-2-pro',
  'A serene mountain landscape at golden hour',
  { width: 1024, height: 1024, outputFormat: 'png' }
);
console.log(`Image URL: ${result.url}`);
console.log(`Image ID:  ${result.id}`);

// I2I 多参考图
const editResult = await client.generateI2I(
  'flux-2-pro',
  'Change the background to a sunset',
  'https://example.com/photo.jpg',
  { additionalImages: ['https://example.com/ref2.jpg'] }
);

// 批量并发
const results = await client.generateBatch('flux-2-pro', [
  'Mountain scene', 'Ocean sunset', 'Forest path'
]);

TypeScript 客户端内部实现了与 Python 版一致的 Semaphore 并发控制、指数退避重试（上限 5 秒）、_validateDimensions 预校验以及完整的 BFLError 异常层次（AuthenticationError、InsufficientCreditsError、RateLimitError、ValidationError、GenerationError）。

完整 TypeScript 客户端源码见 typescript-client.ts

---

十三、FLUX.2 [flex] 模型的特殊用法

flex 模型是 BFL 系列中唯一支持显式控制推理参数的模型，适合文字排版（Typography）和对生成质量有精细控制需求的场景：

curl -X POST "https://api.bfl.ai/v1/flux-2-flex" \
  -H "x-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A poster with text \"SUMMER SALE\" in bold typography",
    "steps": 50,
    "guidance": 7.0
  }'

参数	默认值	范围	调高效果
steps	50	1~50	更多细节，更慢
guidance	4.5	1.5~10	更贴合 prompt，过度会导致伪影

---

十四、相关 Skill 与延伸阅读

本 Skill 文档与 OpenMontage 中的 flux-best-practices Skill 互补，两者分工如下：

关注点	所在 Skill
API 端点、参数、认证、轮询、速率限制、错误处理、Webhook	bfl-api（本文档）
T2I/I2I 提示词写法、文字排版、颜色、多参考图模式、模型选型	flux-best-practices

多参考图进阶模式（角色一致性、风格迁移、姿势引导）的完整指南见 multi-reference-editing.md

---

十五、文件索引

文件	说明
SKILL.md	Skill 主入口，快速参考与四步流程
api-key-setup.md	API Key 获取与配置
endpoints.md	完整端点、参数、分辨率约束
polling-patterns.md	三种轮询策略 + 完整 Python 客户端
rate-limiting.md	并发控制四种策略 + 监控
error-handling.md	状态码、重试分类、熔断器
webhook-integration.md	签名验证、Flask/Express 处理器、幂等性
curl-examples.sh	可直接运行的 bash 脚本
python-client.py	生产级 Python 客户端（476 行）
typescript-client.ts	生产级 TypeScript 客户端（480 行）