feat(videos): serve custom_provider_map BaseVideoConfig providers on /v1/videos (#49785)

* feat(videos): serve custom_provider_map BaseVideoConfig providers on /v1/videos

A provider registered in custom_provider_map with a BaseVideoConfig handler now resolves on every
video call type (create, status, content, list, remix, edit, extension, characters), the same way
CustomLLM handlers already resolve on /v1/images/generations. The video entry points and the HTTP
handler stop casting the provider name to LlmProviders, which raised before any lookup ran, and a
provider without a video config gets a 400 instead of a 500.

* fix(videos): type the custom provider lookups so the type gate holds

Pick the CustomLLM for a custom provider through ProviderConfigManager.get_custom_llm_handler,
which narrows the now wider custom_provider_map handler union, read custom_provider_map instead
of the private _custom_providers list, and annotate the untyped custom_llm_provider parameters
on the video entry points

* test(videos): cover a CustomLLM shadowing a built-in video provider name and type the new tests

Mark custom_handler ReadOnly on CustomLLMItem, type the new tests, and assert that a CustomLLM
registered under openai still leaves openai video on the built-in config

---------

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/images/main.py

@@ -28,7 +28,6 @@ from litellm.litellm_core_utils.mock_functions import mock_image_generation
 from litellm.llms.base_llm import BaseImageEditConfig, BaseImageGenerationConfig
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
 from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
-from litellm.llms.custom_llm import CustomLLM
 from litellm.utils import exception_type, filter_out_litellm_params, get_litellm_params
 
 #################### Initialize provider clients ####################
@@ -608,10 +607,7 @@ def image_generation(
             )
         elif custom_llm_provider in litellm._custom_providers:  # Assume custom LLM provider
             # Get the Custom Handler
-            custom_handler: CustomLLM | None = None
+            custom_handler: Final = ProviderConfigManager.get_custom_llm_handler(custom_llm_provider)
-            for item in litellm.custom_provider_map:
-                if item["provider"] == custom_llm_provider:
-                    custom_handler = item["custom_handler"]
 
             if custom_handler is None:
                 raise LiteLLMUnknownProvider(model=model, custom_llm_provider=custom_llm_provider)
@@ -915,10 +911,7 @@ def image_edit(
 
         # Check for custom provider
         if custom_llm_provider in litellm._custom_providers:
-            custom_handler: CustomLLM | None = None
+            custom_handler: Final = ProviderConfigManager.get_custom_llm_handler(custom_llm_provider)
-            for item in litellm.custom_provider_map:
-                if item["provider"] == custom_llm_provider:
-                    custom_handler = item["custom_handler"]
 
             if custom_handler is None:
                 raise LiteLLMUnknownProvider(model=model, custom_llm_provider=custom_llm_provider)

litellm/llms/custom_httpx/llm_http_handler.py

@@ -8105,7 +8105,7 @@ class BaseLLMHTTPHandler:
         """
         if client is None or not isinstance(client, AsyncHTTPHandler):
             async_httpx_client = get_async_httpx_client(
-                llm_provider=litellm.LlmProviders(custom_llm_provider),
+                llm_provider=custom_llm_provider,
                 params={"ssl_verify": litellm_params.get("ssl_verify", None)},
             )
         else:
@@ -8301,7 +8301,7 @@ class BaseLLMHTTPHandler:
         """
         if client is None or not isinstance(client, AsyncHTTPHandler):
             async_httpx_client = get_async_httpx_client(
-                llm_provider=litellm.LlmProviders(custom_llm_provider),
+                llm_provider=custom_llm_provider,
                 params={"ssl_verify": litellm_params.get("ssl_verify", None)},
             )
         else:
@@ -8482,7 +8482,7 @@ class BaseLLMHTTPHandler:
         """
         if client is None or not isinstance(client, AsyncHTTPHandler):
             async_httpx_client = get_async_httpx_client(
-                llm_provider=litellm.LlmProviders(custom_llm_provider),
+                llm_provider=custom_llm_provider,
                 params={"ssl_verify": litellm_params.get("ssl_verify", None)},
             )
         else:
@@ -8646,7 +8646,7 @@ class BaseLLMHTTPHandler:
     ):
         if client is None or not isinstance(client, AsyncHTTPHandler):
             async_httpx_client = get_async_httpx_client(
-                llm_provider=litellm.LlmProviders(custom_llm_provider),
+                llm_provider=custom_llm_provider,
                 params={"ssl_verify": litellm_params.get("ssl_verify", None)},
             )
         else:
@@ -8787,7 +8787,7 @@ class BaseLLMHTTPHandler:
     ):
         if client is None or not isinstance(client, AsyncHTTPHandler):
             async_httpx_client = get_async_httpx_client(
-                llm_provider=litellm.LlmProviders(custom_llm_provider),
+                llm_provider=custom_llm_provider,
                 params={"ssl_verify": litellm_params.get("ssl_verify", None)},
             )
         else:
@@ -8960,7 +8960,7 @@ class BaseLLMHTTPHandler:
     ):
         if client is None or not isinstance(client, AsyncHTTPHandler):
             async_httpx_client = get_async_httpx_client(
-                llm_provider=litellm.LlmProviders(custom_llm_provider),
+                llm_provider=custom_llm_provider,
                 params={"ssl_verify": litellm_params.get("ssl_verify", None)},
             )
         else:
@@ -9147,7 +9147,7 @@ class BaseLLMHTTPHandler:
     ):
         if client is None or not isinstance(client, AsyncHTTPHandler):
             async_httpx_client = get_async_httpx_client(
-                llm_provider=litellm.LlmProviders(custom_llm_provider),
+                llm_provider=custom_llm_provider,
                 params={"ssl_verify": litellm_params.get("ssl_verify", None)},
             )
         else:
@@ -9280,7 +9280,7 @@ class BaseLLMHTTPHandler:
         """
         if client is None or not isinstance(client, AsyncHTTPHandler):
             async_httpx_client = get_async_httpx_client(
-                llm_provider=litellm.LlmProviders(custom_llm_provider),
+                llm_provider=custom_llm_provider,
                 params={"ssl_verify": litellm_params.get("ssl_verify", None)},
             )
         else:
@@ -9362,7 +9362,7 @@ class BaseLLMHTTPHandler:
         """
         if client is None or not isinstance(client, AsyncHTTPHandler):
             async_httpx_client = get_async_httpx_client(
-                llm_provider=litellm.LlmProviders(custom_llm_provider),
+                llm_provider=custom_llm_provider,
                 params={"ssl_verify": litellm_params.get("ssl_verify", None)},
             )
         else:
@@ -9547,7 +9547,7 @@ class BaseLLMHTTPHandler:
         """
         if client is None or not isinstance(client, AsyncHTTPHandler):
             async_httpx_client = get_async_httpx_client(
-                llm_provider=litellm.LlmProviders(custom_llm_provider),
+                llm_provider=custom_llm_provider,
                 params={"ssl_verify": litellm_params.get("ssl_verify", None)},
             )
         else:

litellm/main.py

@@ -266,7 +266,7 @@ from .llms.codestral.completion.handler import CodestralTextCompletion
 from .llms.cohere.embed import handler as cohere_embed
 from .llms.custom_httpx.aiohttp_handler import BaseLLMAIOHTTPHandler
 from .llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
-from .llms.custom_llm import CustomLLM, custom_chat_llm_router
+from .llms.custom_llm import custom_chat_llm_router
 from .llms.databricks.common_utils import DatabricksBase
 from .llms.databricks.embed.handler import DatabricksEmbeddingHandler
 from .llms.deprecated_providers import aleph_alpha
@@ -5337,10 +5337,7 @@ def _complete_custom_providers(
     stream: Final = ctx.stream
     timeout: Final = ctx.timeout
 
-    custom_handler: CustomLLM | None = None
+    custom_handler: Final = ProviderConfigManager.get_custom_llm_handler(custom_llm_provider)
-    for item in litellm.custom_provider_map:
-        if item["provider"] == custom_llm_provider:
-            custom_handler = item["custom_handler"]
 
     if custom_handler is None:
         raise LiteLLMUnknownProvider(model=model, custom_llm_provider=custom_llm_provider)
@@ -7911,10 +7908,7 @@ def embedding(
                 litellm_params={},
             )
         elif custom_llm_provider in litellm._custom_providers:
-            custom_handler: CustomLLM | None = None
+            custom_handler: Final = ProviderConfigManager.get_custom_llm_handler(custom_llm_provider)
-            for item in litellm.custom_provider_map:
-                if item["provider"] == custom_llm_provider:
-                    custom_handler = item["custom_handler"]
 
             if custom_handler is None:
                 raise LiteLLMUnknownProvider(model=model, custom_llm_provider=custom_llm_provider)

litellm/types/llms/custom_llm.py

@@ -1,11 +1,12 @@
 from typing import TYPE_CHECKING
 
-from typing_extensions import TypedDict
+from typing_extensions import ReadOnly, TypedDict
 
 if TYPE_CHECKING:
+    from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
     from litellm.llms.custom_llm import CustomLLM
 
 
 class CustomLLMItem(TypedDict):
     provider: str
-    custom_handler: "CustomLLM"
+    custom_handler: ReadOnly["CustomLLM | BaseVideoConfig"]

litellm/utils.py

@@ -467,6 +467,7 @@ if TYPE_CHECKING:
     )
     from litellm.llms.cohere.common_utils import CohereModelInfo
     from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
+    from litellm.llms.custom_llm import CustomLLM
     from litellm.types.llms.anthropic import (
         ANTHROPIC_API_ONLY_HEADERS,
         AnthropicThinkingParam,
@@ -10958,8 +10959,13 @@ class ProviderConfigManager:
     @staticmethod
     def get_provider_video_config(
         model: str | None,
-        provider: LlmProviders,
+        provider: LlmProviders | str,
     ) -> BaseVideoConfig | None:
+        if not isinstance(provider, LlmProviders):
+            custom_video_config: Final = ProviderConfigManager.get_custom_provider_video_config(provider)
+            if custom_video_config is not None or provider not in LlmProvidersSet:
+                return custom_video_config
+            return ProviderConfigManager.get_provider_video_config(model=model, provider=LlmProviders(provider))
         if LlmProviders.OPENAI == provider:
             from litellm.llms.openai.videos.transformation import OpenAIVideoConfig
 
@@ -11010,6 +11016,34 @@ class ProviderConfigManager:
             return VolcEngineVideoConfig()
         return None
 
+    @staticmethod
+    def get_custom_llm_handler(provider: str | None) -> CustomLLM | None:
+        """The `CustomLLM` registered for `provider` in `litellm.custom_provider_map`, if its handler is one."""
+        from litellm.llms.custom_llm import CustomLLM
+
+        return next(
+            (
+                item["custom_handler"]
+                for item in litellm.custom_provider_map
+                if item["provider"] == provider and isinstance(item["custom_handler"], CustomLLM)
+            ),
+            None,
+        )
+
+    @staticmethod
+    def get_custom_provider_video_config(provider: str) -> BaseVideoConfig | None:
+        """The `BaseVideoConfig` registered for `provider` in `litellm.custom_provider_map`, if its handler is one."""
+        from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
+
+        return next(
+            (
+                item["custom_handler"]
+                for item in litellm.custom_provider_map
+                if item["provider"] == provider and isinstance(item["custom_handler"], BaseVideoConfig)
+            ),
+            None,
+        )
+
     @staticmethod
     def get_provider_container_config(
         provider: LlmProviders,

litellm/videos/main.py

@@ -26,6 +26,17 @@ from litellm.types.videos.utils import decode_video_id_with_provider
 from litellm.utils import ProviderConfigManager, client
 from litellm.videos.utils import VideoGenerationRequestUtils
 
+
+def _unsupported_video_call(
+    call_name: str, custom_llm_provider: str, model: str | None = None
+) -> litellm.BadRequestError:
+    return litellm.BadRequestError(
+        message=f"{call_name} is not supported for {custom_llm_provider}",
+        model=model or "",
+        llm_provider=custom_llm_provider,
+    )
+
+
 #################### Initialize provider clients ####################
 llm_http_handler: BaseLLMHTTPHandler = BaseLLMHTTPHandler()
 
@@ -40,7 +51,7 @@ async def avideo_generation(
     size: str | None = None,
     user: str | None = None,
     timeout=600,  # default to 10 minutes
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
     # The extra values given here take precedence over values defined on the client or passed to this method.
     extra_headers: dict[str, object] | None = None,
@@ -170,7 +181,7 @@ def video_generation(
     size: str | None = None,
     user: str | None = None,
     timeout=600,  # default to 10 minutes
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
     # The extra values given here take precedence over values defined on the client or passed to this method.
     extra_headers: dict[str, object] | None = None,
@@ -208,11 +219,11 @@ def video_generation(
         # get provider config
         video_generation_provider_config: BaseVideoConfig | None = ProviderConfigManager.get_provider_video_config(
             model=model,
-            provider=litellm.LlmProviders(custom_llm_provider),
+            provider=custom_llm_provider,
         )
 
         if video_generation_provider_config is None:
-            raise ValueError(f"video generation is not supported for {custom_llm_provider}")
+            raise _unsupported_video_call("video generation", custom_llm_provider, model)
 
         local_vars.update(kwargs)
         # Get VideoGenerationOptionalRequestParams with only valid parameters
@@ -329,11 +340,11 @@ def video_content(
         # get provider config
         video_provider_config: Final[BaseVideoConfig | None] = ProviderConfigManager.get_provider_video_config(
             model=None,
-            provider=litellm.LlmProviders(custom_llm_provider),
+            provider=custom_llm_provider,
         )
 
         if video_provider_config is None:
-            raise ValueError(f"video support download is not supported for {custom_llm_provider}")
+            raise _unsupported_video_call("video support download", custom_llm_provider)
 
         local_vars.update(kwargs)
         # For video content download, we don't need complex optional parameter handling
@@ -456,7 +467,7 @@ async def avideo_remix(
     video_id: str,
     prompt: str,
     timeout=600,  # default to 10 minutes
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
     # The extra values given here take precedence over values defined on the client or passed to this method.
     extra_headers: dict[str, object] | None = None,
@@ -559,7 +570,7 @@ def video_remix(
     video_id: str,
     prompt: str,
     timeout=600,  # default to 10 minutes
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
     # The extra values given here take precedence over values defined on the client or passed to this method.
     extra_headers: dict[str, object] | None = None,
@@ -598,11 +609,11 @@ def video_remix(
         # get provider config
         video_remix_provider_config: Final[BaseVideoConfig | None] = ProviderConfigManager.get_provider_video_config(
             model=None,
-            provider=litellm.LlmProviders(custom_llm_provider),
+            provider=custom_llm_provider,
         )
 
         if video_remix_provider_config is None:
-            raise ValueError(f"video remix is not supported for {custom_llm_provider}")
+            raise _unsupported_video_call("video remix", custom_llm_provider)
 
         local_vars.update(kwargs)
         # For video remix, we need the video_id and prompt
@@ -660,7 +671,7 @@ async def avideo_list(
     order: str | None = None,
     api_key: str | None = None,
     timeout=600,  # default to 10 minutes
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
     # The extra values given here take precedence over values defined on the client or passed to this method.
     extra_headers: dict[str, object] | None = None,
@@ -774,7 +785,7 @@ def video_list(
     limit: int | None = None,
     order: str | None = None,
     timeout=600,  # default to 10 minutes
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
     # The extra values given here take precedence over values defined on the client or passed to this method.
     extra_headers: dict[str, object] | None = None,
@@ -810,11 +821,11 @@ def video_list(
         # get provider config
         video_list_provider_config: Final[BaseVideoConfig | None] = ProviderConfigManager.get_provider_video_config(
             model=None,
-            provider=litellm.LlmProviders(custom_llm_provider),
+            provider=custom_llm_provider,
         )
 
         if video_list_provider_config is None:
-            raise ValueError(f"video list is not supported for {custom_llm_provider}")
+            raise _unsupported_video_call("video list", custom_llm_provider)
 
         local_vars.update(kwargs)
         # For video list, we need the query parameters
@@ -871,7 +882,7 @@ def video_list(
 async def avideo_status(
     video_id: str,
     timeout=600,  # default to 10 minutes
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
     # The extra values given here take precedence over values defined on the client or passed to this method.
     extra_headers: dict[str, object] | None = None,
@@ -970,7 +981,7 @@ def video_status(
 def video_status(
     video_id: str,
     timeout=600,  # default to 10 minutes
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
     # The extra values given here take precedence over values defined on the client or passed to this method.
     extra_headers: dict[str, object] | None = None,
@@ -1031,11 +1042,11 @@ def video_status(
         # get provider config
         video_status_provider_config: Final[BaseVideoConfig | None] = ProviderConfigManager.get_provider_video_config(
             model=None,
-            provider=litellm.LlmProviders(custom_llm_provider),
+            provider=custom_llm_provider,
         )
 
         if video_status_provider_config is None:
-            raise ValueError(f"video status is not supported for {custom_llm_provider}")
+            raise _unsupported_video_call("video status", custom_llm_provider)
 
         local_vars.update(kwargs)
         # For video status, we need the video_id
@@ -1088,7 +1099,7 @@ async def avideo_create_character(
     name: str,
     video: FileTypes,
     timeout=600,
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     extra_headers: dict[str, object] | None = None,
     extra_query: dict[str, object] | None = None,
     extra_body: dict[str, object] | None = None,
@@ -1143,7 +1154,7 @@ def video_create_character(
     name: str,
     video: FileTypes,
     timeout=600,
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     extra_headers: dict[str, object] | None = None,
     extra_query: dict[str, object] | None = None,
     extra_body: dict[str, object] | None = None,
@@ -1172,11 +1183,11 @@ def video_create_character(
 
         provider_config: Final[BaseVideoConfig | None] = ProviderConfigManager.get_provider_video_config(
             model=None,
-            provider=litellm.LlmProviders(custom_llm_provider),
+            provider=custom_llm_provider,
         )
 
         if provider_config is None:
-            raise ValueError(f"video create character is not supported for {custom_llm_provider}")
+            raise _unsupported_video_call("video create character", custom_llm_provider)
 
         local_vars.update(kwargs)
         request_params: Final[dict] = {"name": name}
@@ -1218,7 +1229,7 @@ def video_create_character(
 async def avideo_get_character(
     character_id: str,
     timeout=600,
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     extra_headers: dict[str, object] | None = None,
     extra_query: dict[str, object] | None = None,
     extra_body: dict[str, object] | None = None,
@@ -1268,7 +1279,7 @@ async def avideo_get_character(
 def video_get_character(
     character_id: str,
     timeout=600,
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     extra_headers: dict[str, object] | None = None,
     extra_query: dict[str, object] | None = None,
     extra_body: dict[str, object] | None = None,
@@ -1297,11 +1308,11 @@ def video_get_character(
 
         provider_config: Final[BaseVideoConfig | None] = ProviderConfigManager.get_provider_video_config(
             model=None,
-            provider=litellm.LlmProviders(custom_llm_provider),
+            provider=custom_llm_provider,
         )
 
         if provider_config is None:
-            raise ValueError(f"video get character is not supported for {custom_llm_provider}")
+            raise _unsupported_video_call("video get character", custom_llm_provider)
 
         local_vars.update(kwargs)
         request_params: Final[dict] = {"character_id": character_id}
@@ -1343,7 +1354,7 @@ async def avideo_edit(
     video_id: str,
     prompt: str,
     timeout=600,
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     extra_headers: dict[str, object] | None = None,
     extra_query: dict[str, object] | None = None,
     extra_body: dict[str, object] | None = None,
@@ -1398,7 +1409,7 @@ def video_edit(
     video_id: str,
     prompt: str,
     timeout=600,
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     extra_headers: dict[str, object] | None = None,
     extra_query: dict[str, object] | None = None,
     extra_body: dict[str, object] | None = None,
@@ -1430,11 +1441,11 @@ def video_edit(
 
         provider_config: Final[BaseVideoConfig | None] = ProviderConfigManager.get_provider_video_config(
             model=None,
-            provider=litellm.LlmProviders(custom_llm_provider),
+            provider=custom_llm_provider,
         )
 
         if provider_config is None:
-            raise ValueError(f"video edit is not supported for {custom_llm_provider}")
+            raise _unsupported_video_call("video edit", custom_llm_provider)
 
         local_vars.update(kwargs)
         request_params: Final[dict] = {"video_id": video_id, "prompt": prompt}
@@ -1480,7 +1491,7 @@ async def avideo_extension(
     prompt: str,
     seconds: str,
     timeout=600,
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     extra_headers: dict[str, object] | None = None,
     extra_query: dict[str, object] | None = None,
     extra_body: dict[str, object] | None = None,
@@ -1534,7 +1545,7 @@ def video_extension(
     prompt: str,
     seconds: str,
     timeout=600,
-    custom_llm_provider=None,
+    custom_llm_provider: str | None = None,
     extra_headers: dict[str, object] | None = None,
     extra_query: dict[str, object] | None = None,
     extra_body: dict[str, object] | None = None,
@@ -1564,11 +1575,11 @@ def video_extension(
 
         provider_config: Final[BaseVideoConfig | None] = ProviderConfigManager.get_provider_video_config(
             model=None,
-            provider=litellm.LlmProviders(custom_llm_provider),
+            provider=custom_llm_provider,
         )
 
         if provider_config is None:
-            raise ValueError(f"video extension is not supported for {custom_llm_provider}")
+            raise _unsupported_video_call("video extension", custom_llm_provider)
 
         local_vars.update(kwargs)
         request_params: Final[dict] = {

tests/integration/_support/custom_video_provider.py

@@ -0,0 +1,181 @@
+"""A customer-owned video provider for the proxy's `custom_provider_map`: a job queue with its own wire shape."""
+
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import Final
+
+import httpx
+from httpx._types import RequestFiles
+
+from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
+from litellm.llms.custom_llm import CustomLLM
+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
+
+_STATUS_BY_STATE: Final = MappingProxyType({"queued": "queued", "running": "in_progress", "done": "completed"})
+
+
+def _video_object(job: Mapping[str, object], model: str | None, custom_llm_provider: str | None) -> VideoObject:
+    job_id: Final = str(job["job_id"])
+    seconds: Final = job.get("seconds")
+    video: Final = VideoObject(
+        id=encode_video_id_with_provider(job_id, custom_llm_provider, model) if custom_llm_provider else job_id,
+        object="video",
+        status=_STATUS_BY_STATE[str(job["state"])],
+        created_at=0,
+        model=model,
+        seconds=str(seconds) if seconds is not None else None,
+    )
+    return video
+
+
+class QueueVideoConfig(BaseVideoConfig):
+    def get_supported_openai_params(self, model: str) -> list:  # mutable-ok: BaseVideoConfig contract
+        return ["seconds"]
+
+    def map_openai_params(
+        self,
+        video_create_optional_params: VideoCreateOptionalRequestParams,
+        model: str,
+        drop_params: bool,
+    ) -> dict:  # mutable-ok: BaseVideoConfig contract
+        return {key: value for key, value in video_create_optional_params.items() if key == "seconds"}
+
+    def validate_environment(
+        self,
+        headers: dict,  # mutable-ok: BaseVideoConfig contract
+        model: str,
+        api_key: str | None = None,
+        litellm_params: GenericLiteLLMParams | None = None,
+    ) -> dict:  # mutable-ok: BaseVideoConfig contract
+        key: Final = (litellm_params.api_key if litellm_params is not None else None) or api_key or ""
+        return {**headers, "X-Api-Key": key}
+
+    def get_complete_url(
+        self,
+        model: str,
+        api_base: str | None,
+        litellm_params: dict,  # mutable-ok: BaseVideoConfig contract
+    ) -> str:
+        if api_base is None:
+            raise ValueError("api_base is required")
+        return f"{api_base.rstrip('/')}/jobs"
+
+    def transform_video_create_request(
+        self,
+        model: str,
+        prompt: str,
+        api_base: str,
+        video_create_optional_request_params: dict,  # mutable-ok: BaseVideoConfig contract
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig contract
+    ) -> tuple[dict, RequestFiles, str]:  # mutable-ok: BaseVideoConfig contract
+        seconds: Final = video_create_optional_request_params.get("seconds")
+        return ({"model": model, "prompt": prompt, "seconds": seconds}, [], api_base)
+
+    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,
+    ) -> VideoObject:
+        return _video_object(raw_response.json(), model, custom_llm_provider)
+
+    def transform_video_status_retrieve_request(
+        self,
+        video_id: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig contract
+    ) -> tuple[str, dict]:  # mutable-ok: BaseVideoConfig contract
+        return (f"{api_base}/{extract_original_video_id(video_id)}", {})
+
+    def transform_video_status_retrieve_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        custom_llm_provider: str | None = None,
+        client: object = None,
+    ) -> VideoObject:
+        return _video_object(raw_response.json(), None, custom_llm_provider)
+
+    def transform_video_content_request(
+        self,
+        video_id: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig contract
+        variant: str | None = None,
+    ) -> tuple[str, dict]:  # mutable-ok: BaseVideoConfig contract
+        return (f"{api_base}/{extract_original_video_id(video_id)}/file", {})
+
+    def transform_video_content_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        client: object = None,
+    ) -> bytes:
+        return raw_response.content
+
+    def transform_video_list_request(
+        self,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig contract
+        after: str | None = None,
+        limit: int | None = None,
+        order: str | None = None,
+        extra_query: dict[str, object] | None = None,
+    ) -> tuple[str, dict]:  # mutable-ok: BaseVideoConfig contract
+        return (api_base, {})
+
+    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 contract
+        return raw_response.json()
+
+    def transform_video_remix_request(
+        self,
+        video_id: str,
+        prompt: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig contract
+        extra_body: dict[str, object] | None = None,
+    ) -> tuple[str, dict]:  # mutable-ok: BaseVideoConfig contract
+        raise NotImplementedError("the queue backend has no remix")
+
+    def transform_video_remix_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        custom_llm_provider: str | None = None,
+    ) -> VideoObject:
+        raise NotImplementedError("the queue backend has no remix")
+
+    def transform_video_delete_request(
+        self,
+        video_id: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig contract
+    ) -> tuple[str, dict]:  # mutable-ok: BaseVideoConfig contract
+        raise NotImplementedError("the queue backend has no delete")
+
+    def transform_video_delete_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+    ) -> VideoObject:
+        raise NotImplementedError("the queue backend has no delete")
+
+
+VIDEO_CONFIG: Final = QueueVideoConfig()
+IMAGE_ONLY_HANDLER: Final = CustomLLM()

tests/integration/providers/test_custom_video_provider_wire.py

@@ -0,0 +1,113 @@
+"""A provider registered in `custom_provider_map` with a `BaseVideoConfig` serves /v1/videos create, status and content."""
+
+import json
+import shutil
+from contextlib import ExitStack
+from pathlib import Path
+from typing import Final
+
+from integration._support.client import Gateway
+from integration._support.process import owned_proxy_process
+from integration._support.wire import Reply, Request, Wire, wire_server
+
+_SUPPORT_MODULE: Final = Path(__file__).resolve().parents[1] / "_support" / "custom_video_provider.py"
+_PROMPT: Final = "a cat walking across a bridge"
+_MP4: Final = b"\x00\x00\x00\x18ftypmp42" + b"queue-video-bytes"
+
+
+def _queue_backend(request: Request) -> Reply:
+    if (request.method, request.target) == ("POST", "/jobs"):
+        return Reply(body=json.dumps({"job_id": "job-7", "state": "queued"}).encode())
+    if (request.method, request.target) == ("GET", "/jobs/job-7"):
+        return Reply(body=json.dumps({"job_id": "job-7", "state": "done", "seconds": 8}).encode())
+    if (request.method, request.target) == ("GET", "/jobs/job-7/file"):
+        return Reply(body=_MP4, content_type="video/mp4")
+    return Reply(status=404, body=json.dumps({"error": f"no route {request.method} {request.target}"}).encode())
+
+
+def _write_config(path: Path, wire: Wire) -> Path:
+    shutil.copy(_SUPPORT_MODULE, path.parent / "custom_video_provider.py")
+    path.write_text(
+        json.dumps(
+            {
+                "model_list": [
+                    {
+                        "model_name": "queue-video",
+                        "litellm_params": {
+                            "model": "queue-video/seedance-1",
+                            "api_base": wire.url,
+                            "api_key": "queue-secret",
+                        },
+                    },
+                    {
+                        "model_name": "image-only-video",
+                        "litellm_params": {
+                            "model": "image-only/seedance-1",
+                            "api_base": wire.url,
+                            "api_key": "unused",
+                        },
+                    },
+                ],
+                "litellm_settings": {
+                    "custom_provider_map": [
+                        {"provider": "queue-video", "custom_handler": "custom_video_provider.VIDEO_CONFIG"},
+                        {"provider": "image-only", "custom_handler": "custom_video_provider.IMAGE_ONLY_HANDLER"},
+                    ]
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+    return path
+
+
+def _lower_headers(request: Request) -> dict[str, str]:
+    return {name.lower(): value for name, value in request.headers.items()}
+
+
+def test_custom_provider_map_base_video_config_serves_create_status_and_content(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    with ExitStack() as stack:
+        wire: Final = stack.enter_context(wire_server(_queue_backend))
+        config: Final = _write_config(tmp_path / "custom-video.yaml", wire)
+        owned: Final = stack.enter_context(owned_proxy_process(gateway, tmp_path, {}, config=config))
+
+        created: Final = owned.gateway.request(
+            "POST", "/v1/videos", {"model": "queue-video", "prompt": _PROMPT, "seconds": "8"}
+        )
+        assert created.status_code == 200, created.text
+        created_body: Final = created.json()
+        assert created_body["status"] == "queued", created_body
+        video_id: Final = created_body["id"]
+        assert isinstance(video_id, str) and video_id.startswith("video_"), created_body
+
+        submits: Final = wire.drain()
+        assert [(r.method, r.target) for r in submits] == [("POST", "/jobs")], submits
+        assert json.loads(submits[0].body) == {"model": "seedance-1", "prompt": _PROMPT, "seconds": "8"}
+        assert _lower_headers(submits[0])["x-api-key"] == "queue-secret"
+
+        status: Final = owned.gateway.request("GET", f"/v1/videos/{video_id}")
+        assert status.status_code == 200, status.text
+        assert status.json()["status"] == "completed", status.text
+        assert status.json()["id"] == video_id, status.text
+
+        content: Final = owned.gateway.request("GET", f"/v1/videos/{video_id}/content")
+        assert content.status_code == 200, content.text
+        assert content.content == _MP4
+
+        polls: Final = wire.drain()
+        assert [(r.method, r.target) for r in polls] == [("GET", "/jobs/job-7"), ("GET", "/jobs/job-7/file")], polls
+        assert all(_lower_headers(r)["x-api-key"] == "queue-secret" for r in polls)
+
+
+def test_custom_provider_without_a_video_config_is_refused_with_400(gateway: Gateway, tmp_path: Path) -> None:
+    with ExitStack() as stack:
+        wire: Final = stack.enter_context(wire_server(_queue_backend))
+        config: Final = _write_config(tmp_path / "custom-video.yaml", wire)
+        owned: Final = stack.enter_context(owned_proxy_process(gateway, tmp_path, {}, config=config))
+
+        refused: Final = owned.gateway.request("POST", "/v1/videos", {"model": "image-only-video", "prompt": _PROMPT})
+        assert refused.status_code == 400, refused.text
+        assert "video generation is not supported for image-only" in refused.text
+        assert wire.drain() == ()

tests/unit/test_utils.py

@@ -967,21 +967,11 @@ _COST_MAP_ROW_SCHEMA: Final = {
             "cache_creation_input_token_cost": {"type": "number"},
             "cache_creation_input_token_cost_above_1hr": {"type": "number"},
             "cache_creation_input_token_cost_above_12hr": {"type": "number"},
-            "cache_creation_input_token_cost_above_12hr_above_272k_tokens": {
+            "cache_creation_input_token_cost_above_12hr_above_272k_tokens": {"type": "number"},
-                "type": "number"
+            "cache_creation_input_token_cost_above_12hr_above_272k_tokens_flex": {"type": "number"},
-            },
+            "cache_creation_input_token_cost_above_12hr_above_272k_tokens_priority": {"type": "number"},
-            "cache_creation_input_token_cost_above_12hr_above_272k_tokens_flex": {
+            "cache_creation_input_token_cost_above_12hr_above_272k_tokens_batches": {"type": "number"},
-                "type": "number"
+            "cache_creation_input_token_cost_above_12hr_above_272k_tokens_ultrafast": {"type": "number"},
-            },
-            "cache_creation_input_token_cost_above_12hr_above_272k_tokens_priority": {
-                "type": "number"
-            },
-            "cache_creation_input_token_cost_above_12hr_above_272k_tokens_batches": {
-                "type": "number"
-            },
-            "cache_creation_input_token_cost_above_12hr_above_272k_tokens_ultrafast": {
-                "type": "number"
-            },
             "cache_creation_input_token_cost_above_12hr_flex": {"type": "number"},
             "cache_creation_input_token_cost_above_12hr_priority": {"type": "number"},
             "cache_creation_input_token_cost_above_12hr_batches": {"type": "number"},
@@ -1671,6 +1661,38 @@ def test_get_provider_rerank_config():
     assert isinstance(config, HostedVLLMRerankConfig)
 
 
+def test_get_provider_video_config_resolves_a_custom_provider_map_base_video_config(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A `BaseVideoConfig` registered in `custom_provider_map` answers for its provider name, a `CustomLLM`
+    registered there does not, and a built-in name still resolves to its built-in config even when a `CustomLLM`
+    shadows it in the map."""
+    from litellm.llms.custom_llm import CustomLLM
+    from litellm.llms.openai.videos.transformation import OpenAIVideoConfig
+
+    class QueueVideoConfig(OpenAIVideoConfig):
+        pass
+
+    video_config: Final = QueueVideoConfig()
+    monkeypatch.setattr(
+        litellm,
+        "custom_provider_map",
+        [
+            {"provider": "queue-video", "custom_handler": video_config},
+            {"provider": "image-only", "custom_handler": CustomLLM()},
+            {"provider": "openai", "custom_handler": CustomLLM()},
+        ],
+    )
+    monkeypatch.setattr(litellm, "_custom_providers", ["queue-video", "image-only", "openai"])
+
+    assert ProviderConfigManager.get_provider_video_config(model="seedance-1", provider="queue-video") is video_config
+    assert ProviderConfigManager.get_provider_video_config(model="seedance-1", provider="image-only") is None
+    assert isinstance(
+        ProviderConfigManager.get_provider_video_config(model="sora-2", provider="openai"), OpenAIVideoConfig
+    )
+    assert ProviderConfigManager.get_provider_video_config(model="seedance-1", provider="not-registered") is None
+
+
 def test_get_provider_text_to_speech_config_vertex_gemini_skips_cloud_tts():
     """Regression for LIT-6501: mapping vertex Gemini TTS params through Google Cloud TTS
     dropped response_format before the speech_to_completion bridge could honor it."""
@@ -7572,9 +7594,10 @@ async def test_oidc_google_api_key_fails_closed_when_the_metadata_server_fails_a
             side_effect=[metadata_failure(), httpx.Response(200, text=recovered_token)]
         )
         chat: Final = upstream.post(_VLLM_CHAT_URL).respond(200, json=_VLLM_REPLY)
-        with caplog.at_level(logging.WARNING, logger="LiteLLM"), pytest.raises(
+        with (
-            (litellm.AuthenticationError, litellm.ServiceUnavailableError)
+            caplog.at_level(logging.WARNING, logger="LiteLLM"),
-        ) as excinfo:
+            pytest.raises((litellm.AuthenticationError, litellm.ServiceUnavailableError)) as excinfo,
+        ):
             await _ping_vllm_through(call_path, "vllm-sa")
         sent_while_failing: Final = chat.call_count
         await _ping_vllm_through(call_path, "vllm-sa")
@@ -7823,9 +7846,7 @@ async def test_router_resolves_an_oidc_api_key_only_on_a_proxy_level_deployment(
     try:
         with respx.mock(assert_all_called=True) as upstream:
             upstream.post(_MISTRAL_CHAT_URL).respond(200, json=_MISTRAL_REPLY)
-            await router.acompletion(
+            await router.acompletion(model="devstral", messages=_PING, metadata={"user_api_key_team_id": "team-1"})
-                model="devstral", messages=_PING, metadata={"user_api_key_team_id": "team-1"}
-            )
             sent: Final = upstream.calls.last.request.headers["Authorization"]
     finally:
         router.discard()
@@ -7893,7 +7914,9 @@ async def test_an_inline_oidc_api_key_reaches_the_endpoint_its_deployment_config
         with respx.mock(assert_all_called=False) as upstream:
             upstream.route().respond(200, json=_BEDROCK_REPLY)
             await router.acompletion(
-                model="claude", messages=_PING, aws_bedrock_runtime_endpoint="https://BEDROCK-runtime.us-east-1.amazonaws.com"
+                model="claude",
+                messages=_PING,
+                aws_bedrock_runtime_endpoint="https://BEDROCK-runtime.us-east-1.amazonaws.com",
             )
             sent: Final = tuple(
                 (call.request.url.host, call.request.headers["Authorization"]) for call in upstream.calls
@@ -8102,7 +8125,9 @@ async def test_a_team_row_naming_an_oidc_credential_reaches_only_the_credentials
         (
             lambda reference: LiteLLM_Params(
                 **litellm.Router(
-                    model_list=[{"model_name": "openai/*", "litellm_params": {"model": "openai/*", "api_key": reference}}]
+                    model_list=[
+                        {"model_name": "openai/*", "litellm_params": {"model": "openai/*", "api_key": reference}}
+                    ]
                 ).get_model_list(model_name="openai/*")[0]["litellm_params"]
             ),
             lambda reference: "Bearer listing-token",
@@ -9108,7 +9133,9 @@ def test_basic_trimming_no_max_tokens_specified():
     trimmed_messages = trim_messages(messages, model="gpt-4")
     print("trimmed messages for gpt-4")
     print(trimmed_messages)
-    assert (get_token_count(messages=trimmed_messages, model="gpt-4")) <= litellm.model_cost["gpt-4"]["max_input_tokens"]
+    assert (get_token_count(messages=trimmed_messages, model="gpt-4")) <= litellm.model_cost["gpt-4"][
+        "max_input_tokens"
+    ]
 
 
 def test_multiple_messages_trimming():
@@ -11427,7 +11454,9 @@ def test_declared_mid_conversation_system_support_prefers_the_deployment_over_th
     )
 
 
-def test_declared_mid_conversation_system_support_is_none_when_the_model_lookup_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_declared_mid_conversation_system_support_is_none_when_the_model_lookup_raises(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
     monkeypatch.setitem(litellm.model_cost, "hosted_vllm/unreadable-probe", "not a model info entry")
     litellm.get_model_info.cache_clear()
     assert litellm.utils.declared_mid_conversation_system_support("unreadable-probe", "hosted_vllm", None) is None
@@ -11448,7 +11477,6 @@ def test_get_optional_params_forwards_a_tool_config_cache_point_to_a_downstream_
     assert optional_params["extra_body"] == {"kept": True, "cache_control_injection_points": [point]}
 
 
-
 @pytest.mark.parametrize(
     ("provider", "expected_warnings"),
     (("streamlake", 1), ("veniceai", 1), ("my-own-llm", 0)),

tests/unit/videos/test_main.py

@@ -32,7 +32,7 @@ reflect production.
 
 from contextlib import ExitStack
 from dataclasses import dataclass
-from typing import Any, Dict
+from typing import Any, Dict, Final
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -205,9 +205,7 @@ def test_video_edit__dispatch_and_provider_from_id(seams):
 
 
 def test_video_extension__dispatch_and_provider_from_id(seams):
-    result = videos_main.video_extension(
+    result = videos_main.video_extension(video_id=AZURE_VIDEO_ID, prompt="continue", seconds="5")
-        video_id=AZURE_VIDEO_ID, prompt="continue", seconds="5"
-    )
 
     seams.assert_only("video_extension_handler")
     assert result is seams.handler.video_extension_handler.return_value
@@ -285,9 +283,7 @@ def test_generation__mock_response_short_circuits(seams):
 
 
 def test_list__mock_response_short_circuits(seams):
-    resp = videos_main.video_list(
+    resp = videos_main.video_list(mock_response=[{"id": "v1", "object": "video", "status": "completed"}])
-        mock_response=[{"id": "v1", "object": "video", "status": "completed"}]
-    )
 
     assert isinstance(resp, list)
     assert resp[0].id == "v1"
@@ -318,7 +314,7 @@ def test_get_character__mock_response_short_circuits(seams):
 def test_unsupported_provider_raises_without_dispatch(seams):
     seams.get_config.return_value = None
 
-    with pytest.raises(litellm.APIConnectionError):
+    with pytest.raises(litellm.BadRequestError, match="video status is not supported for azure"):
         videos_main.video_status(video_id=AZURE_VIDEO_ID)
 
     seams.handler.video_status_handler.assert_not_called()
@@ -333,9 +329,7 @@ def test_unsupported_provider_raises_without_dispatch(seams):
 async def test_avideo_generation__delegates_with_async_flag():
     sentinel = VideoObject(id="v-async", object="video", status="queued")
     with (
-        patch.object(
+        patch.object(videos_main, "video_generation", MagicMock(return_value=sentinel)) as sync,
-            videos_main, "video_generation", MagicMock(return_value=sentinel)
-        ) as sync,
         patch.object(
             litellm,
             "get_llm_provider",
@@ -352,9 +346,7 @@ async def test_avideo_generation__delegates_with_async_flag():
 @pytest.mark.asyncio
 async def test_avideo_status__delegates_untouched():
     sentinel = VideoObject(id="v-async", object="video", status="queued")
-    with patch.object(
+    with patch.object(videos_main, "video_status", MagicMock(return_value=sentinel)) as sync:
-        videos_main, "video_status", MagicMock(return_value=sentinel)
-    ) as sync:
         result = await videos_main.avideo_status(video_id="video_plain")
 
     assert result is sentinel
@@ -367,9 +359,7 @@ async def test_avideo_content__pre_decodes_provider_before_delegating():
     """avideo_content resolves the provider from the encoded id itself before
     handing off, so the sync worker receives the decoded provider, not None."""
     sentinel = b"mp4-bytes"
-    with patch.object(
+    with patch.object(videos_main, "video_content", MagicMock(return_value=sentinel)) as sync:
-        videos_main, "video_content", MagicMock(return_value=sentinel)
-    ) as sync:
         result = await videos_main.avideo_content(video_id=AZURE_VIDEO_ID)
 
     assert result is sentinel
@@ -393,9 +383,7 @@ DB_YAML_CREDS = {
 CREDENTIAL_OPERATIONS = [
     (
         "video_generation_handler",
-        lambda: videos_main.video_generation(
+        lambda: videos_main.video_generation(prompt="p", model="sora-2", **DB_YAML_CREDS),
-            prompt="p", model="sora-2", **DB_YAML_CREDS
-        ),
     ),
     (
         "video_status_handler",
@@ -407,21 +395,15 @@ CREDENTIAL_OPERATIONS = [
     ),
     (
         "video_remix_handler",
-        lambda: videos_main.video_remix(
+        lambda: videos_main.video_remix(video_id=AZURE_VIDEO_ID, prompt="p", **DB_YAML_CREDS),
-            video_id=AZURE_VIDEO_ID, prompt="p", **DB_YAML_CREDS
-        ),
     ),
     (
         "video_edit_handler",
-        lambda: videos_main.video_edit(
+        lambda: videos_main.video_edit(video_id=AZURE_VIDEO_ID, prompt="p", **DB_YAML_CREDS),
-            video_id=AZURE_VIDEO_ID, prompt="p", **DB_YAML_CREDS
-        ),
     ),
     (
         "video_extension_handler",
-        lambda: videos_main.video_extension(
+        lambda: videos_main.video_extension(video_id=AZURE_VIDEO_ID, prompt="p", seconds="5", **DB_YAML_CREDS),
-            video_id=AZURE_VIDEO_ID, prompt="p", seconds="5", **DB_YAML_CREDS
-        ),
     ),
     (
         "video_list_handler",
@@ -429,9 +411,7 @@ CREDENTIAL_OPERATIONS = [
     ),
     (
         "video_create_character_handler",
-        lambda: videos_main.video_create_character(
+        lambda: videos_main.video_create_character(name="hero", video=MagicMock(name="vid"), **DB_YAML_CREDS),
-            name="hero", video=MagicMock(name="vid"), **DB_YAML_CREDS
-        ),
     ),
     (
         "video_get_character_handler",
@@ -453,3 +433,33 @@ def test_db_yaml_credentials_reach_every_handler(seams, handler_name, invoke):
     assert litellm_params.get("api_base") == DB_YAML_CREDS["api_base"]
     assert litellm_params.get("api_version") == DB_YAML_CREDS["api_version"]
     assert litellm_params.get("vertex_project") == DB_YAML_CREDS["vertex_project"]
+
+
+def test_video_generation__custom_provider_map_base_video_config_is_dispatched(monkeypatch: pytest.MonkeyPatch) -> None:
+    from litellm.llms.openai.videos.transformation import OpenAIVideoConfig
+
+    class QueueVideoConfig(OpenAIVideoConfig):
+        pass
+
+    video_config: Final = QueueVideoConfig()
+    monkeypatch.setattr(litellm, "custom_provider_map", [{"provider": "queue-video", "custom_handler": video_config}])
+    monkeypatch.setattr(litellm, "_custom_providers", ["queue-video"])
+    handler: Final = MagicMock(spec=BaseLLMHTTPHandler)
+    with patch.object(videos_main, "base_llm_http_handler", handler):
+        videos_main.video_generation(model="queue-video/seedance-1", prompt="a cat", api_base="http://backend.test")
+    kw: Final = handler.video_generation_handler.call_args.kwargs
+    assert kw["video_generation_provider_config"] is video_config
+    assert kw["custom_llm_provider"] == "queue-video"
+    assert kw["model"] == "seedance-1"
+
+
+def test_video_generation__custom_provider_without_a_video_config_is_a_400(monkeypatch: pytest.MonkeyPatch) -> None:
+    from litellm.llms.custom_llm import CustomLLM
+
+    monkeypatch.setattr(litellm, "custom_provider_map", [{"provider": "image-only", "custom_handler": CustomLLM()}])
+    monkeypatch.setattr(litellm, "_custom_providers", ["image-only"])
+    handler: Final = MagicMock(spec=BaseLLMHTTPHandler)
+    with patch.object(videos_main, "base_llm_http_handler", handler):
+        with pytest.raises(litellm.BadRequestError, match="video generation is not supported for image-only"):
+            videos_main.video_generation(model="image-only/seedance-1", prompt="a cat")
+    handler.video_generation_handler.assert_not_called()