FLUX 1.1 Pro API Python Tutorial: 5분 안에 이미지 생성하기

핵심 수치 3가지: 평균 생성 시간 ~10초, 이미지 1장당 비용 $0.04, Blind evaluation 기준 품질 점수 FLUX 1.0 Pro 대비 +6% 향상

---

이 튜토리얼은 FLUX 1.1 Pro API를 Python으로 연동해서 실제로 이미지를 생성하는 코드를 다룬다. 환경 설정부터 production-ready 코드까지, 복사해서 바로 쓸 수 있는 구현체를 제공한다. Replicate와 Black Forest Labs의 공식 API 두 가지 경로를 모두 커버한다.

---

Prerequisites

필요한 계정

서비스	용도	무료 크레딧
Black Forest Labs API	공식 FLUX API (직접 접근)	없음 (유료)
Replicate	FLUX 1.1 Pro 호스팅	가입 시 $5
Together AI	대안 호스팅	가입 시 크레딧 제공

이 튜토리얼은 Black Forest Labs 공식 API를 기준으로 진행한다. Replicate 섹션도 별도로 포함했다.

Python 환경 요구사항

• Python 3.8 이상
• pip 또는 uv 패키지 매니저

라이브러리 설치

# 필수 패키지 설치
pip install requests python-dotenv Pillow

# 또는 uv를 사용하는 경우 (더 빠름)
uv pip install requests python-dotenv Pillow

# Replicate를 사용하는 경우 추가 설치
pip install replicate

환경 변수 설정

# .env 파일 생성
touch .env

.env 파일 내용:

BFL_API_KEY=your_black_forest_labs_api_key_here
REPLICATE_API_TOKEN=your_replicate_token_here

API 키는 절대 코드에 하드코딩하지 않는다. .env를 .gitignore에 추가하는 것을 잊지 말것.

echo ".env" >> .gitignore

---

프로젝트 구조

flux-image-gen/
├── .env
├── .gitignore
├── requirements.txt
├── basic_generation.py
├── production_client.py
└── output/

---

1단계: 기본 구현 — API 작동 확인

FLUX 1.1 Pro API는 비동기 방식으로 동작한다. 요청을 보내면 즉시 결과를 받는 게 아니라 request_id를 반환하고, 그 ID로 완료 여부를 polling해야 한다.

# basic_generation.py
# FLUX 1.1 Pro API 기본 호출 예제
# BFL(Black Forest Labs) 공식 API 사용

import requests
import time
import os
from dotenv import load_dotenv

# .env에서 환경 변수 로드
load_dotenv()

API_KEY = os.getenv("BFL_API_KEY")

# BFL 공식 API 엔드포인트
# flux-pro-1.1은 FLUX 1.1 Pro 모델 식별자
GENERATE_URL = "https://api.us1.bfl.ai/v1/flux-pro-1.1"
RESULT_URL = "https://api.us1.bfl.ai/v1/get_result"

def generate_image(prompt: str) -> str:
    """
    이미지 생성 요청을 보내고 완성된 이미지 URL을 반환한다.
    
    Args:
        prompt: 생성할 이미지를 설명하는 텍스트
    
    Returns:
        생성된 이미지의 URL (10분 후 만료됨)
    """
    
    headers = {
        "x-key": API_KEY,           # BFL API 인증 헤더
        "Content-Type": "application/json"
    }
    
    payload = {
        "prompt": prompt,
        "width": 1024,              # 픽셀 단위, 256~1440 범위
        "height": 768,              # 픽셀 단위, 256~1440 범위
    }
    
    # 1단계: 생성 요청 전송
    response = requests.post(GENERATE_URL, json=payload, headers=headers)
    
    # 요청 실패 시 즉시 예외 발생
    response.raise_for_status()
    
    request_id = response.json()["id"]
    print(f"Request ID: {request_id}")
    
    # 2단계: 결과 polling
    # API가 비동기로 동작하므로 완료될 때까지 반복 확인
    while True:
        time.sleep(2)  # 2초 간격으로 polling — 너무 빠르면 rate limit 걸림
        
        result_response = requests.get(
            RESULT_URL,
            headers={"x-key": API_KEY},
            params={"id": request_id}
        )
        result_response.raise_for_status()
        
        result = result_response.json()
        status = result.get("status")
        
        print(f"Status: {status}")
        
        if status == "Ready":
            # 결과 URL은 10분 후 만료됨 — 로컬에 저장 권장
            return result["result"]["sample"]
        
        elif status == "Error":
            raise RuntimeError(f"Image generation failed: {result}")
        
        # "Pending" 또는 "Processing" 상태면 계속 대기

