nodes/ritik-devsecops-ComfyUI-Floyo-Flux2-API-node
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add Floyo Flux2 Pro node at repo root

Browse Source
This commit is contained in:
RITIK
2025-11-26 13:30:31 +05:30
parent 8613c4c976
commit e1d4eec5a9
7 changed files with 409 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+47 -1
View File
@@ -1 +1,47 @@
# ComfyUI-Floyo-Flux2-API-node
# Floyo FLUX.2 [pro] Custom Node
Custom ComfyUI node for Floyo that wraps the Black Forest Labs FLUX.2 **[pro]** API. URL-first (no local file paths): you pass URLs and get back a signed output URL for downstream Floyo nodes to download.
## Features
- Text-to-image generation with width/height, seed, safety tolerance, and output format controls.
- Image editing with multi-reference support (`input_image` + up to 7 extra refs).
- Clear URL-only inputs/outputs to match Floyo’s storage pipeline.
- Configurable API key via `config.ini` or `BFL_API_KEY` env var.
- Built-in polling with basic error handling and timeouts.
## Installation
1. Place this repository folder (e.g., `Flux2-Pro`) into `ComfyUI/custom_nodes/`.
2. Install dependencies:
```bash
cd Flux2-Pro # or your folder name under custom_nodes
pip install -r requirements.txt
```
3. Add your key to `config.ini` or export it:
```ini
[API]
BFL_API_KEY = <your_real_key>
```
```bash
export BFL_API_KEY=<your_real_key>
```
4. Restart ComfyUI.
## Nodes and I/O
### FLUX.2 [pro] Text-to-Image (`Flux2ProTextToImage`)
- **Inputs (required):** `prompt`, `width`, `height`
- **Inputs (optional):** `seed` (`-1` = random), `safety_tolerance` (0-6), `output_format` (`jpeg`/`png`)
- **Output:** `image_url` (STRING, signed URL valid for ~10 minutes)
### FLUX.2 [pro] Image Edit (`Flux2ProImageEdit`)
- **Inputs (required):** `prompt`, `input_image` (URL)
- **Inputs (optional):** `input_image_2` … `input_image_8` (extra refs, keep total <= 8), `width`, `height` (0 = match input), `seed` (`-1` = random), `safety_tolerance`, `output_format`
- **Output:** `image_url` (STRING, signed URL valid for ~10 minutes)
## Usage Notes
- The returned URL expires quickly; run your Floyo download/output node immediately after this node.
- API statuses handled: `Ready`, `Pending`, and error/moderation states. Failures return an error string in the node output.
- Keep total reference images within FLUX.2 [pro] limits (max 8 images, 9MP total). Width/height must be multiples of 16; `0` uses the input image dimensions for edits. Max resolution 4MP (e.g., 2048x2048).
## API Reference
- Endpoint: `https://api.bfl.ai/v1/flux-2-pro`
- Docs: https://docs.bfl.ai/quick_start/introduction
__init__.py
+13
View File
@@ -0,0 +1,13 @@
from .flux2_node import (
Flux2ProImageEdit,
Flux2ProTextToImage,
NODE_CLASS_MAPPINGS,
NODE_DISPLAY_NAME_MAPPINGS,
)
__all__ = [
"Flux2ProTextToImage",
"Flux2ProImageEdit",
"NODE_CLASS_MAPPINGS",
"NODE_DISPLAY_NAME_MAPPINGS",
]
config.ini
+2
View File
@@ -0,0 +1,2 @@
[API]
BFL_API_KEY = <your_bfl_api_key_here>
flux2_config.py
+42
View File
@@ -0,0 +1,42 @@
import configparser
import os
class Flux2Config:
"""Singleton helper to load and expose the Black Forest Labs API key."""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialize()
return cls._instance
def _initialize(self):
current_dir = os.path.dirname(os.path.abspath(__file__))
config_path = os.path.join(current_dir, "config.ini")
config = configparser.ConfigParser()
config.read(config_path)
key_from_env = os.environ.get("BFL_API_KEY")
key_from_file = config.get("API", "BFL_API_KEY", fallback=None)
if key_from_env:
self._key = key_from_env
print("BFL_API_KEY loaded from environment.")
elif key_from_file:
self._key = key_from_file
os.environ.setdefault("BFL_API_KEY", self._key)
print("BFL_API_KEY loaded from config.ini and set in environment.")
else:
self._key = None
print("WARNING: No BFL_API_KEY found. Please set it in config.ini or as an environment variable.")
if self._key and self._key.strip() == "<your_bfl_api_key_here>":
print("WARNING: BFL_API_KEY is still the placeholder value. Replace it with a real key from https://api.bfl.ai.")
def get_key(self):
"""Return the configured API key or None when missing."""
return self._key
flux2_node.py
+182
View File
@@ -0,0 +1,182 @@
from typing import List
from .flux2_utils import Flux2API, merge_reference_images
def _validate_resolution(width: int, height: int) -> str | None:
"""Validate resolution constraints for FLUX.2."""
for dim_name, dim in (("width", width), ("height", height)):
if dim <= 0:
continue # 0 means "use default/match input"
if dim % 16 != 0:
return f"{dim_name} must be a multiple of 16 (got {dim})."
if dim < 64 or dim > 2048:
return f"{dim_name} must be between 64 and 2048 pixels (got {dim})."
return None
class Flux2ProTextToImage:
"""
FLUX.2 [pro] text-to-image node for Floyo.
Accepts prompt + resolution and returns the signed image URL from the API.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"prompt": ("STRING", {"multiline": True, "default": "", "tooltip": "Describe what to generate. Supports long prompts (up to ~32k tokens per API docs)."}),
"width": ("INT", {"default": 1024, "min": 64, "max": 2048, "step": 16, "tooltip": "Output width in pixels. Must be a multiple of 16. Range: 64-2048 (max ~4MP)."}),
"height": ("INT", {"default": 1024, "min": 64, "max": 2048, "step": 16, "tooltip": "Output height in pixels. Must be a multiple of 16. Range: 64-2048 (max ~4MP)."}),
},
"optional": {
"seed": ("INT", {"default": -1, "min": -1, "max": 0xFFFFFFFF, "tooltip": "-1 = random. Any other integer gives reproducible results."}),
"safety_tolerance": ("INT", {"default": 2, "min": 0, "max": 6, "tooltip": "Moderation level. 0 = strict, 6 = most permissive (per BFL docs)."}),
"output_format": (["jpeg", "png"], {"default": "jpeg", "tooltip": "Output format for the returned image URL."}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("image_url",)
FUNCTION = "generate"
CATEGORY = "Floyo/Flux2 Pro"
def generate(
self,
prompt: str,
width: int,
height: int,
seed: int = -1,
safety_tolerance: int = 2,
output_format: str = "jpeg",
):
try:
resolution_error = _validate_resolution(width, height)
if resolution_error:
return (f"Error: {resolution_error}",)
client = Flux2API()
payload = {
"prompt": prompt,
"width": width if width > 0 else None,
"height": height if height > 0 else None,
"seed": None if seed is None or seed < 0 else seed,
"safety_tolerance": safety_tolerance,
"output_format": output_format,
}
run_result = client.run(payload)
image_url = run_result.get("sample")
if not image_url:
return ("Error: FLUX.2 response did not include an image URL.",)
return (image_url,)
except Exception as exc: # noqa: BLE001 - ComfyUI expects string errors
return (f"Error generating with FLUX.2: {exc}",)
class Flux2ProImageEdit:
"""
FLUX.2 [pro] image editing node for Floyo.
Provide an input image URL plus optional reference images to guide edits.
"""
@classmethod
def INPUT_TYPES(cls):
optional_refs = {
f"input_image_{idx}": (
"STRING",
{"default": "", "tooltip": f"Optional reference image URL #{idx} (max 7 extras, total refs <= 8)."},
)
for idx in range(2, 9)
}
return {
"required": {
"prompt": ("STRING", {"multiline": True, "default": "", "tooltip": "Describe the edit you want (what to change/keep)."}),
"input_image": ("STRING", {"default": "", "tooltip": "Base image URL to edit. Required."}),
},
"optional": {
**optional_refs,
"width": ("INT", {"default": 0, "min": 0, "max": 2048, "step": 16, "tooltip": "Optional override. 0 = use input image width. Must be multiple of 16; range 64-2048 if set."}),
"height": ("INT", {"default": 0, "min": 0, "max": 2048, "step": 16, "tooltip": "Optional override. 0 = use input image height. Must be multiple of 16; range 64-2048 if set."}),
"seed": ("INT", {"default": -1, "min": -1, "max": 0xFFFFFFFF, "tooltip": "-1 = random. Any other integer gives reproducible results."}),
"safety_tolerance": ("INT", {"default": 2, "min": 0, "max": 6, "tooltip": "Moderation level. 0 = strict, 6 = most permissive (per BFL docs)."}),
"output_format": (["jpeg", "png"], {"default": "jpeg", "tooltip": "Output format for the returned image URL."}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("image_url",)
FUNCTION = "edit"
CATEGORY = "Floyo/Flux2 Pro"
def edit(
self,
prompt: str,
input_image: str,
input_image_2: str = "",
input_image_3: str = "",
input_image_4: str = "",
input_image_5: str = "",
input_image_6: str = "",
input_image_7: str = "",
input_image_8: str = "",
width: int = 0,
height: int = 0,
seed: int = -1,
safety_tolerance: int = 2,
output_format: str = "jpeg",
):
try:
resolution_error = _validate_resolution(width, height)
if resolution_error:
return (f"Error: {resolution_error}",)
additional_images: List[str] = [
input_image_2,
input_image_3,
input_image_4,
input_image_5,
input_image_6,
input_image_7,
input_image_8,
]
reference_payload = merge_reference_images(input_image, additional_images)
if not reference_payload.get("input_image"):
return ("Error: input_image URL is required for image editing.",)
payload = {
"prompt": prompt,
**reference_payload,
"width": width if width > 0 else None,
"height": height if height > 0 else None,
"seed": None if seed is None or seed < 0 else seed,
"safety_tolerance": safety_tolerance,
"output_format": output_format,
}
client = Flux2API()
run_result = client.run(payload)
image_url = run_result.get("sample")
if not image_url:
return ("Error: FLUX.2 response did not include an image URL.",)
return (image_url,)
except Exception as exc: # noqa: BLE001 - ComfyUI expects string errors
return (f"Error editing with FLUX.2: {exc}",)
NODE_CLASS_MAPPINGS = {
"Flux2ProTextToImage": Flux2ProTextToImage,
"Flux2ProImageEdit": Flux2ProImageEdit,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"Flux2ProTextToImage": "FLUX.2 [pro] Text-to-Image",
"Flux2ProImageEdit": "FLUX.2 [pro] Image Edit",
}
flux2_utils.py
+122
View File
@@ -0,0 +1,122 @@
import time
from typing import Any, Dict, List, Optional
import requests
from .flux2_config import Flux2Config
class Flux2APIError(Exception):
"""Raised when the FLUX.2 API returns an error or an unexpected response."""
class Flux2API:
"""Minimal client for the Black Forest Labs FLUX.2 [pro] endpoint."""
BASE_URL = "https://api.bfl.ai/v1/flux-2-pro"
def __init__(self, timeout: int = 600, poll_interval: float = 1.0):
self.timeout = timeout
self.poll_interval = poll_interval
api_key = Flux2Config().get_key()
if not api_key:
raise Flux2APIError(
"BFL_API_KEY is not set. Define it in config.ini or export BFL_API_KEY before running ComfyUI."
)
self.headers = {
"accept": "application/json",
"Content-Type": "application/json",
"x-key": api_key,
}
@staticmethod
def _strip_empty(payload: Dict[str, Any]) -> Dict[str, Any]:
"""Remove keys with None or empty-string values so we only send what the API needs."""
cleaned: Dict[str, Any] = {}
for key, value in payload.items():
if value is None:
continue
if isinstance(value, str) and value.strip() == "":
continue
cleaned[key] = value
return cleaned
def submit_request(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Submit a generation/editing request and return the raw API response."""
body = self._strip_empty(payload)
response = requests.post(self.BASE_URL, headers=self.headers, json=body, timeout=30)
response.raise_for_status()
data = response.json()
if "polling_url" not in data:
raise Flux2APIError("FLUX.2 API response did not include a polling_url.")
if "id" not in data:
raise Flux2APIError("FLUX.2 API response did not include a request id.")
return data
def poll_result(self, polling_url: str, request_id: Optional[str] = None) -> Dict[str, Any]:
"""Poll the provided polling_url until the job is ready or fails."""
start_time = time.time()
last_status: Optional[str] = None
while (time.time() - start_time) < self.timeout:
response = requests.get(
polling_url,
headers=self.headers,
params={"id": request_id} if request_id else None,
timeout=30,
)
response.raise_for_status()
payload = response.json()
status = payload.get("status")
if status != last_status:
print(f"[Flux2] Status: {status}")
last_status = status
if status == "Ready":
return payload
if status in {"Error", "Failed", "Content Moderated", "Request Moderated"}:
raise Flux2APIError(f"FLUX.2 request failed with status '{status}': {payload}")
time.sleep(self.poll_interval)
raise Flux2APIError(f"Timed out after {self.timeout}s while waiting for the FLUX.2 result.")
def run(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Submit a request and wait for completion, returning convenience fields and raw payloads."""
initial = self.submit_request(payload)
polling_url = initial["polling_url"]
request_id = initial.get("id")
print(f"[Flux2] Submitted request: {request_id}")
final = self.poll_result(polling_url, request_id=request_id)
sample_url = final.get("result", {}).get("sample")
cost = initial.get("cost")
return {
"sample": sample_url,
"cost": cost,
"request": initial,
"result": final,
}
def merge_reference_images(main_image: str, additional_images: List[str]) -> Dict[str, str]:
"""
Build the payload keys for multi-reference inputs.
FLUX.2 [pro] accepts input_image + up to 7 extras: input_image_2 ... input_image_8.
"""
payload: Dict[str, str] = {}
if main_image:
payload["input_image"] = main_image
for idx, image in enumerate(additional_images, start=2):
if idx > 8:
break
if image and image.strip():
payload[f"input_image_{idx}"] = image.strip()
return payload
requirements.txt
+1
View File
@@ -0,0 +1 @@
requests>=2.31.0