From 74bab168c9d6efcc8081afceeb94ae740f0d4bca Mon Sep 17 00:00:00 2001 From: octo-patch <266937838+octo-patch@users.noreply.github.com> Date: Fri, 31 Jul 2026 19:58:43 +0800 Subject: [PATCH] feat(minimax): add video generation support --- litellm/llms/minimax/videos/__init__.py | 5 + litellm/llms/minimax/videos/transformation.py | 435 ++++++++++++++++++ litellm/utils.py | 4 + .../test_minimax_video_transformation.py | 144 ++++++ 4 files changed, 588 insertions(+) create mode 100644 litellm/llms/minimax/videos/__init__.py create mode 100644 litellm/llms/minimax/videos/transformation.py create mode 100644 tests/test_litellm/llms/minimax/videos/test_minimax_video_transformation.py diff --git a/litellm/llms/minimax/videos/__init__.py b/litellm/llms/minimax/videos/__init__.py new file mode 100644 index 00000000000..1dcd26b077e --- /dev/null +++ b/litellm/llms/minimax/videos/__init__.py @@ -0,0 +1,5 @@ +"""MiniMax video generation transformation.""" + +from .transformation import MinimaxVideoConfig + +__all__ = ["MinimaxVideoConfig"] diff --git a/litellm/llms/minimax/videos/transformation.py b/litellm/llms/minimax/videos/transformation.py new file mode 100644 index 00000000000..e42b849e732 --- /dev/null +++ b/litellm/llms/minimax/videos/transformation.py @@ -0,0 +1,435 @@ +"""MiniMax v1 video generation transformations.""" + +from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union +from urllib.parse import quote, urlsplit, urlunsplit + +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.llms.custom_httpx.http_handler import ( + AsyncHTTPHandler, + HTTPHandler, + _get_httpx_client, + get_async_httpx_client, +) +from litellm.secret_managers.main import get_secret_str +from litellm.types.router import GenericLiteLLMParams +from litellm.types.videos.main import VideoCreateOptionalRequestParams, VideoObject +from litellm.types.videos.utils import ( + encode_video_id_with_provider, + extract_original_video_id, +) + +if TYPE_CHECKING: + from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj + + LiteLLMLoggingObj = _LiteLLMLoggingObj +else: + LiteLLMLoggingObj = Any + + +class MinimaxVideoConfig(BaseVideoConfig): + """Configuration for MiniMax's v1 text-to-video and image-to-video API.""" + + def get_supported_openai_params(self, model: str) -> list: + return [ + "model", + "prompt", + "input_reference", + "seconds", + "size", + "user", + "extra_headers", + "extra_body", + "prompt_optimizer", + "fast_pretreatment", + "duration", + "resolution", + "callback_url", + ] + + def map_openai_params( + self, + video_create_optional_params: VideoCreateOptionalRequestParams, + model: str, + drop_params: bool, + ) -> Dict: + mapped_params: Dict[str, Any] = {} + + for key, value in video_create_optional_params.items(): + if value is None or key in {"model", "prompt", "extra_headers", "user"}: + continue + if key == "input_reference": + mapped_params["first_frame_image"] = value + elif key == "seconds": + mapped_params["duration"] = self._coerce_duration(value) + elif key == "size": + mapped_params["resolution"] = value + elif key != "extra_body": + mapped_params[key] = value + + extra_body = video_create_optional_params.get("extra_body") + if isinstance(extra_body, dict): + mapped_params.update({key: value for key, value in extra_body.items() if value is not None}) + + return mapped_params + + def validate_environment( + self, + headers: dict, + model: str, + api_key: Optional[str] = None, + litellm_params: Optional[GenericLiteLLMParams] = None, + ) -> dict: + if litellm_params and litellm_params.api_key: + api_key = api_key or litellm_params.api_key + + api_key = api_key or litellm.api_key or get_secret_str("MINIMAX_API_KEY") + if not api_key: + raise ValueError( + "MiniMax API key is required. Set MINIMAX_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 regional MiniMax v1 root used by all video operations.""" + base_url = api_base or get_secret_str("MINIMAX_API_BASE") or "https://api.minimax.io/v1" + base_url = base_url.rstrip("/") + for suffix in ("/video_generation", "/query/video_generation", "/files/retrieve"): + if base_url.endswith(suffix): + base_url = base_url[: -len(suffix)] + break + if not base_url.endswith("/v1"): + base_url = f"{base_url}/v1" + return base_url + + 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] = {"model": model, "prompt": prompt} + request_data.update(video_create_optional_request_params) + request_data.pop("extra_headers", None) + request_data.pop("extra_body", None) + request_data.pop("user", None) + return request_data, [], f"{api_base.rstrip('/')}/video_generation" + + 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: + response_data = self._parse_json_response(raw_response) + self._raise_for_provider_error(raw_response, response_data) + task_id = response_data.get("task_id") + if task_id is None: + raise ValueError("MiniMax did not return a task_id for video generation") + + video_data: Dict[str, Any] = { + "id": str(task_id), + "object": "video", + "status": self._map_status(response_data.get("status", "queueing")), + "model": model, + } + self._add_request_metadata(video_data, request_data) + video_obj = VideoObject(**video_data) + self._wrap_video_id(video_obj, custom_llm_provider, model) + video_obj.usage = self._usage_from_video(video_obj) + 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]: + task_id = quote(extract_original_video_id(video_id), safe="") + return f"{api_base.rstrip('/')}/query/video_generation?task_id={task_id}", {} + + def transform_video_content_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + ) -> bytes: + query_data = self._parse_json_response(raw_response) + self._raise_for_provider_error(raw_response, query_data) + file_id = query_data.get("file_id") + if file_id is None: + status = self._map_status(query_data.get("status", "processing")) + raise ValueError(f"MiniMax video is not ready for download (status: {status})") + + headers = self._request_headers(raw_response) + api_base = self._api_base_from_response(raw_response) + file_url = f"{api_base}/files/retrieve?file_id={quote(str(file_id), safe='')}" + client: HTTPHandler = _get_httpx_client() + file_response = client.get(file_url, headers=headers) + self._raise_for_status(file_response) + if self._is_binary_response(file_response): + return file_response.content + + file_data = self._parse_json_response(file_response) + self._raise_for_provider_error(file_response, file_data) + download_url = self._get_download_url(file_data) + if not download_url: + raise ValueError("MiniMax did not return a video download URL") + + video_response = client.get(download_url) + self._raise_for_status(video_response) + return video_response.content + + async def async_transform_video_content_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + ) -> bytes: + query_data = self._parse_json_response(raw_response) + self._raise_for_provider_error(raw_response, query_data) + file_id = query_data.get("file_id") + if file_id is None: + status = self._map_status(query_data.get("status", "processing")) + raise ValueError(f"MiniMax video is not ready for download (status: {status})") + + headers = self._request_headers(raw_response) + api_base = self._api_base_from_response(raw_response) + file_url = f"{api_base}/files/retrieve?file_id={quote(str(file_id), safe='')}" + client: AsyncHTTPHandler = get_async_httpx_client(llm_provider=litellm.LlmProviders.MINIMAX) + file_response = await client.get(file_url, headers=headers) + self._raise_for_status(file_response) + if self._is_binary_response(file_response): + return file_response.content + + file_data = self._parse_json_response(file_response) + self._raise_for_provider_error(file_response, file_data) + download_url = self._get_download_url(file_data) + if not download_url: + raise ValueError("MiniMax did not return a video download URL") + + video_response = await client.get(download_url) + self._raise_for_status(video_response) + return video_response.content + + def transform_video_status_retrieve_request( + self, + video_id: str, + api_base: str, + litellm_params: GenericLiteLLMParams, + headers: dict, + ) -> Tuple[str, Dict]: + task_id = quote(extract_original_video_id(video_id), safe="") + return f"{api_base.rstrip('/')}/query/video_generation?task_id={task_id}", {} + + def transform_video_status_retrieve_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + custom_llm_provider: Optional[str] = None, + ) -> VideoObject: + response_data = self._parse_json_response(raw_response) + self._raise_for_provider_error(raw_response, response_data) + task_id = response_data.get("task_id") + if task_id is None: + raise ValueError("MiniMax did not return a task_id for video status") + model = response_data.get("model") + video_data: Dict[str, Any] = { + "id": str(task_id), + "object": "video", + "status": self._map_status(response_data.get("status", "processing")), + "model": model, + } + if response_data.get("status") and self._map_status(response_data["status"]) == "failed": + video_data["error"] = { + "code": "generation_failed", + "message": str(response_data.get("status")), + } + if response_data.get("duration") is not None: + video_data["seconds"] = str(response_data["duration"]) + if response_data.get("resolution") is not None: + video_data["size"] = str(response_data["resolution"]) + + video_obj = VideoObject(**video_data) + self._wrap_video_id(video_obj, custom_llm_provider, model) + return video_obj + + 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 the MiniMax v1 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 the MiniMax v1 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 the MiniMax v1 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 the MiniMax v1 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 the MiniMax v1 API") + + def transform_video_delete_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + ) -> VideoObject: + raise NotImplementedError("Video deletion is not supported by the MiniMax v1 API") + + def get_error_class( + self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers] + ) -> BaseLLMException: + return BaseLLMException(status_code=status_code, message=error_message, headers=headers) + + @staticmethod + def _coerce_duration(value: Any) -> Any: + try: + return int(float(value)) + except (TypeError, ValueError): + return value + + @staticmethod + def _map_status(status: Any) -> str: + normalized = str(status or "").strip().lower().replace(" ", "_") + if normalized in {"success", "succeeded", "completed", "complete"}: + return "completed" + if normalized in {"fail", "failed", "error", "cancelled", "canceled"}: + return "failed" + if normalized in {"queueing", "queued", "preparing", "pending"}: + return "queued" + return "in_progress" + + @staticmethod + def _add_request_metadata(video_data: Dict[str, Any], request_data: Optional[Dict]) -> None: + if not request_data: + return + if request_data.get("duration") is not None: + video_data["seconds"] = str(request_data["duration"]) + if request_data.get("resolution") is not None: + video_data["size"] = str(request_data["resolution"]) + + @staticmethod + def _usage_from_video(video_obj: VideoObject) -> Dict[str, Any]: + if video_obj.seconds is None: + return {} + try: + return {"duration_seconds": float(video_obj.seconds)} + except (TypeError, ValueError): + return {} + + @staticmethod + def _wrap_video_id(video_obj: VideoObject, provider: Optional[str], model: Optional[str]) -> None: + if provider and video_obj.id: + video_obj.id = encode_video_id_with_provider(video_obj.id, provider, model) + + @staticmethod + def _parse_json_response(raw_response: httpx.Response) -> Dict[str, Any]: + try: + return raw_response.json() + except Exception as exc: + raise ValueError(f"MiniMax returned an invalid JSON response: {exc}") from exc + + def _raise_for_provider_error(self, raw_response: httpx.Response, response_data: Dict[str, Any]) -> None: + self._raise_for_status(raw_response) + base_resp = response_data.get("base_resp") or {} + status_code = base_resp.get("status_code") + if status_code not in (None, 0, "0"): + message = base_resp.get("status_msg") or "MiniMax video request failed" + raise self.get_error_class(str(message), raw_response.status_code, raw_response.headers) + + def _raise_for_status(self, raw_response: httpx.Response) -> None: + if raw_response.status_code >= 400: + raise self.get_error_class(raw_response.text, raw_response.status_code, raw_response.headers) + + @staticmethod + def _request_headers(raw_response: httpx.Response) -> Dict[str, str]: + request = getattr(raw_response, "request", None) + request_headers = getattr(request, "headers", None) + if isinstance(request_headers, (dict, httpx.Headers)): + authorization = request_headers.get("Authorization") + if authorization: + return {"Authorization": authorization} + return {} + + @staticmethod + def _api_base_from_response(raw_response: httpx.Response) -> str: + request = getattr(raw_response, "request", None) + request_url = getattr(request, "url", None) or getattr(raw_response, "url", None) + if request_url is None: + return "https://api.minimax.io/v1" + parsed = urlsplit(str(request_url)) + path = parsed.path + v1_index = path.find("/v1/") + root_path = path[: v1_index + len("/v1")] if v1_index >= 0 else "/v1" + return urlunsplit((parsed.scheme, parsed.netloc, root_path, "", "")) + + @staticmethod + def _is_binary_response(raw_response: httpx.Response) -> bool: + content_type = raw_response.headers.get("content-type", "") + return content_type.startswith("video/") or content_type == "application/octet-stream" + + @staticmethod + def _get_download_url(response_data: Dict[str, Any]) -> Optional[str]: + file_data = response_data.get("file") + if isinstance(file_data, dict): + for key in ("download_url", "url"): + if file_data.get(key): + return str(file_data[key]) + for key in ("download_url", "url"): + if response_data.get(key): + return str(response_data[key]) + return None diff --git a/litellm/utils.py b/litellm/utils.py index ad661fd3d16..bca58da6cf9 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -8710,6 +8710,10 @@ class ProviderConfigManager: from litellm.llms.runwayml.videos.transformation import RunwayMLVideoConfig return RunwayMLVideoConfig() + elif LlmProviders.MINIMAX == provider: + from litellm.llms.minimax.videos.transformation import MinimaxVideoConfig + + return MinimaxVideoConfig() return None @staticmethod diff --git a/tests/test_litellm/llms/minimax/videos/test_minimax_video_transformation.py b/tests/test_litellm/llms/minimax/videos/test_minimax_video_transformation.py new file mode 100644 index 00000000000..2b686d44634 --- /dev/null +++ b/tests/test_litellm/llms/minimax/videos/test_minimax_video_transformation.py @@ -0,0 +1,144 @@ +"""Tests for MiniMax v1 video generation transformations.""" + +from unittest.mock import Mock, patch + +import httpx + +import litellm +from litellm.llms.minimax.videos.transformation import MinimaxVideoConfig +from litellm.types.router import GenericLiteLLMParams +from litellm.types.videos.main import VideoObject +from litellm.types.videos.utils import decode_video_id_with_provider, encode_video_id_with_provider +from litellm.utils import ProviderConfigManager + + +class TestMinimaxVideoTransformation: + def setup_method(self): + self.config = MinimaxVideoConfig() + self.logging_obj = Mock() + + def test_provider_config_is_registered(self): + config = ProviderConfigManager.get_provider_video_config( + model="MiniMax-Hailuo-2.3", + provider=litellm.LlmProviders.MINIMAX, + ) + assert isinstance(config, MinimaxVideoConfig) + + def test_transform_create_request_maps_text_and_image_parameters(self): + params = self.config.map_openai_params( + { + "input_reference": "https://example.com/frame.png", + "seconds": "6", + "size": "768P", + "extra_body": {"prompt_optimizer": True}, + }, + model="MiniMax-Hailuo-2.3", + drop_params=False, + ) + data, files, url = self.config.transform_video_create_request( + model="MiniMax-Hailuo-2.3", + prompt="A city at sunrise", + api_base="https://api.minimax.io/v1", + video_create_optional_request_params=params, + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + + assert data == { + "model": "MiniMax-Hailuo-2.3", + "prompt": "A city at sunrise", + "first_frame_image": "https://example.com/frame.png", + "duration": 6, + "resolution": "768P", + "prompt_optimizer": True, + } + assert files == [] + assert url == "https://api.minimax.io/v1/video_generation" + + def test_create_response_wraps_task_id_and_maps_status(self): + response = httpx.Response( + 200, + json={"task_id": "task-123", "base_resp": {"status_code": 0}}, + ) + result = self.config.transform_video_create_response( + model="MiniMax-Hailuo-2.3", + raw_response=response, + logging_obj=self.logging_obj, + custom_llm_provider="minimax", + request_data={"duration": 6, "resolution": "768P"}, + ) + + assert isinstance(result, VideoObject) + decoded = decode_video_id_with_provider(result.id) + assert decoded["custom_llm_provider"] == "minimax" + assert decoded["model_id"] == "MiniMax-Hailuo-2.3" + assert decoded["video_id"] == "task-123" + assert result.status == "queued" + assert result.seconds == "6" + assert result.size == "768P" + + def test_status_request_and_response(self): + encoded_id = encode_video_id_with_provider("task-123", "minimax", "MiniMax-Hailuo-2.3") + url, data = self.config.transform_video_status_retrieve_request( + video_id=encoded_id, + api_base="https://api.minimaxi.com/v1", + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + assert url == "https://api.minimaxi.com/v1/query/video_generation?task_id=task-123" + assert data == {} + + response = httpx.Response( + 200, + json={ + "task_id": "task-123", + "status": "Success", + "file_id": "file-123", + "base_resp": {"status_code": 0}, + }, + ) + result = self.config.transform_video_status_retrieve_response( + raw_response=response, + logging_obj=self.logging_obj, + custom_llm_provider="minimax", + ) + decoded = decode_video_id_with_provider(result.id) + assert decoded["video_id"] == "task-123" + assert decoded["custom_llm_provider"] == "minimax" + assert result.status == "completed" + + def test_content_response_retrieves_file_and_downloads_video(self): + query_request = httpx.Request( + "GET", + "https://api.minimax.io/v1/query/video_generation?task_id=task-123", + headers={"Authorization": "Bearer test-key"}, + ) + query_response = httpx.Response( + 200, + json={"task_id": "task-123", "status": "Success", "file_id": "file-123"}, + request=query_request, + ) + file_response = httpx.Response( + 200, + json={"file": {"download_url": "https://cdn.example.com/video.mp4"}}, + request=httpx.Request("GET", "https://api.minimax.io/v1/files/retrieve"), + ) + video_response = httpx.Response( + 200, + content=b"video-bytes", + headers={"content-type": "video/mp4"}, + request=httpx.Request("GET", "https://cdn.example.com/video.mp4"), + ) + client = Mock() + client.get.side_effect = [file_response, video_response] + + with patch( + "litellm.llms.minimax.videos.transformation._get_httpx_client", + return_value=client, + ): + result = self.config.transform_video_content_response(query_response, self.logging_obj) + + assert result == b"video-bytes" + assert client.get.call_args_list[0].args[0] == ("https://api.minimax.io/v1/files/retrieve?file_id=file-123") + assert client.get.call_args_list[0].kwargs["headers"]["Authorization"] == "Bearer test-key" + assert client.get.call_args_list[1].args[0] == "https://cdn.example.com/video.mp4"