if __name__ == "__main__":
    image_url = generate_image(
        "A photorealistic mountain landscape at golden hour, "
        "sharp focus, cinematic lighting"
    )
    print(f"Image URL: {image_url}")

이 코드를 실행하면 약 8~15초 후 이미지 URL이 출력된다. URL은 10분 후 만료되므로 바로 다운로드해야 한다.

---

2단계: 이미지 다운로드 및 로컬 저장

URL만 받아서는 production에서 쓸 수 없다. 이미지를 로컬에 저장하거나 S3 같은 스토리지로 올려야 한다.

# image_downloader.py
# 생성된 이미지를 로컬 파일로 저장

import requests
import os
from pathlib import Path
from datetime import datetime
from PIL import Image
from io import BytesIO

def download_and_save_image(
    image_url: str,
    output_dir: str = "output",
    filename: str = None
) -> str:
    """
    이미지 URL에서 파일을 다운로드하고 로컬에 저장한다.
    
    Args:
        image_url: 다운로드할 이미지 URL
        output_dir: 저장 디렉토리 경로
        filename: 파일명 (None이면 타임스탬프로 자동 생성)
    
    Returns:
        저장된 파일의 절대 경로
    """
    
    # 출력 디렉토리 생성 (없으면)
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    
    # 파일명 자동 생성 — 같은 이름으로 덮어쓰기 방지
    if filename is None:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"flux_1_1_pro_{timestamp}.png"
    
    output_path = os.path.join(output_dir, filename)
    
    # 이미지 다운로드
    response = requests.get(image_url, timeout=30)
    response.raise_for_status()
    
    # Pillow로 열어서 저장 — 포맷 검증 및 변환에 유용
    image = Image.open(BytesIO(response.content))
    image.save(output_path, format="PNG")
    
    # 실제 저장된 이미지 크기 확인
    print(f"Saved: {output_path} ({image.width}x{image.height}px)")
    
    return os.path.abspath(output_path)

---

3단계: API 파라미터 전체 참조표

FLUX 1.1 Pro API에서 사용 가능한 파라미터 전체 목록이다.

파라미터	타입	기본값	유효 범위	영향
prompt	string	필수	최대 10,000자	이미지 내용 전반
width	integer	1024	256~1440, 32의 배수	출력 이미지 가로 픽셀
height	integer	768	256~1440, 32의 배수	출력 이미지 세로 픽셀
prompt_upsampling	boolean	false	true/false	true 설정 시 모델이 프롬프트를 내부적으로 강화 — 디테일 향상
seed	integer	null (랜덤)	0~2147483647	재현 가능한 결과 생성
safety_tolerance	integer	2	0~6	낮을수록 엄격한 콘텐츠 필터링
output_format	string	"jpeg"	"jpeg", "png"	출력 파일 포맷

width/height 주의사항: 두 값의 곱이 너무 크면 (> 1440 * 1440) 오류가 발생한다. 1024x1024가 가장 안정적인 설정이다.

prompt_upsampling: 짧고 단순한 프롬프트에 효과적이다. 상세한 프롬프트를 이미 작성했다면 오히려 의도치 않은 변형이 생길 수 있다.

---

4단계: Production-Ready 클라이언트

기본 예제를 production 환경에서 쓰려면 retry 로직, 타임아웃, 로깅이 필요하다.

# production_client.py
# Production 환경에서 사용하기 위한 FLUX 1.1 Pro 클라이언트
# retry, timeout, 에러 처리, 이미지 저장 포함

import requests
import time
import os
import logging
from pathlib import Path
from datetime import datetime
from io import BytesIO
from typing import Optional
from dataclasses import dataclass
from dotenv import load_dotenv
from PIL import Image

