教程指南

How to Use Flux Kontext API in Python

AI API Playbook · · 7 分钟阅读
How to Use Flux Kontext API in Python

你将构建什么

本教程将带你使用 Flux Kontext API 在 Python 中实现图像编辑与生成功能——包括文本驱动的图像修改、局部重绘（inpainting）以及风格迁移。Flux Kontext 由 Black Forest Labs 推出，支持最高 2048×2048 分辨率输出，API 调用延迟（TTFT）通常在 800ms–2.5s 之间，具体取决于分辨率和并发负载。

前置条件

  • Python 3.9+
  • requests 库（pip install requests）
  • Pillow 库用于图像处理（pip install Pillow）
  • Flux Kontext API 密钥（来自 Black Forest Labs 或聚合平台）
  • 基础 HTTP API 调用经验

分步骤实现

第一步：安装依赖并配置环境

# Install required packages:
# pip install requests Pillow python-dotenv

import os
import requests
import base64
import time
from pathlib import Path
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

API_KEY = os.getenv("FLUX_API_KEY")
BASE_URL = "https://api.bfl.ml/v1"  # Black Forest Labs official endpoint

if not API_KEY:
    raise EnvironmentError("FLUX_API_KEY not set. Check your .env file.")

创建 .env 文件：

FLUX_API_KEY=your_api_key_here

第二步：封装基础请求客户端

import requests
import time
import base64
from pathlib import Path

class FluxKontextClient:
    """
    A simple client for the Flux Kontext API.
    Handles authentication, polling, and response parsing.
    """

    def __init__(self, api_key: str, base_url: str = "https://api.bfl.ml/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "x-key": self.api_key,
            "Content-Type": "application/json"
        })

    def _encode_image(self, image_path: str) -> str:
        """Encode a local image file to base64 string."""
        with open(image_path, "rb") as f:
            return base64.b64encode(f.read()).decode("utf-8")

    def submit_task(self, endpoint: str, payload: dict) -> str:
        """
        Submit an async generation task and return the task ID.
        Returns task_id (string) for polling.
        """
        url = f"{self.base_url}/{endpoint}"
        response = self.session.post(url, json=payload)
        response.raise_for_status()
        data = response.json()
        return data["id"]

    def poll_result(self, task_id: str, timeout: int = 120, interval: float = 2.0) -> dict:
        """
        Poll the result endpoint until the task is complete or timeout is reached.
        Flux Kontext uses async polling — typical wait is 3–15 seconds.
        """
        url = f"{self.base_url}/get_result"
        start_time = time.time()

        while True:
            elapsed = time.time() - start_time
            if elapsed > timeout:
                raise TimeoutError(f"Task {task_id} timed out after {timeout}s")

            response = self.session.get(url, params={"id": task_id})
            response.raise_for_status()
            result = response.json()

            status = result.get("status")

            if status == "Ready":
                return result
            elif status in ("Error", "Failed"):
                raise RuntimeError(f"Task failed: {result.get('result', 'Unknown error')}")
            elif status == "Content Moderated":
                raise ValueError("Request was rejected by content moderation.")

            # Wait before next poll
            time.sleep(interval)

    def download_image(self, url: str, save_path: str) -> None:
        """Download the generated image from a signed URL."""
        response = requests.get(url)
        response.raise_for_status()
        Path(save_path).parent.mkdir(parents=True, exist_ok=True)
        with open(save_path, "wb") as f:
            f.write(response.content)
        print(f"Image saved to: {save_path}")

第三步：文本到图像生成（Flux Kontext Pro）

def text_to_image(client: FluxKontextClient, prompt: str, output_path: str = "output/generated.jpg"):
    """
    Generate an image from a text prompt using flux-kontext-pro.
    Typical latency: 3–8 seconds end-to-end at 1024x1024.
    """
    payload = {
        "prompt": prompt,
        "width": 1024,
        "height": 1024,
        "output_format": "jpeg",      # "jpeg" or "png"
        "safety_tolerance": 2,         # 0 (strict) to 6 (permissive)
        "prompt_upsampling": False     # Set True to let the model enhance short prompts
    }

    print(f"Submitting text-to-image task...")
    task_id = client.submit_task("flux-kontext-pro", payload)
    print(f"Task ID: {task_id} — polling for result...")

    result = client.poll_result(task_id)
    image_url = result["result"]["sample"]

    client.download_image(image_url, output_path)
    return image_url


