FLUX 1.1 Pro API Python チュートリアル：5分以内で画像生成を始める

3つの重要数字： 推論レイテンシ約8〜15秒/枚、価格$0.04/枚（1024×1024）、Prompthero/ELO評価でFLUX 1.0 Proより最大20%品質向上

---

前提条件

このチュートリアルを実行するには以下が必要です。

アカウント・APIキー

• Black Forest Labs API または Replicate のアカウント
• BFL直接API経由の場合：https://api.us1.bfl.ai エンドポイントを使用
• Replicate経由の場合：black-forest-labs/flux-1.1-pro モデルID

Python環境

• Python 3.9以上
• 以下のパッケージをインストール：

# 必要パッケージのインストール
pip install requests python-dotenv Pillow

# Replicate経由を使う場合はこちらも追加
pip install replicate

環境変数の設定

プロジェクトルートに .env ファイルを作成：

# .env ファイル（Gitにコミットしないこと）
BFL_API_KEY=your_bfl_api_key_here
REPLICATE_API_TOKEN=your_replicate_token_here

.gitignore に .env を追加することを忘れずに。APIキーのハードコードは本番環境で絶対に避ける。

---

エンドポイントと認証の設定

FLUX 1.1 ProはBlack Forest Labs（BFL）の直接APIとReplicateの両方から利用できる。本チュートリアルではBFL直接APIをメインに扱い、Replicate経由のコードも補足する。

BFL APIは非同期ポーリング方式を採用している。リクエストを送るとrequest_idが返り、その後別のエンドポイントで結果をポーリングする。これはタイムアウトを防ぐための設計上の選択であり、長時間実行ジョブに適している。

# auth_check.py — 認証とエンドポイント接続を確認するスクリプト
import os
import requests
from dotenv import load_dotenv

# .envファイルから環境変数を読み込む
load_dotenv()

API_KEY = os.getenv("BFL_API_KEY")
BASE_URL = "https://api.us1.bfl.ai/v1"

def check_auth() -> bool:
    """
    APIキーが有効かどうかを確認する。
    実際のリクエストを送らずに接続を検証するための関数。
    """
    if not API_KEY:
        raise ValueError("BFL_API_KEY が環境変数に設定されていません")
    
    # 認証ヘッダーの構成
    # BFL APIはBearer方式ではなく x-key ヘッダーを使用する点に注意
    headers = {
        "x-key": API_KEY,
        "Content-Type": "application/json"
    }
    
    # テスト用に最小パラメータでリクエストを試みる
    test_payload = {
        "prompt": "test",
        "width": 256,
        "height": 256,
        "steps": 1  # ステップ数を最小にしてコストを抑える
    }
    
    response = requests.post(
        f"{BASE_URL}/flux-pro-1.1",
        headers=headers,
        json=test_payload
    )
    
    if response.status_code == 200:
        print("✓ 認証成功")
        print(f"  Request ID: {response.json().get('id')}")
        return True
    elif response.status_code == 401:
        print("✗ 認証失敗：APIキーを確認してください")
        return False
    else:
        print(f"✗ 予期しないエラー：HTTP {response.status_code}")
        print(f"  レスポンス：{response.text}")
        return False

if __name__ == "__main__":
    check_auth()

---

基本実装：最初の画像を生成する

以下のコードは動作する最小実装である。コピーしてそのまま実行できる。

# generate_basic.py — FLUX 1.1 Pro で画像を生成する基本実装
import os
import time
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("BFL_API_KEY")
BASE_URL = "https://api.us1.bfl.ai/v1"