load_dotenv()

# 로깅 설정 — print 대신 logging 사용 (레벨 제어 가능)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)


@dataclass
class GenerationConfig:
    """이미지 생성 설정을 담는 데이터 클래스"""
    prompt: str
    width: int = 1024
    height: int = 1024
    prompt_upsampling: bool = False
    seed: Optional[int] = None
    safety_tolerance: int = 2
    output_format: str = "png"


class FluxProClient:
    """
    FLUX 1.1 Pro API 클라이언트
    
    - 자동 재시도 (최대 3회)
    - polling 타임아웃 (기본 120초)
    - 이미지 자동 다운로드 및 저장
    """
    
    BASE_URL = "https://api.us1.bfl.ai/v1"
    MODEL = "flux-pro-1.1"
    
    # polling 관련 상수
    POLL_INTERVAL = 2       # 초 단위, 너무 짧으면 rate limit
    MAX_WAIT_TIME = 120     # 초 단위, 2분 초과시 타임아웃
    MAX_RETRIES = 3
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.getenv("BFL_API_KEY")
        
        if not self.api_key:
            raise ValueError(
                "API key not found. Set BFL_API_KEY in environment or pass api_key parameter."
            )
        
        # Session 재사용으로 TCP connection overhead 줄임
        self.session = requests.Session()
        self.session.headers.update({
            "x-key": self.api_key,
            "Content-Type": "application/json"
        })
    
    def generate(
        self,
        config: GenerationConfig,
        output_dir: str = "output"
    ) -> str:
        """
        이미지를 생성하고 로컬에 저장한다.
        
        Args:
            config: GenerationConfig 인스턴스
            output_dir: 이미지 저장 디렉토리
        
        Returns:
            저장된 이미지 파일의 절대 경로
        
        Raises:
            RuntimeError: 생성 실패 또는 타임아웃
            requests.HTTPError: API 요청 실패
        """
        
        request_id = self._submit_request(config)
        image_url = self._poll_for_result(request_id)
        saved_path = self._download_image(image_url, output_dir, config.output_format)
        
        return saved_path
    
    def _submit_request(self, config: GenerationConfig) -> str:
        """생성 요청을 제출하고 request_id를 반환한다."""
        
        payload = {
            "prompt": config.prompt,
            "width": config.width,
            "height": config.height,
            "prompt_upsampling": config.prompt_upsampling,
            "safety_tolerance": config.safety_tolerance,
            "output_format": config.output_format,
        }
        
        # seed가 명시된 경우에만 포함 — null 전송 시 동작이 다를 수 있음
        if config.seed is not None:
            payload["seed"] = config.seed
        
        url = f"{self.BASE_URL}/{self.MODEL}"
        
        for attempt in range(1, self.MAX_RETRIES + 1):
            try:
                response = self.session.post(url, json=payload, timeout=30)
                response.raise_for_status()
                request_id = response.json()["id"]
                logger.info(f"Request submitted (attempt {attempt}): {request_id}")
                return request_id
            
            except requests.HTTPError as e:
                status_code = e.response.status_code
                
                # 429: Rate limit — 재시도 가능
                if status_code == 429 and attempt < self.MAX_RETRIES:
                    wait = 2 ** attempt  # exponential backoff: 2s, 4s, 8s
                    logger.warning(f"Rate limited. Retrying in {wait}s...")
                    time.sleep(wait)
                    continue
                
                # 그 외 HTTP 에러는 재시도 없이 바로 raise
                raise
        
        raise RuntimeError("Max retries exceeded on request submission.")
    
    def _poll_for_result(self, request_id: str) -> str:
        """결과가 Ready 상태가 될 때까지 polling하고 이미지 URL을 반환한다."""
        
        result_url = f"{self.BASE_URL}/get_result"
        elapsed = 0
        
        while elapsed < self.MAX_WAIT_TIME:
            time.sleep(self.POLL_INTERVAL)
            elapsed += self.POLL_INTERVAL
            
            response = self.session.get(
                result_url,
                params={"id": request_id},
                timeout=10
            )
            response.raise_for_status()
            
            data = response.json()
            status = data.get("status")
            
            logger.info(f"[{elapsed}s] Status: {status}")
            
            if status == "Ready":
                return data["result"]["sample"]
            
            elif status == "Error":
                error_detail = data.get("result", "No detail provided")
                raise RuntimeError(f"Generation error: {error_detail}")
            
            # "Pending" 또는 "Processing" → 계속 대기
        
        raise RuntimeError(
            f"Timeout: Image not ready after {self.MAX_WAIT_TIME} seconds. "
            f"Request ID: {request_id}"
        )
    
    def _download_image(self, url: str, output_dir: str, fmt: str) -> str:
        """이미지 URL에서 파일을 다운로드하고 로컬에 저장한다."""
        
        Path(output_dir).mkdir(parents=True, exist_ok=True)
        
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"flux_1_1_pro_{timestamp}.{fmt}"
        output_path = os.path.join(output_dir, filename)
        
        response = self.session.get(url, timeout=60)
        response.raise_for_status()
        
        image = Image.open(BytesIO(response.content))
        image.save(output_path)
        
        logger.info(f"Saved: {output_path} ({image.width}x{image.height})")
        return os.path.abspath(output_path)
    
    def __del__(self):
        # 객체 소멸 시 Session 명시적 종료
        if hasattr(self, "session"):
            self.session.close()


