Bfl Api

BFL API Integration Guide

Use this skill when integrating BFL FLUX APIs into applications for image generation, editing, and processing.

First: Check API Key

Before generating images, verify your API key is set:

echo $BFL_API_KEY

If empty or you see "Not authenticated" errors, see API Key Setup below.

Important: Image URLs Expire in 10 Minutes

Result URLs from the API are temporary. Download images immediately after generation completes - do not store or cache the URLs themselves.

When to Use

• Setting up BFL API client
• Implementing async polling patterns
• Handling rate limits and errors
• Configuring webhooks for production
• Selecting regional endpoints
• Building production-ready integrations

Quick Reference

Base Endpoints

Model Endpoints & Pricing

Credit pricing: 1 credit = $0.01 USD. FLUX.2 uses megapixel-based pricing (cost scales with resolution).

FLUX.2 Models

Pricing formula: (firstMP + (outputMP-1) * mpPrice) + (inputMP * mpPrice) in cents

FLUX.1 Models

Tip: All FLUX.2 models support image editing via the input_image parameter - no separate editing endpoint needed. Use bfl.ai/pricing calculator for exact costs at different resolutions.

Image Input for Editing

Preferred: Use URLs directly - simpler and more convenient than base64.

Single image editing:

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

Multi-reference editing:

curl -X POST "https://api.bfl.ai/v1/flux-2-pro" \
  -H "x-key: $BFL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "The person from image 1 in the environment from image 2",
    "input_image": "https://example.com/person.jpg",
    "input_image_2": "https://example.com/background.jpg"
  }'

The API fetches URLs automatically. Both URL and base64 work, but URLs are recommended when available.

Multi-Reference I2I

FLUX.2 models support multiple input images for combining elements, style transfer, and character consistency:

Parameters: input_image, input_image_2, input_image_3, ... input_image_8

Prompt pattern: Reference images by number in your 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"

For detailed multi-reference patterns (character consistency, style transfer, pose guidance), see flux-best-practices/rules/multi-reference-editing.md

Rate Limits

Polling vs Webhooks

Start with polling - it's simpler and works everywhere. Switch to webhooks when you need to scale or want event-driven architecture.

Key Behaviors

• Polling: Response includes polling_url for async results
• URL Expiration: Result URLs expire after 10 minutes
• Webhook Support: Configure webhook_url for production workloads

API Key Setup

Required: The BFL_API_KEY environment variable must be set before using the API.

Quick Check

echo $BFL_API_KEY

If Not Set

1. Get a key: Go to https://dashboard.bfl.ai/get-started → Click "Create Key" → Select organization 2. Save to `.env` (recommended for persistence):

   echo 'BFL_API_KEY=bfl_your_key_here' >> .env
   echo '.env' >> .gitignore  # Don't commit secrets

See references/api-key-setup.md for detailed setup instructions.

Authentication

x-key: YOUR_API_KEY

Basic Request Flow

1. POST request to model endpoint
   └─> Response: { "polling_url": "..." }

2. GET polling_url (repeat until complete)
   └─> Response: { "status": "Pending" | "Ready" | "Error", ... }

3. When Ready, download result URL
   └─> URL expires in 10 minutes - download immediately

Related

• Prompting best practices (T2I, I2I, typography, colors): see the flux-best-practices skill
• Multi-reference patterns (character consistency, style transfer, pose guidance): see flux-best-practices/rules/multi-reference-editing.md

References

• references/api-key-setup.md - API key creation and configuration
• references/endpoints.md - Complete endpoint documentation
• references/polling-patterns.md - Async polling implementation
• references/rate-limiting.md - Rate limit handling strategies
• references/error-handling.md - Error codes and recovery
• references/webhook-integration.md - Webhook setup and security

Code Examples

Note: cURL examples are preferred by default as they work universally without requiring Python or Node.js. Use language-specific clients when building production applications.

• references/code-examples/curl-examples.sh - cURL examples (recommended)
• references/code-examples/python-client.py - Python client
• references/code-examples/typescript-client.ts - TypeScript client

Quick Start Example

1. Submit Generation Request

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

Response:

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

2. Poll for Result

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

Response when ready:

{ "status": "Ready", "result": { "sample": "https://...", "seed": 1234 } }