def generate_image(
    prompt: str,
    width: int = 1024,
    height: int = 1024,
    steps: int = 25,
    guidance: float = 3.5,
    output_path: str = "output.jpg"
) -> str:
    """
    FLUX 1.1 Pro API で画像を生成してローカルに保存する。
    
    Returns:
        保存した画像ファイルのパス
    """
    headers = {
        "x-key": API_KEY,
        "Content-Type": "application/json"
    }
    
    # ステップ1：生成リクエストを送信して request_id を取得
    payload = {
        "prompt": prompt,
        "width": width,
        "height": height,
        "steps": steps,
        "guidance": guidance,
        # safety_tolerance: 0=最も制限、6=最も緩い。デフォルト2が推奨
        "safety_tolerance": 2,
        "output_format": "jpeg"  # "jpeg" または "png"
    }
    
    print(f"生成リクエスト送信中...")
    response = requests.post(
        f"{BASE_URL}/flux-pro-1.1",
        headers=headers,
        json=payload,
        timeout=30  # ネットワーク接続タイムアウト（生成完了とは別）
    )
    response.raise_for_status()
    
    request_id = response.json()["id"]
    print(f"Request ID: {request_id}")
    
    # ステップ2：結果をポーリングする
    # BFL APIは非同期方式のため、完了まで繰り返し確認が必要
    result_url = f"{BASE_URL}/get_result"
    
    for attempt in range(60):  # 最大60回 = 約120秒待機
        time.sleep(2)  # 2秒間隔でポーリング（過剰なリクエストを避ける）
        
        result_response = requests.get(
            result_url,
            headers=headers,
            params={"id": request_id},
            timeout=10
        )
        result_response.raise_for_status()
        result_data = result_response.json()
        
        status = result_data.get("status")
        print(f"  状態確認 [{attempt + 1}/60]: {status}")
        
        if status == "Ready":
            # 画像URLが返ってくるので、バイナリとしてダウンロード
            image_url = result_data["result"]["sample"]
            image_response = requests.get(image_url, timeout=30)
            
            with open(output_path, "wb") as f:
                f.write(image_response.content)
            
            print(f"✓ 画像を保存しました: {output_path}")
            return output_path
        
        elif status == "Error":
            raise RuntimeError(f"生成エラー: {result_data}")
        
        # "Pending" または "Processing" の場合はループを継続
    
    raise TimeoutError(f"120秒以内に生成が完了しませんでした (request_id: {request_id})")


if __name__ == "__main__":
    result = generate_image(
        prompt="A photorealistic mountain landscape at golden hour, "
               "sharp focus, 8K, cinematic lighting",
        width=1024,
        height=1024,
        output_path="mountain_test.jpg"
    )
    print(f"完了: {result}")

このコードを実行すると、通常8〜15秒で mountain_test.jpg がカレントディレクトリに保存される。

---

APIパラメータ リファレンス

パラメータ名	型	デフォルト	有効範囲	影響する要素
prompt	string	必須	最大10,000文字	生成される画像の内容・スタイル全般
width	integer	1024	256〜1440（64の倍数）	出力解像度の横幅、コストに影響
height	integer	1024	256〜1440（64の倍数）	出力解像度の縦幅、コストに影響
steps	integer	25	1〜50	生成品質と推論時間のトレードオフ
guidance	float	3.5	2.0〜5.0	promptへの忠実度。高いほど指示に従うが多様性が下がる
safety_tolerance	integer	2	0〜6	コンテンツフィルタの厳格度
seed	integer	null	0〜2^32-1	同一seedで再現可能な出力を得る
output_format	string	”jpeg"	"jpeg”, “png”	出力ファイル形式。pngは透過対応だがファイルサイズ大
prompt_upsampling	boolean	false	true/false	BFL側でpromptを自動強化する。品質向上するが一貫性が下がる

注意： width × height が大きくなるほどレイテンシとコストが増加する。1024×1024を基準とした場合、1440×1440では約2倍の推論時間が必要になる。

---

本番向け実装

基本実装をそのまま本番に使うのは危険である。以下のコードはリトライロジック、ログ、バッチ処理、エラーハンドリングを追加した本番対応版である。

# flux_client.py — 本番環境対応の FLUX 1.1 Pro クライアント
import os
import time
import logging
import requests
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
from dotenv import load_dotenv

load_dotenv()

# structuredログを設定（本番ではJSONフォーマットに変更推奨）
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)


@dataclass
class GenerationConfig:
    """
    生成パラメータをデータクラスで管理する。
    dictより型安全で、IDEの補完が効く。
    """
    prompt: str
    width: int = 1024
    height: int = 1024
    steps: int = 25
    guidance: float = 3.5
    safety_tolerance: int = 2
    seed: Optional[int] = None
    output_format: str = "jpeg"
    prompt_upsampling: bool = False