# 실행 예시
if __name__ == "__main__":
    client = FluxProClient()
    
    config = GenerationConfig(
        prompt=(
            "A highly detailed aerial view of a futuristic city at night, "
            "neon lights reflected on wet streets, 8K resolution, "
            "photorealistic, cinematic composition"
        ),
        width=1024,
        height=1024,
        prompt_upsampling=True,
        seed=42,                    # 동일한 결과 재현을 위한 seed 고정
        safety_tolerance=2,
        output_format="png"
    )
    
    saved_path = client.generate(config, output_dir="output")
    print(f"Done: {saved_path}")

---

5단계: Replicate를 통한 FLUX 1.1 Pro 연동

Black Forest Labs 직접 API 대신 Replicate를 쓰고 싶다면 아래 코드를 사용한다. Replicate는 무료 크레딧이 있어서 테스트 비용 없이 시작하기 좋다.

# replicate_flux.py
# Replicate API를 통한 FLUX 1.1 Pro 이미지 생성
# replicate 라이브러리: pip install replicate

import replicate
import requests
import os
from pathlib import Path
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()

# REPLICATE_API_TOKEN은 replicate 라이브러리가 환경 변수에서 자동으로 읽음
# 별도로 세팅할 필요 없음

def generate_with_replicate(
    prompt: str,
    width: int = 1024,
    height: int = 1024,
    output_dir: str = "output"
) -> str:
    """
    Replicate API를 통해 FLUX 1.1 Pro로 이미지를 생성한다.
    
    Replicate는 BFL 직접 API와 달리 동기 방식으로 결과를 반환한다.
    내부적으로는 polling을 처리해주므로 코드가 단순해진다.
    """
    
    # Replicate의 FLUX 1.1 Pro 모델 식별자
    MODEL_ID = "black-forest-labs/flux-1.1-pro"
    
    output = replicate.run(
        MODEL_ID,
        input={
            "prompt": prompt,
            "width": width,
            "height": height,
            "output_format": "png",
            "output_quality": 100,   # 1~100, Replicate 전용 파라미터
            "safety_tolerance": 2,
            "prompt_upsampling": False,
        }
    )
    
    # Replicate는 URL 리스트를 반환 (단일 이미지도 리스트)
    image_url = str(output[0]) if isinstance(output, list) else str(output)
    
    # 로컬 저장
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_path = os.path.join(output_dir, f"replicate_flux_{timestamp}.png")
    
    response = requests.get(image_url, timeout=60)
    response.raise_for_status()
    
    with open(output_path, "wb") as f:
        f.write(response.content)
    
    print(f"Saved: {output_path}")
    return os.path.abspath(output_path)


if __name__ == "__main__":
    generate_with_replicate(
        prompt="A serene Japanese garden in spring, cherry blossoms falling, golden hour light",
        width=1024,
        height=768
    )

