Add LTX Video API support

litellm/llms/base_llm/videos/transformation.py

@@ -31,6 +31,16 @@ class BaseVideoConfig(ABC):
     def __init__(self):
         pass
 
+    def requires_authentication_for_video_content(self) -> bool:
+        """
+        Whether the shared video content handler should call validate_environment()
+        before constructing the provider-specific content request.
+
+        Most providers need authenticated headers even for content retrieval.
+        Providers that serve locally persisted artifacts can override this.
+        """
+        return True
+
     @classmethod
     def get_config(cls):
         return {

litellm/llms/custom_httpx/llm_http_handler.py

@@ -1,5 +1,8 @@
+import asyncio
 import json
 import ssl
+import tempfile
+from pathlib import Path
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -13,6 +16,8 @@ from typing import (
     Union,
     cast,
 )
+from urllib.parse import urlparse
+from urllib.request import url2pathname
 
 import httpx  # type: ignore
 from openai.types.file_deleted import FileDeleted
@@ -150,6 +155,19 @@ else:
     LiteLLMLoggingObj = Any
 
 
+def _read_local_file_url(url: str) -> bytes:
+    parsed = urlparse(url)
+    file_path = Path(url2pathname(f"{parsed.netloc}{parsed.path}")).resolve()
+    allowed_root = Path(tempfile.gettempdir()).resolve()
+    try:
+        file_path.relative_to(allowed_root)
+    except ValueError as exc:
+        raise ValueError(
+            f"file:// URL resolves to a path outside the allowed temp directory: {file_path}"
+        ) from exc
+    return file_path.read_bytes()
+
+
 class BaseLLMHTTPHandler:
     async def _make_common_async_call(
         self,
@@ -5799,13 +5817,14 @@ class BaseLLMHTTPHandler:
         else:
             sync_httpx_client = client
 
-        headers = video_content_provider_config.validate_environment(
-            headers=extra_headers or {},
-            model="",
-            api_key=api_key,
-            litellm_params=litellm_params,
-        )
-
+        headers: Dict[str, Any] = {}
+        if video_content_provider_config.requires_authentication_for_video_content():
+            headers = video_content_provider_config.validate_environment(
+                headers=headers,
+                model="",
+                api_key=api_key,
+                litellm_params=litellm_params,
+            )
         if extra_headers:
             headers.update(extra_headers)
 
@@ -5825,6 +5844,9 @@ class BaseLLMHTTPHandler:
         )
 
         try:
+            if url.startswith("file://"):
+                return _read_local_file_url(url)
+
             # Use POST if params contains data (e.g., Vertex AI fetchPredictOperation)
             # Otherwise use GET (e.g., OpenAI video content download)
             if data:
@@ -5877,13 +5899,14 @@ class BaseLLMHTTPHandler:
         else:
             async_httpx_client = client
 
-        headers = video_content_provider_config.validate_environment(
-            headers=extra_headers or {},
-            model="",
-            api_key=api_key,
-            litellm_params=litellm_params,
-        )
-
+        headers: Dict[str, Any] = {}
+        if video_content_provider_config.requires_authentication_for_video_content():
+            headers = video_content_provider_config.validate_environment(
+                headers=headers,
+                model="",
+                api_key=api_key,
+                litellm_params=litellm_params,
+            )
         if extra_headers:
             headers.update(extra_headers)
 
@@ -5903,6 +5926,9 @@ class BaseLLMHTTPHandler:
         )
 
         try:
+            if url.startswith("file://"):
+                return await asyncio.to_thread(_read_local_file_url, url)
+
             # Use POST if params contains data (e.g., Vertex AI fetchPredictOperation)
             # Otherwise use GET (e.g., OpenAI video content download)
             if data:

litellm/llms/ltx/__init__.py

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

litellm/llms/ltx/videos/transformation.py

