Merge 2066b025b1 into f7220556e1

litellm/llms/black_forest_labs/videos/__init__.py

@@ -0,0 +1,3 @@
+from .transformation import BlackForestLabsVideoConfig
+
+__all__ = ["BlackForestLabsVideoConfig"]

litellm/llms/black_forest_labs/videos/transformation.py

@@ -0,0 +1,540 @@
+"""
+Black Forest Labs FLUX 3 Video Configuration
+
+Handles transformation between OpenAI-compatible video params and the Black
+Forest Labs FLUX 3 video API.
+
+API Reference: https://docs.bfl.ai/api-reference/utility/generate-a-video-with-flux-3
+"""
+
+import time
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Any, Final  # noqa: TID251  # BaseVideoConfig types its payloads dict[str, Any]
+from urllib.parse import urlparse
+
+import httpx
+
+import litellm
+from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
+from litellm.llms.custom_httpx.http_handler import (
+    AsyncHTTPHandler,
+    HTTPHandler,
+    _get_httpx_client,  # pyright: ignore[reportPrivateUsage]  # the shared client factory every provider uses to fetch generated media
+    get_async_httpx_client,
+)
+from litellm.secret_managers.main import get_secret_str
+from litellm.types.router import GenericLiteLLMParams
+from litellm.types.videos.main import VideoCreateOptionalRequestParams, VideoObject
+from litellm.types.videos.utils import (
+    encode_video_id_with_provider,
+    extract_original_video_id,
+)
+
+from ..common_utils import (
+    DEFAULT_API_BASE,
+    BlackForestLabsError,
+    assert_bfl_polling_url,
+)
+
+if TYPE_CHECKING:
+    from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
+
+    LiteLLMLoggingObj = _LiteLLMLoggingObj  # rebind-ok: the TYPE_CHECKING alias for the runtime Any below
+else:
+    LiteLLMLoggingObj = Any  # rebind-ok: runtime stand-in for the type-only Logging alias
+
+VIDEO_MODELS: Final[Mapping[str, str]] = MappingProxyType({"flux-3-video": "/v1/flux-3-video"})
+
+RESOLUTIONS: Final = ("hd", "fhd")
+ASPECT_RATIOS: Final = ("21:9", "2:1", "16:9", "4:3", "1:1", "3:4", "9:16", "auto")
+MIN_DURATION: Final = 5
+MAX_DURATION: Final = 20
+
+# BFL reports one of these on GET /v1/get_result. Anything outside the terminal
+# set is still running.
+_TERMINAL_STATUSES: Final[Mapping[str, str]] = MappingProxyType(
+    {
+        "Ready": "completed",
+        "Error": "failed",
+        "Content Moderated": "failed",
+        "Request Moderated": "failed",
+        "Task not found": "failed",
+    }
+)
+_IN_PROGRESS_STATUSES: Final[Mapping[str, str]] = MappingProxyType(
+    {
+        "Pending": "queued",
+        "Queued": "queued",
+        "Reasoning": "in_progress",
+        "Generating": "in_progress",
+        "Uploading": "in_progress",
+    }
+)
+
+# BFL dispatches each job to a regional host and hands that host back in
+# ``polling_url``. The global host answers 404 for a regional job, so the region
+# has to survive from submission through to status and content retrieval. The
+# only value carried across those calls is the video id, so the region travels
+# inside it, behind a separator BFL's own UUIDs never contain.
+_REGION_SEPARATOR: Final = "@"
+
+
+def _pack_region(job_id: str, polling_url: str | None) -> str:
+    """Attach the polling host to a job id, when it differs from the default."""
+    if not polling_url:
+        return job_id
+    host: Final = (urlparse(polling_url).hostname or "").lower()
+    if not host or host == urlparse(DEFAULT_API_BASE).hostname:
+        return job_id
+    return f"{job_id}{_REGION_SEPARATOR}{host}"
+
+
+def _unpack_region(job_id: str) -> tuple[str, str | None]:
+    """Split a packed id back into the bare job id and its polling host."""
+    if _REGION_SEPARATOR not in job_id:
+        return job_id, None
+    bare, _, host = job_id.partition(_REGION_SEPARATOR)
+    return bare, host or None
+
+
+class BlackForestLabsVideoConfig(BaseVideoConfig):
+    """
+    Configuration for Black Forest Labs FLUX 3 video generation.
+
+    FLUX 3 is a single endpoint with a ``mode`` discriminator: ``t2v`` from a
+    prompt alone, ``i2v`` from keyframe images, ``v2v`` to continue an existing
+    clip, and ``draft_enhance`` to re-render a draft at full quality.
+
+    Submission returns a job id plus a regional ``polling_url``. That URL has to
+    be reused verbatim, because the global host answers 404 for a job dispatched
+    to a region.
+    """
+
+    def get_supported_openai_params(self, model: str) -> list:  # mutable-ok: BaseVideoConfig signature
+        return [  # mutable-ok: BaseVideoConfig signature
+            "model",
+            "seconds",
+            "size",
+            "input_reference",
+            "user",
+            "extra_headers",
+        ]
+
+    def map_openai_params(
+        self,
+        video_create_optional_params: VideoCreateOptionalRequestParams,
+        model: str,
+        drop_params: bool,
+    ) -> dict:  # mutable-ok: BaseVideoConfig signature
+        """
+        Map OpenAI video params onto FLUX 3 params.
+
+        - ``seconds`` -> ``duration`` (whole seconds, 5 to 20)
+        - ``size`` -> ``resolution`` tier, by the shorter side
+        - ``input_reference`` -> a single ``keyframes`` entry, which selects i2v
+        """
+        supported: Final = self.get_supported_openai_params(model)
+        remapped: Final = ("seconds", "size", "input_reference")
+
+        duration: Final = self._map_duration(video_create_optional_params.get("seconds"))
+        resolution: Final = self._map_size_to_resolution(video_create_optional_params.get("size"))
+        input_reference: Final = video_create_optional_params.get("input_reference")
+
+        translated: Final = {  # mutable-ok: JSON request payload
+            key: value
+            for key, value in (
+                ("duration", duration),
+                ("resolution", resolution),
+                (
+                    "keyframes",
+                    [input_reference]  # mutable-ok: JSON request payload
+                    if input_reference is not None
+                    else None,  # mutable-ok: JSON request payload
+                ),
+            )
+            if value is not None
+        }
+        passthrough: Final = {  # mutable-ok: JSON request payload
+            key: value
+            for key, value in video_create_optional_params.items()
+            if key not in supported and key not in remapped
+        }
+
+        return {  # mutable-ok: JSON request payload
+            **translated,
+            **passthrough,
+        }  # mutable-ok: JSON request payload
+
+    def _map_duration(self, seconds: object) -> int | None:
+        if not isinstance(seconds, (int, float, str)):
+            return None
+        try:
+            duration: Final = int(float(seconds))
+        except (TypeError, ValueError):
+            return None
+        return max(MIN_DURATION, min(MAX_DURATION, duration))
+
+    def _map_size_to_resolution(self, size: object) -> str | None:
+        """
+        FLUX 3 takes a named tier, not pixel dimensions, so map by the shorter
+        side: at most 720 is ``hd`` and anything larger is ``fhd``.
+        """
+        if not isinstance(size, str):
+            return None
+        if size in RESOLUTIONS:
+            return size
+        if "x" not in size.lower():
+            return None
+        try:
+            width, height = (int(part) for part in size.lower().split("x", 1))
+        except ValueError:
+            return None
+        return "hd" if min(width, height) <= 720 else "fhd"
+
+    def validate_environment(
+        self,
+        headers: dict,  # mutable-ok: BaseVideoConfig signature
+        model: str,
+        api_key: str | None = None,
+        litellm_params: GenericLiteLLMParams | None = None,
+    ) -> dict:  # mutable-ok: BaseVideoConfig signature
+        # Bind a new name rather than rebinding the caller's parameter.
+        request_api_key: Final = api_key or (litellm_params.api_key if litellm_params else None)
+
+        # Resolve BFL credentials only. Falling back to ``litellm.api_key`` would
+        # send another provider's generic key to api.bfl.ai.
+        final_api_key: Final = (
+            request_api_key or get_secret_str("BFL_API_KEY") or get_secret_str("BLACK_FOREST_LABS_API_KEY")
+        )
+
+        if not final_api_key:
+            raise BlackForestLabsError(
+                status_code=401,
+                message="BFL_API_KEY is not set. Please set it via environment variable or pass api_key parameter.",
+            )
+
+        headers.update(
+            {  # mutable-ok: JSON request payload
+                "x-key": final_api_key,
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+            }
+        )
+        return headers
+
+    def get_complete_url(
+        self,
+        model: str,
+        api_base: str | None,
+        litellm_params: dict,  # mutable-ok: BaseVideoConfig signature
+    ) -> str:
+        base_url: Final[str] = api_base or get_secret_str("BFL_API_BASE") or DEFAULT_API_BASE
+        return base_url.rstrip("/")
+
+    def _get_model_endpoint(self, model: str) -> str:
+        model_name: Final = model.lower().split("/")[-1]
+        if model_name in VIDEO_MODELS:
+            return VIDEO_MODELS[model_name]
+        raise ValueError(
+            f"Unknown BFL video model: {model_name}. Supported models: {list(VIDEO_MODELS.keys())}"  # mutable-ok: one-shot, for an error message
+        )
+
+    def transform_video_create_request(
+        self,
+        model: str,
+        prompt: str,
+        api_base: str,
+        video_create_optional_request_params: dict,  # mutable-ok: BaseVideoConfig signature
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig signature
+    ) -> tuple[dict, list, str]:  # mutable-ok: BaseVideoConfig signature
+        base_request: Final = {  # mutable-ok: JSON request payload
+            "prompt": prompt,
+            **video_create_optional_request_params,
+        }  # mutable-ok: JSON request payload
+        mode: Final = self._infer_mode(base_request)
+
+        # draft_enhance re-renders a cached draft, so it carries no prompt.
+        request_data: Final = {  # mutable-ok: JSON request payload
+            **{  # mutable-ok: JSON request payload
+                key: value for key, value in base_request.items() if not (mode == "draft_enhance" and key == "prompt")
+            },  # mutable-ok: JSON request payload
+            "mode": mode,
+        }
+
+        url: Final = f"{api_base}{self._get_model_endpoint(model)}"
+        return request_data, [], url  # mutable-ok: JSON request payload
+
+    def _infer_mode(self, request_data: Mapping[str, Any]) -> str:
+        """FLUX 3 discriminates on ``mode``; derive it from the inputs given."""
+        if request_data.get("mode"):
+            return str(request_data["mode"])
+        if request_data.get("draft_cache"):
+            return "draft_enhance"
+        if request_data.get("start_video"):
+            return "v2v"
+        if request_data.get("keyframes"):
+            return "i2v"
+        return "t2v"
+
+    def transform_video_create_response(
+        self,
+        model: str,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        custom_llm_provider: str | None = None,
+        request_data: dict | None = None,  # mutable-ok: BaseVideoConfig signature
+    ) -> VideoObject:
+        """
+        Submission answers with the job handle, not the finished video:
+        ``{"id": ..., "polling_url": ..., "cost": null}``.
+
+        The regional polling URL is kept on the video id so status and content
+        calls hit the same region.
+        """
+        response_data: Final = self._parse_json(raw_response)
+
+        job_id: Final = response_data.get("id")
+        if not job_id:
+            raise BlackForestLabsError(
+                status_code=raw_response.status_code,
+                message=f"No job id in BFL response: {response_data}",
+            )
+
+        polling_url: Final = response_data.get("polling_url")
+        if polling_url:
+            assert_bfl_polling_url(polling_url)
+
+        # The region has to outlive this response, and the id is the only value
+        # the status and content calls receive.
+        regional_job_id: Final = _pack_region(job_id, polling_url)
+
+        video_obj: Final = VideoObject(
+            id=regional_job_id,
+            object="video",
+            status="queued",
+            created_at=int(time.time()),
+            model=model,
+        )
+
+        if request_data:
+            if request_data.get("duration") not in (None, "auto"):
+                video_obj.seconds = str(request_data["duration"])
+            if request_data.get("resolution"):
+                video_obj.size = str(request_data["resolution"])
+
+        if custom_llm_provider:
+            video_obj.id = encode_video_id_with_provider(regional_job_id, custom_llm_provider, model)
+
+        return video_obj
+
+    def transform_video_status_retrieve_request(
+        self,
+        video_id: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig signature
+    ) -> tuple[str, dict]:  # mutable-ok: BaseVideoConfig signature
+        return self._get_result_url(video_id, api_base), {}  # mutable-ok: JSON request payload
+
+    def transform_video_status_retrieve_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        custom_llm_provider: str | None = None,
+    ) -> VideoObject:
+        response_data: Final = self._parse_json(raw_response)
+        bfl_status: Final = response_data.get("status", "Pending")
+
+        video_obj: Final = VideoObject(
+            id=response_data.get("id", ""),
+            object="video",
+            status=self._map_status(bfl_status),
+            progress=self._map_progress(response_data.get("progress")),
+        )
+
+        if bfl_status in _TERMINAL_STATUSES and _TERMINAL_STATUSES[bfl_status] == "failed":
+            video_obj.error = {  # mutable-ok: JSON request payload
+                "code": bfl_status,
+                "message": str(response_data.get("details") or bfl_status),
+            }
+
+        cost: Final = response_data.get("cost")
+        if cost is not None:
+            video_obj.usage = {  # mutable-ok: VideoObject field
+                "credits": cost
+            }  # mutable-ok: JSON request payload
+
+        if custom_llm_provider and video_obj.id:
+            # Re-attach the region: BFL echoes a bare job id, but the caller may
+            # reuse this id for content retrieval and must land on the same host.
+            polled_url: Final = self._polled_url(raw_response)
+            video_obj.id = encode_video_id_with_provider(
+                _pack_region(video_obj.id, polled_url), custom_llm_provider, None
+            )
+
+        return video_obj
+
+    def _polled_url(self, raw_response: httpx.Response) -> str | None:
+        """The URL this response came from, when httpx recorded one.
+
+        ``Response.request`` raises rather than returning None when the response
+        was built without a request, which is the case in unit tests.
+        """
+        try:
+            return str(raw_response.request.url)
+        except RuntimeError:
+            return None
+
+    def _map_status(self, bfl_status: str) -> str:
+        if bfl_status in _TERMINAL_STATUSES:
+            return _TERMINAL_STATUSES[bfl_status]
+        return _IN_PROGRESS_STATUSES.get(bfl_status, "in_progress")
+
+    def _map_progress(self, progress: object) -> int | None:
+        if not isinstance(progress, (int, float, str)):
+            return None
+        try:
+            value: Final = float(progress)
+        except (TypeError, ValueError):
+            return None
+        # BFL reports a 0..1 fraction; VideoObject.progress is a percentage.
+        return round(value * 100) if value <= 1 else round(value)
+
+    def transform_video_content_request(
+        self,
+        video_id: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig signature
+        variant: str | None = None,
+    ) -> tuple[str, dict]:  # mutable-ok: BaseVideoConfig signature
+        return self._get_result_url(video_id, api_base), {}  # mutable-ok: JSON request payload
+
+    def transform_video_content_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+    ) -> bytes:
+        video_url: Final = self._extract_video_url(self._parse_json(raw_response))
+        httpx_client: Final[HTTPHandler] = _get_httpx_client()
+        video_response: Final = httpx_client.get(video_url)
+        video_response.raise_for_status()
+        return video_response.content
+
+    async def async_transform_video_content_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+    ) -> bytes:
+        video_url: Final = self._extract_video_url(self._parse_json(raw_response))
+        async_client: Final[AsyncHTTPHandler] = get_async_httpx_client(
+            llm_provider=litellm.LlmProviders.BLACK_FOREST_LABS,
+        )
+        video_response: Final = await async_client.get(video_url)
+        video_response.raise_for_status()
+        return video_response.content
+
+    def _extract_video_url(self, response_data: Mapping[str, Any]) -> str:
+        status: Final = response_data.get("status", "Pending")
+        result: Final[Mapping[str, Any]] = response_data.get("result") or MappingProxyType({})
+        video_url: Final = result.get("sample")
+
+        if video_url:
+            return video_url
+
+        if status in _TERMINAL_STATUSES:
+            raise BlackForestLabsError(
+                status_code=500,
+                message=f"Video generation did not produce a video (status: {status}).",
+            )
+        raise BlackForestLabsError(
+            status_code=409,
+            message=f"Video is still processing (status: {status}). Please wait and try again.",
+        )
+
+    def _get_result_url(self, video_id: str, api_base: str) -> str:
+        """Build the polling URL, honouring the region the job was dispatched to.
+
+        The global host answers 404 for a regional job, so a packed id sends the
+        request back to the host BFL named at submission.
+        """
+        original_video_id: Final = extract_original_video_id(video_id)
+        job_id, region_host = _unpack_region(original_video_id)
+        host: Final = f"https://{region_host}" if region_host else api_base.rstrip("/")
+        result_url: Final = f"{host}/v1/get_result?id={job_id}"
+        assert_bfl_polling_url(result_url)
+        return result_url
+
+    def _parse_json(self, raw_response: httpx.Response) -> dict:  # mutable-ok: decoded JSON body
+        try:
+            return raw_response.json()
+        except Exception as e:
+            raise BlackForestLabsError(
+                status_code=raw_response.status_code,
+                message=f"Error parsing BFL response: {e}",
+            )
+
+    def transform_video_remix_request(
+        self,
+        video_id: str,
+        prompt: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig signature
+        extra_body: dict[str, Any] | None = None,  # mutable-ok: BaseVideoConfig signature
+    ) -> tuple[str, dict]:  # mutable-ok: BaseVideoConfig signature
+        raise NotImplementedError("video remix is not supported by the FLUX 3 video API")
+
+    def transform_video_remix_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        custom_llm_provider: str | None = None,
+    ) -> VideoObject:
+        raise NotImplementedError("video remix is not supported by the FLUX 3 video API")
+
+    def transform_video_list_request(
+        self,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig signature
+        after: str | None = None,
+        limit: int | None = None,
+        order: str | None = None,
+        extra_query: dict[str, Any] | None = None,  # mutable-ok: BaseVideoConfig signature
+    ) -> tuple[str, dict]:  # mutable-ok: BaseVideoConfig signature
+        raise NotImplementedError("video listing is not supported by the FLUX 3 video API")
+
+    def transform_video_list_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        custom_llm_provider: str | None = None,
+    ) -> dict[str, str]:  # mutable-ok: BaseVideoConfig signature
+        raise NotImplementedError("video listing is not supported by the FLUX 3 video API")
+
+    def transform_video_delete_request(
+        self,
+        video_id: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig signature
+    ) -> tuple[str, dict]:  # mutable-ok: BaseVideoConfig signature
+        raise NotImplementedError("video delete is not supported by the FLUX 3 video API")
+
+    def transform_video_delete_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+    ) -> VideoObject:
+        raise NotImplementedError("video delete is not supported by the FLUX 3 video API")
+
+    def get_error_class(
+        self,
+        error_message: str,
+        status_code: int,
+        headers: dict | httpx.Headers,  # mutable-ok: BaseVideoConfig signature
+    ) -> BlackForestLabsError:
+        return BlackForestLabsError(status_code=status_code, message=error_message)