3. Download Image

curl -s -o output.png "IMAGE_URL"

Tip: Result URLs expire in 10 minutes. Download immediately after status becomes Ready.

4. Multi-Reference Example

Combine elements from multiple images:

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": "The cat from image 1 sitting in the cozy room from image 2",
    "input_image": "https://example.com/cat.jpg",
    "input_image_2": "https://example.com/room.jpg",
    "width": 1024,
    "height": 1024
  }'

Reference images by number in your prompt. See Multi-Reference I2I for limits and patterns.

API Key Setup

Important: Always verify your API key before attempting image generation. Missing or invalid keys result in "Not authenticated" errors.

Quick Validation

Run this first to check if your key is configured and valid:

# Check if key is set
[ -z "$BFL_API_KEY" ] && echo "Error: BFL_API_KEY not set" || echo "OK: Key configured"

If not set, follow the steps below.

Get a Key

1. Go to https://dashboard.bfl.ai/get-started 2. Click "Create Key" 3. Select organization (ask user if multiple options) 4. Copy the key (starts with bfl_)

For Agents Making Direct API Calls

When BFL_API_KEY is not set in the current session:

1. Check for existing `.env`:

   grep BFL_API_KEY .env 2>/dev/null

2. If found, export it:

   export BFL_API_KEY=$(grep BFL_API_KEY .env | cut -d '=' -f2)

3. If not found, ask the user for their key:

"I need a BFL API key to generate images. Please:

1. Go to https://dashboard.bfl.ai/get-started

2. Click 'Create Key' and copy it

3. Paste it here"

4. Save and export:

   echo 'BFL_API_KEY=bfl_provided_key' >> .env
   echo '.env' >> .gitignore
   export BFL_API_KEY=bfl_provided_key

Now $BFL_API_KEY is available for direct curl/API calls in the session.

BFL API Endpoints

Complete reference for all BFL FLUX API endpoints.

Base URLs

Recommendation: Use the global endpoint (api.bfl.ai) unless you have specific regional requirements.

Authentication

All requests require the x-key header with your API key:

x-key: YOUR_API_KEY

FLUX.2 Text-to-Image and Image-to-Image Endpoints

FLUX.2 [klein] 4B

POST /v1/flux-2-klein-4b

Fastest generation, 4B parameters.

FLUX.2 [klein] 9B

POST /v1/flux-2-klein-9b

Fast generation with better quality, 9B parameters.

FLUX.2 [max]

POST /v1/flux-2-max

Highest quality, supports grounding search.

FLUX.2 [pro]

POST /v1/flux-2-pro

Production balanced quality and speed.

FLUX.2 [flex]

POST /v1/flux-2-flex

Typography optimized, adjustable steps/guidance.

FLUX.1 Endpoints

FLUX1.1 [pro]

POST /v1/flux-pro-1.1

Text-to-image generation.

FLUX.1 Kontext

POST /v1/flux-kontext

FLUX.1 Kontext Max

POST /v1/flux-kontext-max

FLUX.1 Fill

POST /v1/flux-fill

Inpainting and object removal - you can achieve inpainting and object removal with specific prompting style with FLUX.2 models for better performance.

Common Request Parameters

Text-to-Image (T2I)

Image-to-Image (I2I)

Important: All FLUX.2 models (klein, pro, max, flex) support image-to-image editing via the input_image parameter. FLUX.2 is recommended over FLUX.1 Kontext for editing.

Preferred: Use URLs directly - The API fetches URLs automatically, which is simpler and more convenient than downloading and encoding to base64. Both URL and base64 work, but URLs are recommended when available.

FLUX.2 [flex] Specific

Resolution Constraints

• Minimum: 64x64 pixels
• Maximum: 4MP total (width x height)
• Multiple of: 16 (both dimensions)

Common Resolutions

Example Requests

Basic T2I Request

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 serene mountain landscape at golden hour",
    "width": 1024,
    "height": 1024
  }'

Response

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

T2I with All Options

curl -X POST "https://api.bfl.ai/v1/flux-2-max" \
  -H "x-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "Professional headshot of a business executive",
    "width": 1024,
    "height": 1280,
    "seed": 42,
    "safety_tolerance": 2,
    "output_format": "png",
    "webhook_url": "https://your-server.com/webhook",
    "webhook_secret": "your-secret-key"
  }'