class FluxProClient:
    """
    FLUX 1.1 Pro API クライアント。
    コネクション管理・リトライ・エラーハンドリングを内包する。
    """
    
    BASE_URL = "https://api.us1.bfl.ai/v1"
    
    # リトライ対象のHTTPステータスコード
    # 429（レート制限）と5xx（サーバーエラー）のみリトライする
    RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
    
    def __init__(
        self,
        api_key: Optional[str] = None,
        max_retries: int = 3,
        poll_interval: float = 2.0,
        max_poll_attempts: int = 60
    ):
        self.api_key = api_key or os.getenv("BFL_API_KEY")
        if not self.api_key:
            raise ValueError("BFL_API_KEY が設定されていません")
        
        self.max_retries = max_retries
        self.poll_interval = poll_interval
        self.max_poll_attempts = max_poll_attempts
        
        # Sessionを再利用してTCPコネクションのオーバーヘッドを削減
        self.session = requests.Session()
        self.session.headers.update({
            "x-key": self.api_key,
            "Content-Type": "application/json"
        })
    
    def _build_payload(self, config: GenerationConfig) -> dict:
        """GenerationConfigをAPIペイロード dictに変換する。"""
        payload = {
            "prompt": config.prompt,
            "width": config.width,
            "height": config.height,
            "steps": config.steps,
            "guidance": config.guidance,
            "safety_tolerance": config.safety_tolerance,
            "output_format": config.output_format,
            "prompt_upsampling": config.prompt_upsampling
        }
        # seedはNoneの場合はペイロードから除外する（APIがnullを許可しないため）
        if config.seed is not None:
            payload["seed"] = config.seed
        return payload
    
    def _submit_request(self, config: GenerationConfig) -> str:
        """
        生成リクエストを送信してrequest_idを返す。
        リトライロジックを含む。
        """
        payload = self._build_payload(config)
        
        for attempt in range(self.max_retries):
            try:
                response = self.session.post(
                    f"{self.BASE_URL}/flux-pro-1.1",
                    json=payload,
                    timeout=30
                )
                
                if response.status_code == 200:
                    request_id = response.json()["id"]
                    logger.info(f"リクエスト送信成功: {request_id}")
                    return request_id
                
                elif response.status_code in self.RETRYABLE_STATUS_CODES:
                    wait_time = 2 ** attempt  # 指数バックオフ: 1s, 2s, 4s
                    logger.warning(
                        f"HTTP {response.status_code} — "
                        f"{wait_time}秒後にリトライ ({attempt + 1}/{self.max_retries})"
                    )
                    time.sleep(wait_time)
                    continue
                
                elif response.status_code == 422:
                    # バリデーションエラーはリトライしない
                    raise ValueError(
                        f"パラメータエラー (HTTP 422): {response.json()}"
                    )
                
                elif response.status_code == 402:
                    raise RuntimeError(
                        "クレジット不足 (HTTP 402): BFLアカウントに残高を追加してください"
                    )
                
                else:
                    response.raise_for_status()
            
            except requests.exceptions.Timeout:
                logger.warning(f"接続タイムアウト (試行 {attempt + 1}/{self.max_retries})")
                if attempt == self.max_retries - 1:
                    raise
        
        raise RuntimeError(f"{self.max_retries}回のリトライ後も失敗しました")
    
    def _poll_result(self, request_id: str) -> dict:
        """
        結果が返るまでポーリングする。
        Readyステータスを受け取ったら結果dictを返す。
        """
        for attempt in range(self.max_poll_attempts):
            time.sleep(self.poll_interval)
            
            response = self.session.get(
                f"{self.BASE_URL}/get_result",
                params={"id": request_id},
                timeout=10
            )
            response.raise_for_status()
            data = response.json()
            status = data.get("status")
            
            logger.debug(f"ポーリング [{attempt + 1}/{self.max_poll_attempts}]: {status}")
            
            if status == "Ready":
                return data["result"]
            
            elif status == "Error":
                error_detail = data.get("error", "詳細不明")
                raise RuntimeError(f"生成エラー: {error_detail}")
            
            # "Pending" / "Processing" はループを継続
        
        timeout_seconds = self.poll_interval * self.max_poll_attempts
        raise TimeoutError(
            f"{timeout_seconds}秒以内に完了しませんでした (id: {request_id})"
        )
    
    def generate(
        self,
        config: GenerationConfig,
        output_path: str = "output.jpg"
    ) -> dict:
        """
        画像を生成してファイルに保存する。
        成功時は画像URLとローカルパスを含むdictを返す。
        """
        start_time = time.time()
        
        # リクエスト送信
        request_id = self._submit_request(config)
        
        # 結果待機
        result = self._poll_result(request_id)
        
        # 画像ダウンロードと保存
        image_url = result["sample"]
        output_file = Path(output_path)
        output_file.parent.mkdir(parents=True, exist_ok=True)
        
        image_response = requests.get(image_url, timeout=60)
        image_response.raise_for_status()
        
        output_file.write_bytes(image_response.content)
        
        elapsed = time.time() - start_time
        logger.info(f"完了: {output_path} ({elapsed:.1f}秒)")
        
        return {
            "request_id": request_id,
            "image_url": image_url,
            "local_path": str(output_file),
            "elapsed_seconds": elapsed
        }
    
    def __del__(self):
        """オブジェクト破棄時にSessionを閉じる。"""
        if hasattr(self, "session"):
            self.session.close()


