FLUX API Integration Guide

import requests
import time
import os

# Submit request to global endpoint
response = requests.post(
    'https://api.bfl.ai/v1/flux-2-pro-preview',
    headers={
        'accept': 'application/json',
        'x-key': os.environ.get("BFL_API_KEY"),
        'Content-Type': 'application/json',
    },
    json={
        'prompt': 'A serene landscape with mountains',
        'width': 1440,
        'height': 810
    }
)

data = response.json()
request_id = data['id']
polling_url = data['polling_url']  # Use this URL for polling

# Poll using the returned polling_url
while True:
    time.sleep(0.5)
    result = requests.get(
        polling_url,
        headers={
            'accept': 'application/json',
            'x-key': os.environ.get("BFL_API_KEY"),
        }
    ).json()
    
    if result['status'] == 'Ready':
        print(f"Image ready: {result['result']['sample']}")
        break
    elif result['status'] in ['Error', 'Failed']:
        print(f"Generation failed: {result}")
        break

import requests
import os
from datetime import datetime
from typing import Dict, Any

def download_and_store_image(result_url: str, local_path: str) -> str:
    """
    Download image from BFL delivery URL and store locally
    """
    response = requests.get(result_url)
    response.raise_for_status()
    
    with open(local_path, 'wb') as f:
        f.write(response.content)
    
    return local_path

def handle_generation_result(result: Dict[str, Any]) -> Dict[str, Any]:
    """
    Process generation result and store image locally
    """
    if result['status'] == 'Ready':
        sample_url = result['result']['sample']
        
        # Generate unique filename
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"generated_image_{timestamp}.jpg"
        local_path = os.path.join("./images", filename)
        
        # Ensure directory exists
        os.makedirs(os.path.dirname(local_path), exist_ok=True)
        
        # Download and store
        stored_path = download_and_store_image(sample_url, local_path)
        
        # Now serve from your own infrastructure
        return {
            'status': 'ready',
            'local_path': stored_path,
            'public_url': f"https://yourdomain.com/images/{filename}"
        }
    
    return result

import requests
import time
from typing import Dict, Any, Optional

def robust_api_call(url: str, headers: Dict[str, str], json_data: Dict[str, Any], max_retries: int = 3) -> Dict[str, Any]:
    """
    Robust API call with retry logic and proper error handling
    """
    for attempt in range(max_retries):
        try:
            response = requests.post(url, headers=headers, json=json_data)
            
            if response.status_code == 429:
                # Rate limit exceeded, wait and retry
                wait_time = 2 ** attempt  # Exponential backoff
                time.sleep(wait_time)
                continue
                
            elif response.status_code == 402:
                # Insufficient credits
                raise Exception("Insufficient credits. Please add credits to your account.")
                
            elif response.status_code >= 400:
                # Other client/server errors
                response.raise_for_status()
            
            return response.json()
            
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise e
            time.sleep(2 ** attempt)
    
    raise Exception(f"Failed after {max_retries} attempts")