# --- Run it ---
if __name__ == "__main__":
    import os
    from dotenv import load_dotenv
    load_dotenv()

    client = FluxKontextClient(api_key=os.getenv("FLUX_API_KEY"))

    url = text_to_image(
        client,
        prompt="A futuristic Tokyo street at night, neon lights reflected in rain puddles, cinematic lighting, 8K",
        output_path="output/tokyo_night.jpg"
    )
    print(f"Generated image URL: {url}")

第四步：图像编辑（Kontext 核心功能）

Flux Kontext 的核心能力是基于参考图的图像编辑（context-aware editing）。你可以传入一张原图和编辑指令，模型会在保留原图结构的同时应用修改。

def edit_image(
    client: FluxKontextClient,
    input_image_path: str,
    edit_prompt: str,
    output_path: str = "output/edited.jpg"
):
    """
    Edit an existing image using a text instruction.
    Flux Kontext preserves structure/identity while applying edits.
    Input image must be JPEG or PNG, max 10MB.
    """
    # Encode input image to base64
    image_b64 = client._encode_image(input_image_path)

    payload = {
        "prompt": edit_prompt,
        "input_image": image_b64,
        "width": 1024,
        "height": 1024,
        "output_format": "jpeg",
        "safety_tolerance": 2,
        "prompt_upsampling": False
    }

    print(f"Submitting image edit task for: {input_image_path}")
    task_id = client.submit_task("flux-kontext-pro", payload)
    print(f"Task ID: {task_id} — waiting for edit to complete...")

    result = client.poll_result(task_id, timeout=120)
    image_url = result["result"]["sample"]

    client.download_image(image_url, output_path)
    print(f"Edited image saved to: {output_path}")
    return image_url


# --- Example usage ---
if __name__ == "__main__":
    import os
    from dotenv import load_dotenv
    load_dotenv()

    client = FluxKontextClient(api_key=os.getenv("FLUX_API_KEY"))

    edit_image(
        client,
        input_image_path="input/portrait.jpg",
        edit_prompt="Change the background to a snowy mountain landscape, keep the person unchanged",
        output_path="output/portrait_edited.jpg"
    )

第五步：批量处理多张图像

import concurrent.futures
from typing import List, Tuple

def batch_edit_images(
    client: FluxKontextClient,
    tasks: List[Tuple[str, str, str]],
    max_workers: int = 3
) -> List[str]:
    """
    Process multiple image edit tasks in parallel using ThreadPoolExecutor.
    tasks: list of (input_image_path, edit_prompt, output_path) tuples
    max_workers: limit concurrency to avoid API rate limits (recommended: 3–5)
    Returns: list of output image URLs
    """

    def process_single(task: Tuple[str, str, str]) -> str:
        input_path, prompt, output_path = task
        try:
            url = edit_image(client, input_path, prompt, output_path)
            print(f"✓ Done: {output_path}")
            return url
        except Exception as e:
            print(f"✗ Failed for {input_path}: {e}")
            return ""

    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(process_single, task) for task in tasks]
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())

    return results


# --- Example batch job ---
if __name__ == "__main__":
    import os
    from dotenv import load_dotenv
    load_dotenv()

    client = FluxKontextClient(api_key=os.getenv("FLUX_API_KEY"))

    tasks = [
        ("input/photo1.jpg", "Add dramatic sunset lighting", "output/photo1_sunset.jpg"),
        ("input/photo2.jpg", "Convert to watercolor painting style", "output/photo2_watercolor.jpg"),
        ("input/photo3.jpg", "Remove background, keep subject only", "output/photo3_nobg.jpg"),
    ]

    urls = batch_edit_images(client, tasks, max_workers=3)
    print(f"Completed {len([u for u in urls if u])} / {len(tasks)} tasks")

第六步：curl 等效调用参考

#!/bin/bash
# Step 1: Submit a text-to-image task
RESPONSE=$(curl -s -X POST "https://api.bfl.ml/v1/flux-kontext-pro" \
  -H "x-key: $FLUX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A futuristic Tokyo street at night, neon lights, cinematic",
    "width": 1024,
    "height": 1024,
    "output_format": "jpeg",
    "safety_tolerance": 2
  }')