# 使用例
if __name__ == "__main__":
    client = FluxProClient(max_retries=3)
    
    config = GenerationConfig(
        prompt="Professional product photography of a ceramic coffee mug, "
               "white background, studio lighting, sharp focus, commercial quality",
        width=1024,
        height=1024,
        steps=28,
        guidance=3.5,
        seed=42,          # 再現性のためにseedを固定
        output_format="jpeg"
    )
    
    result = client.generate(config, output_path="outputs/product_shot.jpg")
    print(f"画像URL: {result['image_url']}")
    print(f"生成時間: {result['elapsed_seconds']:.1f}秒")

---

エラーハンドリング：実際のエラーコードと対処法

エラーコード	発生原因	対処法
HTTP 401	APIキーが無効または未設定	.env の BFL_API_KEY を確認。スペースや改行が混入していないか確認
HTTP 402	アカウントのクレジット残高不足	BFLダッシュボードでクレジットを追加
HTTP 422	パラメータバリデーション失敗	width/height が64の倍数か確認。steps が1〜50の範囲か確認
HTTP 429	レート制限超過	指数バックオフでリトライ。無料tierは1分あたり約10リクエスト
HTTP 500/503	BFLサーバーエラー	2〜8秒の指数バックオフでリトライ。継続する場合はBFLステータスページを確認
status: “Error”	生成失敗（promptフィルタ等）	safety_tolerance を確認。不適切なコンテンツがpromptに含まれていないか確認
TimeoutError	ポーリング上限到達	max_poll_attempts を増やすか、サーバー負荷が高い時間帯を避ける
ConnectionError	ネットワーク断	VPN/Proxyの確認。エンドポイント api.us1.bfl.ai への疎通確認

セーフガードとして追加すべきチェック：

# validation.py — リクエスト送信前のパラメータ検証
def validate_config(config: GenerationConfig) -> None:
    """
    APIに送る前にローカルでバリデーションする。
    無効なリクエストでAPIクレジットを消費しないようにするため。
    """
    errors = []
    
    # widthとheightは64の倍数である必要がある
    if config.width % 64 != 0:
        errors.append(f"width ({config.width}) は64の倍数である必要があります")
    if config.height % 64 != 0:
        errors.append(f"height ({config.height}) は64の倍数である必要があります")
    
    # 解像度の範囲チェック
    if not (256 <= config.width <= 1440):
        errors.append(f"width は256〜1440の範囲内である必要があります")
    if not (256 <= config.height <= 1440):
        errors.append(f"height は256〜1440の範囲内である必要があります")
    
    # stepsの範囲チェック
    if not (1 <= config.steps <= 50):
        errors.append(f"steps は1〜50の範囲内である必要があります")
    
    # promptが空でないことを確認
    if not config.prompt or len(config.prompt.strip()) == 0:
        errors.append("prompt は空にできません")
    
    if errors:
        raise ValueError(f"バリデーションエラー:\n" + "\n".join(f"  - {e}" for e in errors))

---

パフォーマンスとコストの実測値

以下の数値はBFL公式ドキュメントおよびコミュニティ計測値に基づく。