I2I Request (FLUX.2 - Recommended)

Edit images using any FLUX.2 model by passing the source image URL directly:

curl -X POST "https://api.bfl.ai/v1/flux-2-klein-9b" \
  -H "x-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "Change the floor color to light blue",
    "input_image": "https://example.com/room-photo.jpg"
  }'

For higher quality edits, use FLUX.2 [pro] or [max]:

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

Multi-Reference I2I (FLUX.2)

curl -X POST "https://api.bfl.ai/v1/flux-2-max" \
  -H "x-key: YOUR_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"
  }'

FLUX.2 [flex] with Custom Steps

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

Polling Endpoint

Get Result

GET /v1/get_result?id={generation_id}

Response States

// Pending
{ "status": "Pending" }

// Ready
{
  "status": "Ready",
  "result": {
    "sample": "https://bfldeliveryprod.blob.core.windows.net/results/...",
    "prompt": "...",
    "seed": 1234567890
  }
}

// Error
{
  "status": "Error",
  "error": "Error description"
}

Error Responses

Error Handling

Comprehensive guide to handling errors from the BFL API.

HTTP Status Codes

Error Response Format

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

Common Errors and Solutions

Authentication Errors (401)

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

Solution:

def verify_api_key(api_key):
    if not api_key:
        raise ValueError("API key is required")
    if not api_key.startswith("bfl_"):
        raise ValueError("Invalid API key format")

Insufficient Credits (402)

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

Solution:

def handle_payment_error(response):
    if response.status_code == 402:
        # Log and alert
        logging.error("Insufficient credits - add funds")
        # Optionally pause operations
        raise InsufficientCreditsError("Add credits to continue")

Rate Limiting (429)

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

Solution:

def handle_rate_limit(response):
    if response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 5))
        time.sleep(retry_after)
        return True  # Signal to retry
    return False

Validation Errors (400)

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

Solution:

def validate_request(prompt, width, height):
    errors = []

    if not prompt or not prompt.strip():
        errors.append("Prompt cannot be empty")

    if width % 16 != 0:
        errors.append(f"Width {width} must be multiple of 16")

    if height % 16 != 0:
        errors.append(f"Height {height} must be multiple of 16")

    if width * height > 4_000_000:
        errors.append("Total pixels cannot exceed 4MP")

    if errors:
        raise ValidationError(errors)

Generation Failures

Failures during polling:

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

Common failure reasons:

• content_policy_violation - Prompt/image flagged by safety
• generation_timeout - Took too long to generate
• internal_error - Server-side issue
• invalid_image - Input image couldn't be processed

Retry Strategy

import time
import random

class RetryableError(Exception):
    """Errors that can be retried."""
    pass

class NonRetryableError(Exception):
    """Errors that should not be retried."""
    pass

def classify_error(status_code, error_code):
    """Determine if error is retryable."""
    # Retryable
    if status_code in [429, 500, 502, 503]:
        return RetryableError

    # Non-retryable
    if status_code in [400, 401, 402, 403]:
        return NonRetryableError

    # Generation failures
    if error_code in ['generation_timeout', 'internal_error']:
        return RetryableError

    if error_code in ['content_policy_violation', 'invalid_image']:
        return NonRetryableError

    return RetryableError  # Default to retryable

def make_request_with_retry(func, max_retries=3):
    """Execute function with retry logic."""
    last_exception = None

    for attempt in range(max_retries):
        try:
            return func()
        except RetryableError as e:
            last_exception = e
            wait_time = (2 ** attempt) + random.uniform(0, 1)
            print(f"Attempt {attempt + 1} failed, retrying in {wait_time:.1f}s")
            time.sleep(wait_time)
        except NonRetryableError:
            raise  # Don't retry

    raise last_exception

Comprehensive Error Handler

import logging

class BFLError(Exception):
    """Base exception for BFL API errors."""
    def __init__(self, message, status_code=None, error_code=None):
        self.message = message
        self.status_code = status_code
        self.error_code = error_code
        super().__init__(message)

class AuthenticationError(BFLError):
    """API key or authentication issue."""
    pass