@@ -0,0 +1,367 @@
+import tempfile
+import time
+import uuid
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
+
+import httpx
+from httpx._types import RequestFiles
+
+import litellm
+from litellm.llms.base_llm.chat.transformation import BaseLLMException
+from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
+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,
+)
+
+if TYPE_CHECKING:
+    from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
+
+    LiteLLMLoggingObj = _LiteLLMLoggingObj
+else:
+    LiteLLMLoggingObj = Any
+
+
+LTX_VIDEO_STORAGE_DIR = Path(tempfile.gettempdir()) / "litellm_ltx_videos"
+
+
+def _get_ltx_video_storage_path(video_id: str) -> Path:
+    return LTX_VIDEO_STORAGE_DIR / f"{video_id}.mp4"
+
+
+def _persist_ltx_video_bytes(video_id: str, video_bytes: bytes) -> Path:
+    LTX_VIDEO_STORAGE_DIR.mkdir(parents=True, exist_ok=True)
+    video_path = _get_ltx_video_storage_path(video_id)
+    video_path.write_bytes(video_bytes)
+    return video_path
+
+
+class LTXVideoConfig(BaseVideoConfig):
+    """
+    Configuration class for LTX Video generation.
+
+    LTX Video API is synchronous — it returns binary video data directly
+    in the response rather than a task ID to poll.
+
+    Supports two endpoints:
+    - POST /v1/text-to-video (text prompt only)
+    - POST /v1/image-to-video (text prompt + source image)
+    """
+
+    def __init__(self):
+        super().__init__()
+
+    def requires_authentication_for_video_content(self) -> bool:
+        return False
+
+    def get_supported_openai_params(self, model: str) -> list:
+        return [
+            "model",
+            "prompt",
+            "input_reference",
+            "seconds",
+            "size",
+            "user",
+            "extra_headers",
+        ]
+
+    def map_openai_params(
+        self,
+        video_create_optional_params: VideoCreateOptionalRequestParams,
+        model: str,
+        drop_params: bool,
+    ) -> Dict:
+        mapped_params: Dict[str, Any] = {}
+
+        if "input_reference" in video_create_optional_params:
+            mapped_params["image_uri"] = video_create_optional_params["input_reference"]
+
+        if "size" in video_create_optional_params:
+            size = video_create_optional_params["size"]
+            if isinstance(size, str):
+                mapped_params["resolution"] = size
+
+        if "seconds" in video_create_optional_params:
+            seconds = video_create_optional_params["seconds"]
+            if seconds is not None:
+                try:
+                    mapped_params["duration"] = (
+                        int(float(seconds))
+                        if isinstance(seconds, str)
+                        else int(seconds)
+                    )
+                except (ValueError, TypeError):
+                    # Ignore invalid seconds values and let the provider fall back to defaults.
+                    pass
+
+        # Pass through LTX-specific parameters
+        supported_openai_params = self.get_supported_openai_params(model)
+        for key, value in video_create_optional_params.items():
+            if key not in supported_openai_params:
+                mapped_params[key] = value
+
+        return mapped_params
+
+    def validate_environment(
+        self,
+        headers: dict,
+        model: str,
+        api_key: Optional[str] = None,
+        litellm_params: Optional[GenericLiteLLMParams] = None,
+    ) -> dict:
+        api_key = (
+            api_key
+            or (litellm_params.api_key if litellm_params else None)
+            or litellm.api_key
+            or get_secret_str("LTX_API_KEY")
+        )
+
+        if api_key is None:
+            raise ValueError(
+                "LTX API key is required. Set LTX_API_KEY environment variable "
+                "or pass api_key parameter."
+            )
+
+        headers.update(
+            {
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": "application/json",
+            }
+        )
+        return headers
+
+    def get_complete_url(
+        self,
+        model: str,
+        api_base: Optional[str],
+        litellm_params: dict,
+    ) -> str:
+        """
+        Return the shared LTX API base URL.
+
+        The final create endpoint depends on whether the request includes
+        an input image, so `transform_video_create_request()` appends the
+        `/text-to-video` or `/image-to-video` suffix.
+        """
+        if api_base is None:
+            api_base = "https://api.ltx.video/v1"
+
+        return api_base.rstrip("/")
+
+    def transform_video_create_request(
+        self,
+        model: str,
+        prompt: str,
+        api_base: str,
+        video_create_optional_request_params: Dict,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,
+    ) -> Tuple[Dict, RequestFiles, str]:
+        request_data: Dict[str, Any] = {
+            "prompt": prompt,
+            "model": model,
+        }
+
+        # Add mapped parameters
+        request_data.update(video_create_optional_request_params)
+
+        files_list: List[Tuple[str, Any]] = []
+
+        # Choose endpoint based on whether image_uri is present
+        if "image_uri" in request_data:
+            full_api_base = f"{api_base}/image-to-video"
+        else:
+            full_api_base = f"{api_base}/text-to-video"
+
+        return request_data, files_list, full_api_base
+
+    def transform_video_create_response(
+        self,
+        model: str,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        custom_llm_provider: Optional[str] = None,
+        request_data: Optional[Dict] = None,
+    ) -> VideoObject:
+        """
+        Transform the LTX video creation response.
+
+        LTX returns binary video data directly (application/octet-stream).
+        We generate a UUID for the video ID and set status to "completed".
+        """
+        if not raw_response.content:
+            raise BaseLLMException(
+                status_code=raw_response.status_code or 502,
+                message="LTX returned an empty video response body.",
+                request=raw_response.request,
+                response=raw_response,
+            )
+
+        video_id = str(uuid.uuid4())
+        created_at = int(time.time())
+        stored_video_path = _persist_ltx_video_bytes(
+            video_id=video_id, video_bytes=raw_response.content
+        )
+
+        video_data: Dict[str, Any] = {
+            "id": video_id,
+            "object": "video",
+            "status": "completed",
+            "created_at": created_at,
+            "completed_at": created_at,
+        }
+
+        if request_data:
+            if "model" in request_data:
+                video_data["model"] = request_data["model"]
+            if "resolution" in request_data:
+                video_data["size"] = request_data["resolution"]
+            if "duration" in request_data:
+                video_data["seconds"] = str(request_data["duration"])
+
+        video_obj = VideoObject(**video_data)  # type: ignore[arg-type]
+
+        if custom_llm_provider and video_obj.id:
+            video_obj.id = encode_video_id_with_provider(
+                video_obj.id, custom_llm_provider, model
+            )
+
+        usage_data: Dict[str, Any] = {}
+        if request_data and "duration" in request_data:
+            try:
+                usage_data["duration_seconds"] = float(request_data["duration"])
+            except (ValueError, TypeError):
+                # Duration is used only for cost accounting, so skip invalid values.
+                pass
+        video_obj.usage = usage_data
+        video_obj._hidden_params = {
+            "video_content_path": str(stored_video_path),
+            "video_content_url": stored_video_path.resolve().as_uri(),
+        }
+
+        return video_obj
+
+    def transform_video_content_request(
+        self,
+        video_id: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,
+        variant: Optional[str] = None,
+    ) -> Tuple[str, Dict]:
+        if variant is not None:
+            raise NotImplementedError(
+                "LTX video content variants are not supported. "
+                "Only the generated MP4 can be retrieved."
+            )
+
+        original_video_id = extract_original_video_id(video_id)
+        stored_video_path = _get_ltx_video_storage_path(original_video_id)
+        if not stored_video_path.exists():
+            raise BaseLLMException(
+                status_code=404,
+                message=(
+                    "No locally stored LTX video content was found for this video_id. "
+                    "Recreate the video before calling video_content()."
+                ),
+            )
+
+        return stored_video_path.resolve().as_uri(), {}
+
+    def transform_video_content_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+    ) -> bytes:
+        return raw_response.content
+
+    def transform_video_remix_request(
+        self,
+        video_id: str,
+        prompt: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,
+        extra_body: Optional[Dict[str, Any]] = None,
+    ) -> Tuple[str, Dict]:
+        raise NotImplementedError("Video remix is not supported by LTX API")
+
+    def transform_video_remix_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        custom_llm_provider: Optional[str] = None,
+    ) -> VideoObject:
+        raise NotImplementedError("Video remix is not supported by LTX API")
+
+    def transform_video_list_request(
+        self,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,
+        after: Optional[str] = None,
+        limit: Optional[int] = None,
+        order: Optional[str] = None,
+        extra_query: Optional[Dict[str, Any]] = None,
+    ) -> Tuple[str, Dict]:
+        raise NotImplementedError("Video listing is not supported by LTX API")
+
+    def transform_video_list_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        custom_llm_provider: Optional[str] = None,
+    ) -> Dict[str, str]:
+        raise NotImplementedError("Video listing is not supported by LTX API")
+
+    def transform_video_delete_request(
+        self,
+        video_id: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,
+    ) -> Tuple[str, Dict]:
+        raise NotImplementedError("Video deletion is not supported by LTX API")
+
+    def transform_video_delete_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+    ) -> VideoObject:
+        raise NotImplementedError("Video deletion is not supported by LTX API")
+
+    def transform_video_status_retrieve_request(
+        self,
+        video_id: str,
+        api_base: str,
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,
+    ) -> Tuple[str, Dict]:
+        raise NotImplementedError(
+            "Video status retrieval is not supported by LTX API. "
+            "LTX video generation is synchronous."
+        )
+
+    def transform_video_status_retrieve_response(
+        self,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        custom_llm_provider: Optional[str] = None,
+    ) -> VideoObject:
+        raise NotImplementedError(
+            "Video status retrieval is not supported by LTX API. "
+            "LTX video generation is synchronous."
+        )
+
+    def get_error_class(
+        self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers]
+    ) -> BaseLLMException:
+        raise BaseLLMException(
+            status_code=status_code,
+            message=error_message,
+            headers=headers,
+        )