litellm/utils.py

@@ -9037,6 +9037,12 @@ class ProviderConfigManager:
             from litellm.llms.runwayml.videos.transformation import RunwayMLVideoConfig
 
             return RunwayMLVideoConfig()
+        elif LlmProviders.BLACK_FOREST_LABS == provider:
+            from litellm.llms.black_forest_labs.videos.transformation import (
+                BlackForestLabsVideoConfig,
+            )
+
+            return BlackForestLabsVideoConfig()
         return None
 
     @staticmethod

model_prices_and_context_window.json

@@ -12160,6 +12160,23 @@
             "/v1/images/generations"
         ]
     },
+    "black_forest_labs/flux-3-video": {
+        "litellm_provider": "black_forest_labs",
+        "mode": "video_generation",
+        "output_cost_per_video_per_second": 0.17,
+        "source": "https://docs.bfl.ai/quick_start/pricing",
+        "supported_modalities": [
+            "text",
+            "image",
+            "video"
+        ],
+        "supported_output_modalities": [
+            "video"
+        ],
+        "metadata": {
+            "comment": "Per-second pricing varies by mode and resolution: t2v and i2v are $0.17/s hd and $0.29/s fhd, v2v is $0.43/s hd and $0.54/s fhd, drafts are $0.06/s (t2v, i2v) and $0.12/s (v2v). The t2v hd rate is used here."
+        }
+    },
     "cerebras/llama-3.3-70b": {
         "input_cost_per_token": 8.5e-07,
         "litellm_provider": "cerebras",

tests/test_litellm/llms/black_forest_labs/videos/test_bfl_video_transformation.py

@@ -0,0 +1,360 @@
+"""
+Tests for Black Forest Labs FLUX 3 video generation transformation.
+
+Payload and response shapes are taken from a live FLUX 3 video generation
+against https://api.bfl.ai/v1/flux-3-video and its regional polling URL.
+"""
+
+from unittest.mock import Mock
+
+import httpx
+import pytest
+
+from litellm.llms.black_forest_labs.common_utils import BlackForestLabsError
+from litellm.llms.black_forest_labs.videos.transformation import (
+    BlackForestLabsVideoConfig,
+)
+from litellm.types.router import GenericLiteLLMParams
+from litellm.types.videos.utils import extract_original_video_id
+
+API_BASE = "https://api.bfl.ai"
+JOB_ID = "90307d3a-deec-47cb-bdb9-bf1c5bae1f04"
+POLLING_URL = f"https://api.us7.bfl.ai/v1/get_result?id={JOB_ID}"
+SAMPLE_URL = "https://delivery.us7.bfl.ai/durable/2026081720/video.mp4?se=2026-08-17T21%3A26%3A38Z&sig=abc"
+
+
+def _response(payload: dict, status_code: int = 200) -> httpx.Response:
+    return httpx.Response(
+        status_code=status_code,
+        json=payload,
+        request=httpx.Request("POST", f"{API_BASE}/v1/flux-3-video"),
+    )
+
+
+class TestBlackForestLabsVideoTransformation:
+    def setup_method(self):
+        self.config = BlackForestLabsVideoConfig()
+        self.mock_logging_obj = Mock()
+
+    def test_text_to_video_request_sets_t2v_mode(self):
+        data, files, url = self.config.transform_video_create_request(
+            model="flux-3-video",
+            prompt="A white kitten chases a butterfly across a sunlit garden.",
+            api_base=API_BASE,
+            video_create_optional_request_params={
+                "duration": 8,
+                "resolution": "fhd",
+                "aspect_ratio": "16:9",
+            },
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert url == "https://api.bfl.ai/v1/flux-3-video"
+        assert data["mode"] == "t2v"
+        assert data["prompt"] == "A white kitten chases a butterfly across a sunlit garden."
+        assert data["duration"] == 8
+        assert data["resolution"] == "fhd"
+        assert files == []
+
+    @pytest.mark.parametrize(
+        "params, expected_mode",
+        [
+            ({}, "t2v"),
+            ({"keyframes": ["https://example.com/first.png"]}, "i2v"),
+            ({"start_video": "https://example.com/clip.mp4"}, "v2v"),
+            ({"draft_cache": "https://example.com/bundle.bin"}, "draft_enhance"),
+        ],
+    )
+    def test_mode_is_inferred_from_inputs(self, params, expected_mode):
+        data, _, _ = self.config.transform_video_create_request(
+            model="flux-3-video",
+            prompt="a prompt",
+            api_base=API_BASE,
+            video_create_optional_request_params=dict(params),
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert data["mode"] == expected_mode
+
+    def test_draft_enhance_drops_the_prompt(self):
+        """FLUX 3 rejects a prompt in draft_enhance mode; the bundle carries it."""
+        data, _, _ = self.config.transform_video_create_request(
+            model="flux-3-video",
+            prompt="a prompt the API would reject here",
+            api_base=API_BASE,
+            video_create_optional_request_params={"draft_cache": "bundle"},
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert data["mode"] == "draft_enhance"
+        assert "prompt" not in data
+
+    def test_explicit_mode_is_not_overridden(self):
+        data, _, _ = self.config.transform_video_create_request(
+            model="flux-3-video",
+            prompt="a prompt",
+            api_base=API_BASE,
+            video_create_optional_request_params={
+                "mode": "i2v",
+                "keyframes": ["https://example.com/first.png"],
+            },
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert data["mode"] == "i2v"
+
+    def test_unknown_model_is_rejected(self):
+        with pytest.raises(ValueError, match="Unknown BFL video model"):
+            self.config.transform_video_create_request(
+                model="flux-9-video",
+                prompt="a prompt",
+                api_base=API_BASE,
+                video_create_optional_request_params={},
+                litellm_params=GenericLiteLLMParams(),
+                headers={},
+            )
+
+    @pytest.mark.parametrize(
+        "size, expected_resolution",
+        [
+            ("1280x720", "hd"),
+            ("1920x1080", "fhd"),
+            ("1080x1920", "fhd"),
+            ("720x1280", "hd"),
+            ("fhd", "fhd"),
+        ],
+    )
+    def test_size_maps_to_a_resolution_tier(self, size, expected_resolution):
+        mapped = self.config.map_openai_params(
+            video_create_optional_params={"size": size},
+            model="flux-3-video",
+            drop_params=False,
+        )
+
+        assert mapped["resolution"] == expected_resolution
+
+    @pytest.mark.parametrize(
+        "seconds, expected_duration",
+        [("8", 8), (8, 8), ("2", 5), ("60", 20), ("8.6", 8)],
+    )
+    def test_seconds_maps_into_the_supported_duration_range(self, seconds, expected_duration):
+        mapped = self.config.map_openai_params(
+            video_create_optional_params={"seconds": seconds},
+            model="flux-3-video",
+            drop_params=False,
+        )
+
+        assert mapped["duration"] == expected_duration
+
+    def test_input_reference_becomes_a_keyframe(self):
+        mapped = self.config.map_openai_params(
+            video_create_optional_params={"input_reference": "https://example.com/first.png"},
+            model="flux-3-video",
+            drop_params=False,
+        )
+
+        assert mapped["keyframes"] == ["https://example.com/first.png"]
+
+    def test_create_response_returns_the_job_handle_and_keeps_the_region(self):
+        video = self.config.transform_video_create_response(
+            model="flux-3-video",
+            raw_response=_response({"id": JOB_ID, "polling_url": POLLING_URL, "cost": None}),
+            logging_obj=self.mock_logging_obj,
+            custom_llm_provider="black_forest_labs",
+            request_data={"duration": 8, "resolution": "fhd"},
+        )
+
+        assert video.status == "queued"
+        assert video.seconds == "8"
+        assert video.size == "fhd"
+
+        # The region survives inside the id, which is all the status call gets.
+        status_url, _ = self.config.transform_video_status_retrieve_request(
+            video_id=video.id,
+            api_base=API_BASE,
+            litellm_params=None,
+            headers={},
+        )
+        assert status_url == POLLING_URL
+
+    def test_status_url_falls_back_to_the_api_base_without_a_region(self):
+        video = self.config.transform_video_create_response(
+            model="flux-3-video",
+            raw_response=_response({"id": JOB_ID, "polling_url": None}),
+            logging_obj=self.mock_logging_obj,
+            custom_llm_provider="black_forest_labs",
+        )
+
+        assert extract_original_video_id(video.id) == JOB_ID
+
+        status_url, _ = self.config.transform_video_status_retrieve_request(
+            video_id=video.id,
+            api_base=API_BASE,
+            litellm_params=None,
+            headers={},
+        )
+        assert status_url == f"{API_BASE}/v1/get_result?id={JOB_ID}"
+
+    def test_content_request_targets_the_same_region_as_the_status_call(self):
+        video = self.config.transform_video_create_response(
+            model="flux-3-video",
+            raw_response=_response({"id": JOB_ID, "polling_url": POLLING_URL}),
+            logging_obj=self.mock_logging_obj,
+            custom_llm_provider="black_forest_labs",
+        )
+
+        content_url, _ = self.config.transform_video_content_request(
+            video_id=video.id,
+            api_base=API_BASE,
+            litellm_params=None,
+            headers={},
+        )
+        assert content_url == POLLING_URL
+
+    def test_a_packed_region_outside_bfl_is_rejected(self):
+        with pytest.raises(BlackForestLabsError, match="not within the bfl.ai domain"):
+            self.config.transform_video_status_retrieve_request(
+                video_id=f"{JOB_ID}@attacker.example.com",
+                api_base=API_BASE,
+                litellm_params=None,
+                headers={},
+            )
+
+    def test_create_response_without_a_job_id_raises(self):
+        with pytest.raises(BlackForestLabsError):
+            self.config.transform_video_create_response(
+                model="flux-3-video",
+                raw_response=_response({"detail": "bad request"}, status_code=422),
+                logging_obj=self.mock_logging_obj,
+            )
+
+    def test_create_response_rejects_a_polling_url_outside_bfl(self):
+        with pytest.raises(BlackForestLabsError, match="not within the bfl.ai domain"):
+            self.config.transform_video_create_response(
+                model="flux-3-video",
+                raw_response=_response(
+                    {"id": JOB_ID, "polling_url": "https://attacker.example.com/v1/get_result"}
+                ),
+                logging_obj=self.mock_logging_obj,
+            )
+
+    @pytest.mark.parametrize(
+        "bfl_status, expected_status",
+        [
+            ("Pending", "queued"),
+            ("Queued", "queued"),
+            ("Reasoning", "in_progress"),
+            ("Generating", "in_progress"),
+            ("Ready", "completed"),
+            ("Error", "failed"),
+            ("Content Moderated", "failed"),
+            ("Task not found", "failed"),
+        ],
+    )
+    def test_bfl_status_maps_to_openai_status(self, bfl_status, expected_status):
+        video = self.config.transform_video_status_retrieve_response(
+            raw_response=_response({"id": JOB_ID, "status": bfl_status, "result": {}}),
+            logging_obj=self.mock_logging_obj,
+        )
+
+        assert video.status == expected_status
+
+    def test_failed_status_carries_the_bfl_detail(self):
+        video = self.config.transform_video_status_retrieve_response(
+            raw_response=_response(
+                {"id": JOB_ID, "status": "Content Moderated", "details": "flagged by moderation"}
+            ),
+            logging_obj=self.mock_logging_obj,
+        )
+
+        assert video.status == "failed"
+        assert video.error["code"] == "Content Moderated"
+        assert video.error["message"] == "flagged by moderation"
+
+    def test_completed_status_has_no_error(self):
+        video = self.config.transform_video_status_retrieve_response(
+            raw_response=_response({"id": JOB_ID, "status": "Ready", "result": {"sample": SAMPLE_URL}}),
+            logging_obj=self.mock_logging_obj,
+        )
+
+        assert video.error is None
+
+    def test_credit_cost_is_reported_as_usage(self):
+        video = self.config.transform_video_status_retrieve_response(
+            raw_response=_response({"id": JOB_ID, "status": "Ready", "cost": 30.0}),
+            logging_obj=self.mock_logging_obj,
+        )
+
+        assert video.usage == {"credits": 30.0}
+
+    def test_fractional_progress_is_reported_as_a_percentage(self):
+        video = self.config.transform_video_status_retrieve_response(
+            raw_response=_response({"id": JOB_ID, "status": "Generating", "progress": 0.42}),
+            logging_obj=self.mock_logging_obj,
+        )
+
+        assert video.progress == 42
+
+    def test_status_request_targets_get_result_with_the_original_job_id(self):
+        encoded_id = self.config.transform_video_create_response(
+            model="flux-3-video",
+            raw_response=_response({"id": JOB_ID, "polling_url": POLLING_URL}),
+            logging_obj=self.mock_logging_obj,
+            custom_llm_provider="black_forest_labs",
+        ).id
+
+        url, params = self.config.transform_video_status_retrieve_request(
+            video_id=encoded_id,
+            api_base="https://api.us7.bfl.ai",
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert url == POLLING_URL
+        assert params == {}
+
+    def test_content_request_targets_get_result(self):
+        url, _ = self.config.transform_video_content_request(
+            video_id=JOB_ID,
+            api_base="https://api.us7.bfl.ai",
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert url == POLLING_URL
+
+    def test_content_response_raises_while_still_generating(self):
+        with pytest.raises(BlackForestLabsError, match="still processing"):
+            self.config.transform_video_content_response(
+                raw_response=_response({"id": JOB_ID, "status": "Generating", "result": {}}),
+                logging_obj=self.mock_logging_obj,
+            )
+
+    def test_content_response_raises_when_a_terminal_job_has_no_video(self):
+        with pytest.raises(BlackForestLabsError, match="did not produce a video"):
+            self.config.transform_video_content_response(
+                raw_response=_response({"id": JOB_ID, "status": "Error", "result": {}}),
+                logging_obj=self.mock_logging_obj,
+            )
+
+    def test_validate_environment_sets_the_x_key_header(self):
+        headers = self.config.validate_environment(
+            headers={},
+            model="flux-3-video",
+            api_key="test-key",
+        )
+
+        assert headers["x-key"] == "test-key"
+        assert headers["Content-Type"] == "application/json"
+
+    def test_validate_environment_without_a_key_raises(self, monkeypatch):
+        monkeypatch.delenv("BFL_API_KEY", raising=False)
+        monkeypatch.delenv("BLACK_FOREST_LABS_API_KEY", raising=False)
+        monkeypatch.setattr("litellm.api_key", None)
+
+        with pytest.raises(BlackForestLabsError, match="BFL_API_KEY is not set"):
+            self.config.validate_environment(headers={}, model="flux-3-video")