echo "Submit response: $RESPONSE"
TASK_ID=$(echo $RESPONSE | python3 -c "import sys,json; print(json.load(sys.stdin)['id'])")
echo "Task ID: $TASK_ID"

# Step 2: Poll for result (repeat until status == "Ready")
while true; do
  RESULT=$(curl -s "https://api.bfl.ml/v1/get_result?id=$TASK_ID" \
    -H "x-key: $FLUX_API_KEY")
  
  STATUS=$(echo $RESULT | python3 -c "import sys,json; print(json.load(sys.stdin)['status'])")
  echo "Status: $STATUS"

  if [ "$STATUS" = "Ready" ]; then
    IMAGE_URL=$(echo $RESULT | python3 -c "import sys,json; print(json.load(sys.stdin)['result']['sample'])")
    echo "Image URL: $IMAGE_URL"
    # Step 3: Download the image
    curl -o "output/generated.jpg" "$IMAGE_URL"
    break
  elif [ "$STATUS" = "Error" ] || [ "$STATUS" = "Failed" ]; then
    echo "Task failed: $RESULT"
    break
  fi

  sleep 2
done

错误处理与最佳实践

完整错误处理示例

import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

def safe_generate(client: FluxKontextClient, prompt: str, retries: int = 3) -> str | None:
    """
    Robustly call the API with retry logic, exponential backoff,
    and specific error handling for common failure modes.
    """
    for attempt in range(1, retries + 1):
        try:
            task_id = client.submit_task("flux-kontext-pro", {
                "prompt": prompt,
                "width": 1024,
                "height": 1024,
                "output_format": "jpeg",
                "safety_tolerance": 2
            })
            result = client.poll_result(task_id)
            return result["result"]["sample"]

        except requests.exceptions.HTTPError as e:
            status_code = e.response.status_code
            if status_code == 402:
                # Insufficient credits — do not retry
                logger.error("402 Payment Required: insufficient API credits. Stopping.")
                raise
            elif status_code == 422:
                # Validation error (bad parameters) — do not retry
                logger.error(f"422 Unprocessable Entity: check your payload. {e.response.text}")
                raise
            elif status_code == 429:
                # Rate limit — exponential backoff
                wait = 2 ** attempt
                logger.warning(f"429 Rate Limited. Waiting {wait}s before retry {attempt}/{retries}...")
                time.sleep(wait)
            elif status_code >= 500:
                # Server error — retry with backoff
                wait = 2 ** attempt
                logger.warning(f"Server error {status_code}. Retrying in {wait}s ({attempt}/{retries})...")
                time.sleep(wait)
            else:
                raise

        except TimeoutError:
            logger.warning(f"Task timed out on attempt {attempt}. Retrying...")

        except ValueError as e:
            # Content moderation rejection — do not retry
            logger.error(f"Content moderation rejection: {e}")
            return None

    logger.error(f"All {retries} attempts failed.")
    return None

最佳实践要点

图像输入规范：输入图像建议压缩至 2MB 以下，分辨率不超过 2048px，避免因 base64 编码导致请求体过大（API 限制单次请求 10MB）。

提示词工程：Flux Kontext 对具体的结构描述（如”keep the person’s face unchanged”）响应显著优于模糊指令，在图像编辑任务中准确率可

在 AtlasCloud 上试用此 API

AtlasCloud

标签

Flux Image Generation Python API Tutorial Black Forest Labs

相关文章

教程指南

FLUX 1.1 Pro API Python教程：5分钟实现AI图像生成

本教程手把手教你用Python调用FLUX 1.1 Pro API，快速实现高质量AI图像生成。包含完整代码示例、参数配置和常见问题解决方案，新手也能5分钟上手！

教程指南

Kling v3 API使用教程：2026年Python完整开发指南

本文详细介绍如何使用Kling v3 API进行Python开发，涵盖环境配置、身份验证、视频生成及高级功能，附完整代码示例，助你快速上手Kling v3 API开发。

教程指南

AI 图像生成 API 入门：DALL-E 3、Midjourney 和 Stable Diffusion 实战教程

一篇实用的 AI 图像生成 API 集成教程，学习使用 DALL-E 3、Midjourney 和 Stable Diffusion API 的代码示例与最佳实践。