解像度	おおよその生成時間	コスト（BFL直接）	コスト（Replicate経由）	推奨用途
512×512	4〜6秒	$0.01/枚	$0.02〜0.03/枚	プロトタイプ・テスト
768×768	6〜9秒	$0.02/枚	$0.03〜0.04/枚	SNS・ブログ用途
1024×1024	8〜15秒	$0.04/枚	$0.055/枚	標準的な本番用途
1024×1440	12〜20秒	$0.06/枚	$0.08/枚	ポートレート・印刷
1440×1440	18〜30秒	$0.08/枚	$0.10/枚	高解像度・商業印刷

バッチ処理のコスト試算（1024×1024の場合）：

月間生成枚数	BFL直接コスト	Replicate経由コスト
100枚	$4.00	$5.50
1,000枚	$40.00	$55.00
10,000枚	$400.00	$550.00
100,000枚	$4,000.00	$5,500.00（ボリューム割引要確認）

BFL直接API vs Replicate経由の選択基準：

• BFL直接API：コストを最小化したい、低レイテンシが必要、大量バッチ処理を行う場合
• Replicate経由：既存のReplicate基盤を使い回したい、モデル切り替えの柔軟性が必要、小規模プロトタイプの場合

FLUX 1.1 Pro を使うべきでないケース：

• リアルタイム（1秒未満）のレスポンスが必要なユースケース — レイテンシ上8〜15秒は避けられない
• 月間100万枚以上のスケールで$0.04/枚が予算に収まらない場合 — Stable Diffusion XLのセルフホストを検討
• 厳密な再現性が必要で同一promptで完全一致した出力を要求する場合 — seedを固定しても微妙な差異が発生することがある

---

Replicate経由の実装（補足）

Replicateを既に使用している場合の最小実装：

# generate_via_replicate.py — Replicate経由でFLUX 1.1 Proを使う
import os
import replicate
from dotenv import load_dotenv

load_dotenv()

# REPLICATE_API_TOKEN は replicate ライブラリが自動で読み込む
os.environ["REPLICATE_API_TOKEN"] = os.getenv("REPLICATE_API_TOKEN", "")

def generate_with_replicate(prompt: str, output_path: str = "output_replicate.webp") -> str:
    """
    Replicate経由でFLUX 1.1 Proを実行する。
    BFLの直接APIと比べてセットアップが少ない反面、コストが約38%高い。
    """
    output = replicate.run(
        # モデルIDのバージョンは変わることがあるので最新IDを確認すること
        "black-forest-labs/flux-1.1-pro",
        input={
            "prompt": prompt,
            "width": 1024,
            "height": 1024,
            "steps": 25,
            "guidance": 3.5,
            "output_format": "webp",  # Replicateはwebpも対応
            "output_quality": 90,     # webp品質（1〜100）
            "safety_tolerance": 2,
            "prompt_upsampling": False
        }
    )
    
    # replicateのrunはイテレータを返す場合がある
    # 出力をバイナリとして保存する
    import requests
    if isinstance(output, list):
        image_url = output[0]
    else:
        image_url = str(output)
    
    response = requests.get(image_url, timeout=60)
    with open(output_path, "wb") as f:
        f.write(response.content)
    
    print(f"✓ 保存完了: {output_path}")
    return output_path


if __name__ == "__main__":
    generate_with_replicate(
        prompt="Aerial view of a dense forest in autumn, "
               "drone photography, golden and red foliage, misty morning",
        output_path="forest_replicate.webp"
    )

---

まとめ

FLUX 1.1 Pro API はBFL直接エンドポイント経由で$0.04/枚・8〜15秒の推論時間で動作し、非同期ポーリング方式のためBFL Client + 指数バックオフのリトライが本番実装の最低要件になる。月間1,000枚未満の小規模用途ならReplicateで十分だが、それ以上のスケールではBFL直接APIへの移行がコスト面で合理的な選択となる。本記事のコードはそのままコピーして使える状態にしてあるが、safety_tolerance とシード管理はプロダクトの要件に合わせて必ず調整すること。

メモ： 複数の AI モデルを一つのパイプラインで使う場合、AtlasCloud は Kling、Flux、Seedance、Claude、GPT など 300+ モデルへの統一 API アクセスを提供します。API キー一つで全モデル対応。新規ユーザーは初回チャージで 25% ボーナス（最大 $100）。