litellm/types/utils.py

@@ -3179,6 +3179,7 @@ class LlmProviders(str, Enum):
     BYTEZ = "bytez"
     REPLICATE = "replicate"
     RUNWAYML = "runwayml"
+    LTX = "ltx"
     AWS_POLLY = "aws_polly"
     HUGGINGFACE = "huggingface"
     TOGETHER_AI = "together_ai"

litellm/utils.py

@@ -783,9 +783,9 @@ def function_setup(  # noqa: PLR0915
         coroutine_checker = get_coroutine_checker_fn()
 
         ## DYNAMIC CALLBACKS ##
-        dynamic_callbacks: Optional[
-            List[Union[str, Callable, "CustomLogger"]]
-        ] = kwargs.pop("callbacks", None)
+        dynamic_callbacks: Optional[List[Union[str, Callable, "CustomLogger"]]] = (
+            kwargs.pop("callbacks", None)
+        )
         all_callbacks = get_dynamic_callbacks(dynamic_callbacks=dynamic_callbacks)
 
         if len(all_callbacks) > 0:
@@ -1691,9 +1691,9 @@ def client(original_function):  # noqa: PLR0915
                         exception=e,
                         retry_policy=kwargs.get("retry_policy"),
                     )
-                    kwargs[
-                        "retry_policy"
-                    ] = reset_retry_policy()  # prevent infinite loops
+                    kwargs["retry_policy"] = (
+                        reset_retry_policy()
+                    )  # prevent infinite loops
                 litellm.num_retries = (
                     None  # set retries to None to prevent infinite loops
                 )