class InsufficientCreditsError(BFLError):
    """Account needs more credits."""
    pass

class RateLimitError(BFLError):
    """Too many concurrent requests."""
    def __init__(self, message, retry_after=5):
        super().__init__(message, 429, "rate_limit_exceeded")
        self.retry_after = retry_after

class ValidationError(BFLError):
    """Invalid request parameters."""
    pass

class GenerationError(BFLError):
    """Generation failed."""
    pass

def handle_response(response):
    """Process API response and raise appropriate errors."""
    if response.status_code == 200:
        return response.json()

    try:
        error_data = response.json()
    except:
        error_data = {"message": response.text}

    error_code = error_data.get("error", "unknown")
    message = error_data.get("message", "Unknown error")

    if response.status_code == 401:
        raise AuthenticationError(message, 401, error_code)

    if response.status_code == 402:
        raise InsufficientCreditsError(message, 402, error_code)

    if response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 5))
        raise RateLimitError(message, retry_after)

    if response.status_code == 400:
        raise ValidationError(message, 400, error_code)

    if response.status_code >= 500:
        raise BFLError(f"Server error: {message}", response.status_code, error_code)

    raise BFLError(message, response.status_code, error_code)

Logging Best Practices

import logging
import json

def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s'
    )

def log_request(endpoint, payload):
    logging.info(f"Request: POST {endpoint}")
    logging.debug(f"Payload: {json.dumps(payload, indent=2)}")

def log_error(error, context=None):
    logging.error(f"Error: {error}")
    if context:
        logging.error(f"Context: {context}")

def log_generation_failure(status, error, prompt):
    logging.warning(f"Generation failed: {error}")
    logging.debug(f"Failed prompt: {prompt[:100]}...")

Circuit Breaker Pattern

For production systems:

import time
from threading import Lock

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.last_failure_time = None
        self.state = "closed"  # closed, open, half-open
        self.lock = Lock()

    def record_success(self):
        with self.lock:
            self.failures = 0
            self.state = "closed"

    def record_failure(self):
        with self.lock:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "open"

    def can_proceed(self):
        with self.lock:
            if self.state == "closed":
                return True

            if self.state == "open":
                if time.time() - self.last_failure_time > self.reset_timeout:
                    self.state = "half-open"
                    return True
                return False

            # half-open: allow one request to test
            return True

Polling Patterns

BFL API uses asynchronous generation. All requests return a polling_url for status checking.

Basic Flow

1. POST request to model endpoint
   └─> Immediate response: { "polling_url": "..." }

2. GET polling_url (repeat until complete)
   └─> { "status": "Pending" | "Ready" | "Error" }

3. When "Ready", download result sample URL
   └─> URL expires in 10 minutes

Response States

Polling Strategies

Simple Fixed Interval

import time
import requests

def poll_fixed_interval(polling_url, headers, interval=2, timeout=120):
    """Simple polling with fixed interval."""
    start_time = time.time()

    while time.time() - start_time < timeout:
        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"))

        time.sleep(interval)

    raise TimeoutError("Polling timeout exceeded")

Exponential Backoff (Recommended)

import time
import random
import requests

def poll_with_backoff(polling_url, headers, max_attempts=30):
    """Polling with exponential backoff and jitter."""
    base_delay = 0.5  # Start with 500ms
    max_delay = 10    # Cap at 10 seconds

    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"))

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

    raise TimeoutError("Max polling attempts exceeded")

Adaptive Polling

import time
import requests

def poll_adaptive(polling_url, headers, timeout=120):
    """Adaptive polling that adjusts based on status."""
    start_time = time.time()
    delays = {
        "Pending": 2.0,      # Queue/processing
        None: 1.5            # Unknown/default
    }

    while time.time() - start_time < timeout:
        response = requests.get(polling_url, headers=headers)
        data = response.json()

        status = data.get("status")

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

        delay = delays.get(status, delays[None])
        time.sleep(delay)

    raise TimeoutError("Polling timeout exceeded")

Complete Example: Submit and Poll

import time
import requests