---

에러 처리 레퍼런스

실제 운영 중 만나는 에러와 해결 방법이다.

HTTP 상태 코드	에러 메시지	원인	해결 방법
401	Unauthorized	API 키 없음 또는 잘못됨	.env 파일의 BFL_API_KEY 값 확인
402	Payment Required	크레딧 부족	BFL 대시보드에서 크레딧 충전
422	Unprocessable Entity	파라미터 값이 유효 범위 밖	width/height가 256~1440 사이이고 32의 배수인지 확인
429	Too Many Requests	Rate limit 초과	Exponential backoff 적용 (위 production_client.py 참고)
500	Internal Server Error	서버 측 문제	5~10초 후 재시도. 지속되면 BFL Status 확인
status: "Error"	응답 내 error 필드	프롬프트 안전 정책 위반	safety_tolerance 값 낮추거나 프롬프트 수정
TimeoutError	polling 타임아웃	서버 과부하 또는 네트워크 문제	MAX_WAIT_TIME 늘리거나 off-peak 시간대 재시도

자주 발생하는 비코드 문제

문제: 이미지 URL에 접속했더니 403 Forbidden 원인: BFL API가 반환하는 이미지 URL은 10분 후 만료된다. 해결: 생성 직후 _download_image()로 로컬에 저장한다. URL만 DB에 저장하면 나중에 접근 불가.

문제: width=1920, height=1080 설정했더니 오류 원인: 1920 * 1080 = 2,073,600 px² — BFL API의 최대 해상도 초과 해결: 1280x720 또는 1024x576으로 조정

---

성능 및 비용 비교

항목	BFL 직접 API	Replicate	Together AI
이미지 1장 비용	$0.04	~$0.055	~$0.045
평균 생성 시간	8~15초	10~20초	10~18초
무료 크레딧	없음	$5 (가입 시)	있음
Rate limit (기본)	문서화 미흡	Tier별 상이	Tier별 상이
SLA/Uptime	99.9% (공식 발표)	99.5% (공식 발표)	미공개
추가 파라미터	기본 파라미터만	output_quality 등 추가	제한적

언제 무엇을 쓸까:

• BFL 직접 API: 비용 최적화가 중요하고, 볼륨이 큰 production 환경
• Replicate: 프로토타이핑, 처음 테스트할 때, 무료 크레딧으로 검증 후 전환
• Together AI: Streamlit이나 다른 Together 모델과 함께 묶어서 쓸 때 편리

---

이 API를 쓰지 말아야 할 경우

모든 상황에 FLUX 1.1 Pro API가 적합한 것은 아니다.

• 실시간 응답이 필요한 경우: 평균 10초 지연은 동기 UX에 부적합하다. 비동기 큐 기반 아키텍처가 필수다.
• 100장 이상 배치 생성: 이 경우 비용이 $4+ 이상 나온다. ComfyUI + 로컬 GPU 운영이 더 경제적일 수 있다.
• 매우 정확한 텍스트 렌더링이 필요한 경우: FLUX 계열은 이미지 내 텍스트 렌더링 정확도가 DALL-E 3보다 낮다.
• 월 생성량이 1,000장 이하인 small project: Replicate의 서버리스 방식이 인프라 관리 없이 더 편하다.

---

마무리

이 튜토리얼의 코드 5개 블록은 모두 복사 후 환경 변수만 세팅하면 즉시 실행 가능하다. basic_generation.py로 API 작동을 확인하고, production_client.py의 FluxProClient를 서비스 코드에 통합하면 된다. 이미지 URL 만료(10분), width/height의 32 배수 제약, polling 타임아웃 처리 세 가지가 실제 운영에서 가장 자주 문제가 되는 지점이므로 이 부분을 중심으로 통합 테스트를 작성하는 것을 권장한다.

참고: 여러 AI 모델을 하나의 파이프라인에서 사용한다면, AtlasCloud는 Kling, Flux, Seedance, Claude, GPT 등 300개 이상의 모델에 단일 API로 접근할 수 있습니다. API 키 하나로 모든 모델 사용 가능. 신규 사용자는 첫 충전 시 25% 보너스(최대 $100).