@@ -1740,9 +1740,9 @@ def client(original_function):  # noqa: PLR0915
                         exception=e,
                         retry_policy=kwargs.get("retry_policy"),
                     )
-                    kwargs[
-                        "retry_policy"
-                    ] = reset_retry_policy()  # prevent infinite loops
+                    kwargs["retry_policy"] = (
+                        reset_retry_policy()
+                    )  # prevent infinite loops
                 litellm.num_retries = (
                     None  # set retries to None to prevent infinite loops
                 )
@@ -3771,10 +3771,10 @@ def pre_process_non_default_params(
 
     if "response_format" in non_default_params:
         if provider_config is not None:
-            non_default_params[
-                "response_format"
-            ] = provider_config.get_json_schema_from_pydantic_object(
-                response_format=non_default_params["response_format"]
+            non_default_params["response_format"] = (
+                provider_config.get_json_schema_from_pydantic_object(
+                    response_format=non_default_params["response_format"]
+                )
             )
         else:
             non_default_params["response_format"] = type_to_response_format_param(
@@ -3903,16 +3903,16 @@ def pre_process_optional_params(
                     True  # so that main.py adds the function call to the prompt
                 )
                 if "tools" in non_default_params:
-                    optional_params[
-                        "functions_unsupported_model"
-                    ] = non_default_params.pop("tools")
+                    optional_params["functions_unsupported_model"] = (
+                        non_default_params.pop("tools")
+                    )
                     non_default_params.pop(
                         "tool_choice", None
                     )  # causes ollama requests to hang
                 elif "functions" in non_default_params:
-                    optional_params[
-                        "functions_unsupported_model"
-                    ] = non_default_params.pop("functions")
+                    optional_params["functions_unsupported_model"] = (
+                        non_default_params.pop("functions")
+                    )
             elif (
                 litellm.add_function_to_prompt
             ):  # if user opts to add it to prompt instead
@@ -4893,9 +4893,7 @@ def _get_order_filtered_deployments(
 ) -> List:
     if target_order is not None:
         filtered = [
-            d
-            for d in healthy_deployments
-            if _get_deployment_order(d) == target_order
+            d for d in healthy_deployments if _get_deployment_order(d) == target_order
         ]
         if filtered:
             return filtered
@@ -7549,9 +7547,9 @@ class ModelResponseIterator:
         if convert_to_delta is True:
             _stream_response = ModelResponseStream()
             _stream_response.choices[0].delta.content = model_response.choices[0].message.content  # type: ignore
-            self.model_response: Union[
-                ModelResponse, ModelResponseStream
-            ] = _stream_response
+            self.model_response: Union[ModelResponse, ModelResponseStream] = (
+                _stream_response
+            )
         else:
             self.model_response = model_response
         self.is_done = False
@@ -8943,6 +8941,10 @@ class ProviderConfigManager:
             from litellm.llms.runwayml.videos.transformation import RunwayMLVideoConfig
 
             return RunwayMLVideoConfig()
+        elif LlmProviders.LTX == provider:
+            from litellm.llms.ltx.videos.transformation import LTXVideoConfig
+
+            return LTXVideoConfig()
         return None
 
     @staticmethod

model_prices_and_context_window.json

@@ -33670,6 +33670,50 @@
             "comment": "Estimated cost based on standard TTS pricing. RunwayML uses ElevenLabs models."
         }
     },
+    "ltx/ltx-2-3-fast": {
+        "litellm_provider": "ltx",
+        "mode": "video_generation",
+        "output_cost_per_video_per_second": 0.04,
+        "source": "https://docs.ltx.video/pricing",
+        "supported_modalities": [
+            "text",
+            "image"
+        ],
+        "supported_output_modalities": [
+            "video"
+        ],
+        "supported_resolutions": [
+            "1280x720",
+            "1920x1080",
+            "2560x1440",
+            "3840x2160"
+        ],
+        "metadata": {
+            "comment": "$0.04/sec at 1080p, $0.08/sec at 1440p, $0.16/sec at 4K. Using 1080p as base cost."
+        }
+    },
+    "ltx/ltx-2-3-pro": {
+        "litellm_provider": "ltx",
+        "mode": "video_generation",
+        "output_cost_per_video_per_second": 0.06,
+        "source": "https://docs.ltx.video/pricing",
+        "supported_modalities": [
+            "text",
+            "image"
+        ],
+        "supported_output_modalities": [
+            "video"
+        ],
+        "supported_resolutions": [
+            "1280x720",
+            "1920x1080",
+            "2560x1440",
+            "3840x2160"
+        ],
+        "metadata": {
+            "comment": "$0.06/sec at 1080p, $0.12/sec at 1440p, $0.24/sec at 4K. Using 1080p as base cost."
+        }
+    },
     "fireworks_ai/accounts/fireworks/models/qwen3-coder-480b-a35b-instruct": {
         "max_tokens": 262144,
         "max_input_tokens": 262144,

tests/test_litellm/llms/ltx/videos/test_ltx_video_transformation.py

@@ -0,0 +1,497 @@
+"""
+Tests for LTX Video generation transformation.
+"""
+
+import asyncio
+from unittest.mock import Mock
+
+import httpx
+import pytest
+
+import litellm.llms.ltx.videos.transformation as ltx_video_transformation
+from litellm.llms.base_llm.chat.transformation import BaseLLMException
+from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
+from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
+from litellm.llms.ltx.videos.transformation import LTXVideoConfig
+from litellm.types.router import GenericLiteLLMParams
+from litellm.types.videos.main import VideoObject
+from litellm.types.videos.utils import (
+    encode_video_id_with_provider,
+    extract_original_video_id,
+)
+
+
+class TestLTXVideoTransformation:
+    """Test LTXVideoConfig transformation class."""
+
+    def setup_method(self):
+        """Setup test fixtures."""
+        self.config = LTXVideoConfig()
+        self.mock_logging_obj = Mock()
+
+    def test_get_supported_openai_params(self):
+        """Test supported OpenAI parameters list."""
+        params = self.config.get_supported_openai_params("ltx-2-3-fast")
+        assert "model" in params
+        assert "prompt" in params
+        assert "input_reference" in params
+        assert "seconds" in params
+        assert "size" in params
+        assert "user" in params
+        assert "extra_headers" in params
+
+    def test_map_openai_params_basic(self):
+        """Test parameter mapping from OpenAI format to LTX format."""
+        mapped = self.config.map_openai_params(
+            video_create_optional_params={
+                "input_reference": "https://example.com/image.jpg",
+                "seconds": "5",
+                "size": "1920x1080",
+            },
+            model="ltx-2-3-fast",
+            drop_params=False,
+        )
+
+        assert mapped["image_uri"] == "https://example.com/image.jpg"
+        assert mapped["duration"] == 5
+        assert mapped["resolution"] == "1920x1080"
+
+    def test_map_openai_params_passthrough(self):
+        """Test that LTX-specific params are passed through."""
+        mapped = self.config.map_openai_params(
+            video_create_optional_params={
+                "fps": 30,
+                "generate_audio": False,
+                "camera_motion": "dolly_in",
+            },
+            model="ltx-2-3-fast",
+            drop_params=False,
+        )
+
+        assert mapped["fps"] == 30
+        assert mapped["generate_audio"] is False
+        assert mapped["camera_motion"] == "dolly_in"
+
+    def test_map_openai_params_seconds_int(self):
+        """Test seconds conversion when provided as int."""
+        mapped = self.config.map_openai_params(
+            video_create_optional_params={"seconds": 10},
+            model="ltx-2-3-fast",
+            drop_params=False,
+        )
+        assert mapped["duration"] == 10
+
+    def test_validate_environment(self):
+        """Test authentication header setup."""
+        headers = self.config.validate_environment(
+            headers={},
+            model="ltx-2-3-fast",
+            api_key="test-api-key",
+        )
+
+        assert headers["Authorization"] == "Bearer test-api-key"
+        assert headers["Content-Type"] == "application/json"
+
+    def test_validate_environment_missing_key(self):
+        """Test that missing API key raises ValueError."""
+        with pytest.raises(ValueError, match="LTX API key is required"):
+            self.config.validate_environment(
+                headers={},
+                model="ltx-2-3-fast",
+                api_key=None,
+            )
+
+    def test_validate_environment_empty_model_still_requires_key(self):
+        """Test that content retrieval no longer relies on a model='' sentinel."""
+        with pytest.raises(ValueError, match="LTX API key is required"):
+            self.config.validate_environment(
+                headers={},
+                model="",
+                api_key=None,
+            )
+
+    def test_get_complete_url_default(self):
+        """Test default API base URL."""
+        url = self.config.get_complete_url(
+            model="ltx-2-3-fast",
+            api_base=None,
+            litellm_params={},
+        )
+        assert url == "https://api.ltx.video/v1"
+
+    def test_get_complete_url_custom(self):
+        """Test custom API base URL."""
+        url = self.config.get_complete_url(
+            model="ltx-2-3-fast",
+            api_base="https://custom.api.com/v1/",
+            litellm_params={},
+        )
+        assert url == "https://custom.api.com/v1"
+
+    def test_transform_video_create_request_text_to_video(self):
+        """Test text-to-video request transformation."""
+        data, files, url = self.config.transform_video_create_request(
+            model="ltx-2-3-fast",
+            prompt="A serene mountain landscape at sunset",
+            api_base="https://api.ltx.video/v1",
+            video_create_optional_request_params={
+                "duration": 5,
+                "resolution": "1920x1080",
+            },
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert data["model"] == "ltx-2-3-fast"
+        assert data["prompt"] == "A serene mountain landscape at sunset"
+        assert data["duration"] == 5
+        assert data["resolution"] == "1920x1080"
+        assert "image_uri" not in data
+        assert files == []
+        assert url == "https://api.ltx.video/v1/text-to-video"
+
+    def test_transform_video_create_request_image_to_video(self):
+        """Test image-to-video request transformation when image_uri is present."""
+        data, files, url = self.config.transform_video_create_request(
+            model="ltx-2-3-pro",
+            prompt="Animate this image with gentle motion",
+            api_base="https://api.ltx.video/v1",
+            video_create_optional_request_params={
+                "image_uri": "https://example.com/source.jpg",
+                "duration": 3,
+                "resolution": "1280x720",
+            },
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert data["model"] == "ltx-2-3-pro"
+        assert data["prompt"] == "Animate this image with gentle motion"
+        assert data["image_uri"] == "https://example.com/source.jpg"
+        assert files == []
+        assert url == "https://api.ltx.video/v1/image-to-video"
+
+    def test_transform_video_create_request_with_optional_params(self):
+        """Test request with LTX-specific optional parameters."""
+        data, files, url = self.config.transform_video_create_request(
+            model="ltx-2-3-pro",
+            prompt="A cinematic pan across a cityscape",
+            api_base="https://api.ltx.video/v1",
+            video_create_optional_request_params={
+                "duration": 8,
+                "resolution": "1920x1080",
+                "fps": 30,
+                "generate_audio": True,
+                "camera_motion": "dolly_in",
+            },
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert data["fps"] == 30
+        assert data["generate_audio"] is True
+        assert data["camera_motion"] == "dolly_in"
+        assert url == "https://api.ltx.video/v1/text-to-video"
+
+    def test_transform_video_create_response_binary(self, monkeypatch, tmp_path):
+        """Test that binary response produces a completed VideoObject."""
+        monkeypatch.setattr(ltx_video_transformation, "LTX_VIDEO_STORAGE_DIR", tmp_path)
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.content = b"\x00\x00\x00\x1cftypisom"  # fake video bytes
+        mock_response.status_code = 200
+        mock_response.request = httpx.Request("POST", "https://api.ltx.video/v1")
+
+        request_data = {
+            "model": "ltx-2-3-fast",
+            "prompt": "test",
+            "duration": 5,
+            "resolution": "1920x1080",
+        }
+
+        result = self.config.transform_video_create_response(
+            model="ltx-2-3-fast",
+            raw_response=mock_response,
+            logging_obj=self.mock_logging_obj,
+            custom_llm_provider="ltx",
+            request_data=request_data,
+        )
+
+        stored_video_path = tmp_path / f"{extract_original_video_id(result.id)}.mp4"
+
+        assert isinstance(result, VideoObject)
+        assert result.status == "completed"
+        assert result.created_at is not None
+        assert result.created_at > 0
+        assert result.completed_at is not None
+        assert result.model == "ltx-2-3-fast"
+        assert result.size == "1920x1080"
+        assert result.seconds == "5"
+        assert result.id.startswith("video_")
+        assert result.usage is not None
+        assert result.usage["duration_seconds"] == 5.0
+        assert stored_video_path.read_bytes() == mock_response.content
+        assert result._hidden_params["video_content_path"] == str(stored_video_path)
+
+    def test_transform_video_create_response_without_request_data(
+        self, monkeypatch, tmp_path
+    ):
+        """Test response transformation without request data."""
+        monkeypatch.setattr(ltx_video_transformation, "LTX_VIDEO_STORAGE_DIR", tmp_path)
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.content = b"\x00\x00\x00"
+        mock_response.status_code = 200
+        mock_response.request = httpx.Request("POST", "https://api.ltx.video/v1")
+
+        result = self.config.transform_video_create_response(
+            model="ltx-2-3-fast",
+            raw_response=mock_response,
+            logging_obj=self.mock_logging_obj,
+            custom_llm_provider=None,
+            request_data=None,
+        )
+
+        assert isinstance(result, VideoObject)
+        assert result.status == "completed"
+        assert result.id  # should have a UUID
+
+    def test_transform_video_create_response_empty_binary_raises(self):
+        """Test that empty create responses fail loudly."""
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.content = b""
+        mock_response.status_code = 200
+        mock_response.request = httpx.Request("POST", "https://api.ltx.video/v1")
+
+        with pytest.raises(BaseLLMException, match="empty video response body"):
+            self.config.transform_video_create_response(
+                model="ltx-2-3-fast",
+                raw_response=mock_response,
+                logging_obj=self.mock_logging_obj,
+                custom_llm_provider="ltx",
+                request_data={"model": "ltx-2-3-fast"},
+            )
+
+    def test_transform_video_content_request_uses_local_file(
+        self, monkeypatch, tmp_path
+    ):
+        """Test content requests resolve to the locally persisted file."""
+        monkeypatch.setattr(ltx_video_transformation, "LTX_VIDEO_STORAGE_DIR", tmp_path)
+        original_video_id = "ltx-local-video"
+        stored_video_path = tmp_path / f"{original_video_id}.mp4"
+        stored_video_path.write_bytes(b"fake-video-binary-data")
+
+        url, data = self.config.transform_video_content_request(
+            video_id=encode_video_id_with_provider(
+                original_video_id, "ltx", "ltx-2-3-fast"
+            ),
+            api_base="https://api.ltx.video/v1",
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert url == stored_video_path.resolve().as_uri()
+        assert data == {}
+
+    def test_video_content_handler_reads_local_file(self, monkeypatch, tmp_path):
+        """Test the shared video content handler can serve local LTX artifacts."""
+        monkeypatch.setattr(ltx_video_transformation, "LTX_VIDEO_STORAGE_DIR", tmp_path)
+        original_video_id = "ltx-local-video"
+        expected_bytes = b"fake-video-binary-data"
+        stored_video_path = tmp_path / f"{original_video_id}.mp4"
+        stored_video_path.write_bytes(expected_bytes)
+
+        result = BaseLLMHTTPHandler().video_content_handler(
+            video_id=encode_video_id_with_provider(
+                original_video_id, "ltx", "ltx-2-3-fast"
+            ),
+            video_content_provider_config=self.config,
+            custom_llm_provider="ltx",
+            litellm_params=GenericLiteLLMParams(),
+            logging_obj=self.mock_logging_obj,
+            timeout=30,
+        )
+
+        assert result == expected_bytes
+
+    def test_async_video_content_handler_reads_local_file(self, monkeypatch, tmp_path):
+        """Test the async shared content handler can serve local LTX artifacts."""
+        monkeypatch.setattr(ltx_video_transformation, "LTX_VIDEO_STORAGE_DIR", tmp_path)
+        original_video_id = "ltx-local-video"
+        expected_bytes = b"fake-video-binary-data"
+        stored_video_path = tmp_path / f"{original_video_id}.mp4"
+        stored_video_path.write_bytes(expected_bytes)
+
+        result = asyncio.run(
+            BaseLLMHTTPHandler().async_video_content_handler(
+                video_id=encode_video_id_with_provider(
+                    original_video_id, "ltx", "ltx-2-3-fast"
+                ),
+                video_content_provider_config=self.config,
+                custom_llm_provider="ltx",
+                litellm_params=GenericLiteLLMParams(),
+                logging_obj=self.mock_logging_obj,
+                timeout=30,
+                client=Mock(spec=AsyncHTTPHandler),
+            )
+        )
+
+        assert result == expected_bytes
+
+    def test_unsupported_operations(self, monkeypatch, tmp_path):
+        """Test that unsupported operations raise NotImplementedError."""
+        monkeypatch.setattr(ltx_video_transformation, "LTX_VIDEO_STORAGE_DIR", tmp_path)
+        stored_video_path = tmp_path / "ltx-local-video.mp4"
+        stored_video_path.write_bytes(b"fake-video-binary-data")
+
+        with pytest.raises(NotImplementedError, match="content variants"):
+            self.config.transform_video_content_request(
+                video_id=encode_video_id_with_provider(
+                    "ltx-local-video", "ltx", "ltx-2-3-fast"
+                ),
+                api_base="",
+                litellm_params=GenericLiteLLMParams(),
+                headers={},
+                variant="thumbnail",
+            )
+
+        with pytest.raises(BaseLLMException, match="No locally stored LTX video"):
+            self.config.transform_video_content_request(
+                video_id="missing-video",
+                api_base="",
+                litellm_params=GenericLiteLLMParams(),
+                headers={},
+            )
+
+        with pytest.raises(NotImplementedError):
+            self.config.transform_video_status_retrieve_request(
+                video_id="test",
+                api_base="",
+                litellm_params=GenericLiteLLMParams(),
+                headers={},
+            )
+
+        with pytest.raises(NotImplementedError):
+            self.config.transform_video_remix_request(
+                video_id="test",
+                prompt="test",
+                api_base="",
+                litellm_params=GenericLiteLLMParams(),
+                headers={},
+            )
+
+        with pytest.raises(NotImplementedError):
+            self.config.transform_video_delete_request(
+                video_id="test",
+                api_base="",
+                litellm_params=GenericLiteLLMParams(),
+                headers={},
+            )
+
+        with pytest.raises(NotImplementedError):
+            self.config.transform_video_list_request(
+                api_base="", litellm_params=GenericLiteLLMParams(), headers={}
+            )
+
+    def test_full_text_to_video_workflow(self, monkeypatch, tmp_path):
+        """Test complete text-to-video workflow."""
+        monkeypatch.setattr(ltx_video_transformation, "LTX_VIDEO_STORAGE_DIR", tmp_path)
+        config = LTXVideoConfig()
+        mock_logging_obj = Mock()
+
+        # Step 1: Map params
+        mapped = config.map_openai_params(
+            video_create_optional_params={
+                "seconds": "5",
+                "size": "1920x1080",
+            },
+            model="ltx-2-3-fast",
+            drop_params=False,
+        )
+
+        assert mapped["duration"] == 5
+        assert mapped["resolution"] == "1920x1080"
+
+        # Step 2: Create request
+        data, files, url = config.transform_video_create_request(
+            model="ltx-2-3-fast",
+            prompt="A serene mountain landscape",
+            api_base="https://api.ltx.video/v1",
+            video_create_optional_request_params=mapped,
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert url == "https://api.ltx.video/v1/text-to-video"
+        assert data["prompt"] == "A serene mountain landscape"
+        assert data["model"] == "ltx-2-3-fast"
+        assert data["duration"] == 5
+
+        # Step 3: Parse response (binary)
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.content = b"fake-video-binary-data"
+        mock_response.status_code = 200
+        mock_response.request = httpx.Request("POST", "https://api.ltx.video/v1")
+
+        video_obj = config.transform_video_create_response(
+            model="ltx-2-3-fast",
+            raw_response=mock_response,
+            logging_obj=mock_logging_obj,
+            custom_llm_provider="ltx",
+            request_data=data,
+        )
+
+        assert video_obj.status == "completed"
+        assert video_obj.model == "ltx-2-3-fast"
+        assert video_obj.seconds == "5"
+
+    def test_full_image_to_video_workflow(self, monkeypatch, tmp_path):
+        """Test complete image-to-video workflow."""
+        monkeypatch.setattr(ltx_video_transformation, "LTX_VIDEO_STORAGE_DIR", tmp_path)
+        config = LTXVideoConfig()
+        mock_logging_obj = Mock()
+
+        # Step 1: Map params
+        mapped = config.map_openai_params(
+            video_create_optional_params={
+                "input_reference": "https://example.com/image.jpg",
+                "seconds": "3",
+                "size": "1280x720",
+            },
+            model="ltx-2-3-pro",
+            drop_params=False,
+        )
+
+        assert mapped["image_uri"] == "https://example.com/image.jpg"
+
+        # Step 2: Create request
+        data, files, url = config.transform_video_create_request(
+            model="ltx-2-3-pro",
+            prompt="Animate the scene with flowing water",
+            api_base="https://api.ltx.video/v1",
+            video_create_optional_request_params=mapped,
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+        assert url == "https://api.ltx.video/v1/image-to-video"
+        assert data["image_uri"] == "https://example.com/image.jpg"
+
+        # Step 3: Parse response
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.content = b"fake-video-binary-data"
+        mock_response.status_code = 200
+        mock_response.request = httpx.Request("POST", "https://api.ltx.video/v1")
+
+        video_obj = config.transform_video_create_response(
+            model="ltx-2-3-pro",
+            raw_response=mock_response,
+            logging_obj=mock_logging_obj,
+            custom_llm_provider="ltx",
+            request_data=data,
+        )
+
+        assert video_obj.status == "completed"
+        assert video_obj.size == "1280x720"
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

tests/test_litellm/test_video_generation.py

@@ -3,6 +3,7 @@ import io
 import json
 import os
 import sys
+from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -1023,6 +1024,16 @@ def test_video_content_handler_uses_get_for_openai():
     assert called_url == "https://api.openai.com/v1/videos/video_abc/content"
 
 
+def test_read_local_file_url_rejects_non_temp_paths():
+    """Local file helper should not read files outside the temp directory."""
+    from litellm.llms.custom_httpx.llm_http_handler import _read_local_file_url
+
+    disallowed_url = Path(__file__).resolve().as_uri()
+
+    with pytest.raises(ValueError, match="outside the allowed temp directory"):
+        _read_local_file_url(disallowed_url)
+
+
 def test_video_content_respects_api_base_and_api_key_from_kwargs():
     """Test that video_content respects api_base and api_key from kwargs (simulating database entry)."""
     from litellm.videos.main import video_content