class BFLClient:
    def __init__(self, api_key, base_url="https://api.bfl.ai"):
        self.base_url = base_url
        self.headers = {
            "x-key": api_key,
            "Content-Type": "application/json"
        }

    def generate(self, model, prompt, **kwargs):
        """Submit generation request and poll for result."""
        # Submit request
        endpoint = f"{self.base_url}/v1/{model}"
        payload = {"prompt": prompt, **kwargs}

        response = requests.post(endpoint, headers=self.headers, json=payload)
        response.raise_for_status()

        polling_url = response.json()["polling_url"]

        # Poll for result
        return self._poll(polling_url)

    def _poll(self, polling_url, timeout=120):
        """Poll until completion or timeout."""
        start = time.time()
        delay = 1.0

        while time.time() - start < timeout:
            response = requests.get(polling_url, headers=self.headers)
            data = response.json()

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

            time.sleep(delay)
            delay = min(delay * 1.5, 5.0)  # Gradual backoff

        raise TimeoutError("Generation timed out")

# Usage
client = BFLClient("your-api-key")
result = client.generate(
    model="flux-2-pro",
    prompt="A beautiful sunset over mountains"
)
print(f"Image URL: {result['sample']}")

URL Expiration

Critical: Result URLs expire after 10 minutes. Always download immediately.

def download_result(result_url, output_path):
    """Download result image before URL expires."""
    response = requests.get(result_url)
    response.raise_for_status()

    with open(output_path, 'wb') as f:
        f.write(response.content)

    return output_path

Batch Processing with Polling

import asyncio
import aiohttp

async def generate_batch(client, prompts, model="flux-2-pro"):
    """Generate multiple images concurrently."""
    async with aiohttp.ClientSession() as session:
        # Submit all requests
        tasks = []
        for prompt in prompts:
            task = submit_and_poll(session, client, model, prompt)
            tasks.append(task)

        # Wait for all to complete
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

async def submit_and_poll(session, client, model, prompt):
    """Async submit and poll for single image."""
    # Submit
    async with session.post(
        f"{client.base_url}/v1/{model}",
        headers=client.headers,
        json={"prompt": prompt}
    ) as response:
        data = await response.json()
        polling_url = data["polling_url"]

    # Poll
    while True:
        async with session.get(polling_url, headers=client.headers) as response:
            data = await response.json()

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

        await asyncio.sleep(2)

Best Practices

1. Always implement timeouts - Never poll indefinitely 2. Use exponential backoff - Reduces server load, handles congestion 3. Add jitter - Prevents thundering herd when polling multiple requests 4. Handle all status values - Including unexpected ones 5. Download immediately - URLs expire in 10 minutes 6. Log polling attempts - Useful for debugging and monitoring 7. Respect rate limits - Implement proper backoff on 429 responses

Rate Limiting

BFL API enforces rate limits to ensure fair usage and system stability.

Current Limits

Concurrent requests means in-flight requests (submitted but not yet completed).

Rate Limit Headers

Check response headers for rate limit status:

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

HTTP 429 Response

When rate limited, you receive HTTP 429:

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

Handling Strategies

1. Client-Side Tracking

Track active requests to stay under limits:

from threading import Lock, 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:  # Blocks if at limit
            return self._make_request(model, prompt, **kwargs)

    def _make_request(self, model, prompt, **kwargs):
        # Submit request
        response = requests.post(...)
        polling_url = response.json()["polling_url"]

        # Poll until complete (request still "active")
        return self._poll(polling_url)

2. Retry with Exponential Backoff

import time

def request_with_retry(endpoint, payload, headers, max_retries=5):
    """Make request with automatic retry on rate limit."""
    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)  # Exponential backoff
            print(f"Rate limited. Waiting {wait_time}s...")
            time.sleep(wait_time)
            continue

        response.raise_for_status()
        return response

    raise Exception("Max retries exceeded due to rate limiting")

3. Queue-Based Architecture

For high-volume applications:

from queue import Queue
from threading import Thread
import time

class RequestQueue:
    def __init__(self, api_key, max_concurrent=24):
        self.api_key = api_key
        self.queue = Queue()
        self.active = 0
        self.max_concurrent = max_concurrent
        self.lock = Lock()

        # Start worker threads
        for _ in range(max_concurrent):
            worker = Thread(target=self._worker, daemon=True)
            worker.start()

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

    def _worker(self):
        """Process queue items."""
        while True:
            item = self.queue.get()
            try:
                result = self._process(item)
                item['callback'](result, None)
            except Exception as e:
                item['callback'](None, e)
            finally:
                self.queue.task_done()

    def _process(self, item):
        # Make request and poll
        ...

# Usage
queue = RequestQueue("your-api-key")

def handle_result(result, error):
    if error:
        print(f"Error: {error}")
    else:
        print(f"Generated: {result['sample']}")

queue.submit("flux-2-pro", "A sunset", handle_result)

4. Async with Semaphore

import asyncio
import aiohttp

class AsyncRateLimitedClient:
    def __init__(self, api_key, max_concurrent=24):
        self.api_key = api_key
        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:
                # Submit
                async with session.post(
                    f"https://api.bfl.ai/v1/{model}",
                    headers=self.headers,
                    json={"prompt": prompt}
                ) as response:
                    data = await response.json()
                    polling_url = data["polling_url"]

                # Poll until complete
                while True:
                    async with session.get(
                        polling_url,
                        headers=self.headers
                    ) as response:
                        data = await response.json()
                        if data["status"] == "Ready":
                            return data["result"]
                        elif data["status"] == "Error":
                            raise Exception(data.get("error"))
                    await asyncio.sleep(2)

# Usage
async def main():
    client = AsyncRateLimitedClient("your-api-key")

    # Generate 50 images with rate limiting
    prompts = [f"Image {i}" for i in range(50)]
    tasks = [client.generate("flux-2-pro", p) for p in prompts]
    results = await asyncio.gather(*tasks)

asyncio.run(main())

Monitoring Rate Limits

class RateLimitMonitor:
    def __init__(self):
        self.requests_made = 0
        self.rate_limit_hits = 0
        self.lock = Lock()

    def record_request(self, response):
        with self.lock:
            self.requests_made += 1
            if response.status_code == 429:
                self.rate_limit_hits += 1

    def get_stats(self):
        return {
            "total_requests": self.requests_made,
            "rate_limit_hits": self.rate_limit_hits,
            "hit_rate": self.rate_limit_hits / max(self.requests_made, 1)
        }

Best Practices

1. Track active requests - Know how many are in-flight 2. Implement client-side limits - Stay under limits proactively 3. Use semaphores - Clean way to limit concurrency 4. Queue for high volume - Buffer requests when traffic spikes 5. Monitor headers - React to remaining quota 6. Graceful degradation - Queue or delay when near limits 7. Different limits per endpoint - Remember Kontext Max is 6, not 24

Regional Distribution

For very high volume, consider distributing across regions:

ENDPOINTS = [
    "https://api.bfl.ai",
    "https://api.eu.bfl.ai",
    "https://api.us.bfl.ai"
]

def get_endpoint():
    """Round-robin or least-loaded selection."""
    return random.choice(ENDPOINTS)

Note: Verify regional rate limits are independent before relying on this strategy.

Webhook Integration

For production workloads, use webhooks instead of polling to receive generation results.

Benefits Over Polling

• Reduced API calls - No repeated polling requests
• Immediate notification - Know exactly when generation completes
• Better resource efficiency - No wasted compute on polling
• Scalable architecture - Event-driven design

Setup

Request with Webhook

Include webhook_url and optionally webhook_secret in your request:

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

Webhook Payload

When generation completes, BFL sends a POST request to your webhook URL:

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

For failures:

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

Security

Signature Verification

When webhook_secret is provided, BFL signs the payload with HMAC-SHA256:

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

Verification Implementation

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)

Flask Handler with Verification

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)

Express.js Handler

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' });
});

Requirements

HTTPS Required

Webhook URLs must use HTTPS in production. BFL will not send webhooks to HTTP endpoints.

Response Requirements

• Respond with 2xx status code to acknowledge receipt
• Respond within 30 seconds
• Keep handler fast - offload heavy processing

Retry Policy

BFL retries failed webhook deliveries:

After 3 failed attempts, the webhook is abandoned. Fall back to polling if critical.

Idempotency

Handle duplicate webhook deliveries:

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...

Hybrid Approach

Combine webhooks with polling fallback:

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)

Monitoring

Track webhook health:

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
        }