feat: add MiniMax video and image generation providers

litellm/constants.py

@@ -200,6 +200,8 @@
 KLING_DEFAULT_API_BASE = str(os.getenv("KLING_DEFAULT_API_BASE", "https://api-singapore.klingai.com/v1"))
 KLING_POLLING_TIMEOUT = int(os.getenv("KLING_POLLING_TIMEOUT", 900))
 
+MINIMAX_MEDIA_DEFAULT_API_BASE = str(os.getenv("MINIMAX_MEDIA_DEFAULT_API_BASE", "https://api.minimax.io"))
+
 ########## Networking constants ##############################################################
 _DEFAULT_TTL_FOR_HTTPX_CLIENTS = 3600  # 1 hour, re-use the same httpx client for 1 hour
 

litellm/images/main.py

@@ -394,6 +394,7 @@ def image_generation(
             litellm.LlmProviders.VERTEX_AI,
             litellm.LlmProviders.OPENROUTER,
             litellm.LlmProviders.DASHSCOPE,
+            litellm.LlmProviders.MINIMAX,
         ):
             if image_generation_config is None:
                 raise ValueError(f"image generation config is not supported for {custom_llm_provider}")

litellm/litellm_core_utils/llm_cost_calc/utils.py

@@ -1196,6 +1196,15 @@ def route_image_generation_cost_calculator(
                 model=model,
                 image_response=completion_response,
             )
+        elif custom_llm_provider == litellm.LlmProviders.MINIMAX.value:
+            from litellm.llms.minimax.cost_calculator import (
+                image_cost_calculator as minimax_image_cost_calculator,
+            )
+
+            return minimax_image_cost_calculator(
+                model=model,
+                image_response=completion_response,
+            )
         elif custom_llm_provider == litellm.LlmProviders.BLACK_FOREST_LABS.value:
             from litellm.llms.black_forest_labs.cost_calculator import (
                 image_cost_calculator as bfl_image_cost_calculator,

litellm/llms/minimax/common_utils.py

@@ -0,0 +1,31 @@
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import Any  # noqa: TID251  # base transformation contracts type these payloads as Any
+
+import litellm
+from litellm.constants import MINIMAX_MEDIA_DEFAULT_API_BASE
+from litellm.secret_managers.main import get_secret_str
+
+EMPTY_MAP: Mapping[str, Any] = MappingProxyType({})  # mutable-ok: frozen shared empty mapping
+
+
+def resolve_minimax_media_api_base(api_base: str | None) -> str:
+    return (api_base or MINIMAX_MEDIA_DEFAULT_API_BASE).rstrip("/")
+
+
+def strip_minimax_prefix(model: str) -> str:
+    return model.removeprefix("minimax/")
+
+
+def drop_none_values(values: Mapping[str, Any]) -> Mapping[str, Any]:
+    return {key: value for key, value in values.items() if value is not None}
+
+
+def minimax_bearer_headers(
+    headers: Mapping[str, Any],
+    api_key: str | None,
+) -> dict:  # mutable-ok: validate_environment contracts return dict
+    final_api_key = api_key or get_secret_str("MINIMAX_API_KEY") or litellm.api_key
+    if not final_api_key:
+        raise ValueError("MINIMAX_API_KEY is not set")
+    return {**headers, "Authorization": f"Bearer {final_api_key}", "Content-Type": "application/json"}

litellm/llms/minimax/cost_calculator.py

@@ -0,0 +1,23 @@
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from litellm.types.utils import ImageResponse
+
+
+def image_cost_calculator(
+    model: str,
+    image_response: "ImageResponse",
+) -> float:
+    import litellm
+    from litellm.types.utils import ImageResponse as _ImageResponse
+
+    if not isinstance(image_response, _ImageResponse):
+        raise ValueError(f"image_response must be of type ImageResponse got type={type(image_response)}")
+
+    model_info = litellm.get_model_info(
+        model=model,
+        custom_llm_provider=litellm.LlmProviders.MINIMAX.value,
+    )
+    output_cost_per_image: float = model_info.get("output_cost_per_image") or 0.0
+    num_images = len(image_response.data) if image_response.data else 0
+    return output_cost_per_image * num_images

litellm/llms/minimax/image_generation/__init__.py

@@ -0,0 +1,14 @@
+from litellm.llms.base_llm.image_generation.transformation import (
+    BaseImageGenerationConfig,
+)
+
+from .transformation import MinimaxImageGenerationConfig
+
+__all__ = (
+    "MinimaxImageGenerationConfig",
+    "get_minimax_image_generation_config",
+)
+
+
+def get_minimax_image_generation_config(model: str) -> BaseImageGenerationConfig:
+    return MinimaxImageGenerationConfig()

litellm/llms/minimax/image_generation/transformation.py

@@ -0,0 +1,229 @@
+from collections.abc import Mapping, Sequence
+from math import gcd
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Any  # noqa: TID251  # base transformation contracts type these payloads as Any
+
+import httpx
+
+from litellm.llms.base_llm.image_generation.transformation import (
+    BaseImageGenerationConfig,
+)
+from litellm.llms.minimax.common_utils import (
+    EMPTY_MAP,
+    drop_none_values,
+    minimax_bearer_headers,
+    resolve_minimax_media_api_base,
+    strip_minimax_prefix,
+)
+from litellm.types.llms.openai import (
+    AllMessageValues,
+    OpenAIImageGenerationOptionalParams,
+)
+from litellm.types.utils import ImageObject, ImageResponse
+
+if TYPE_CHECKING:
+    from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
+
+    LiteLLMLoggingObj = _LiteLLMLoggingObj
+else:
+    LiteLLMLoggingObj = Any
+
+MINIMAX_IMAGE_PASSTHROUGH_PARAMS = frozenset(
+    {"aspect_ratio", "width", "height", "seed", "prompt_optimizer", "subject_reference", "image_url"}
+)
+
+_V1_ERROR_HTTP_STATUS: Mapping[int, int] = MappingProxyType(
+    {1002: 429, 1004: 401, 1008: 402, 2049: 401}  # mutable-ok: frozen constant lookup table
+)
+
+_SIZE_TO_ASPECT_RATIO: Mapping[str, str] = MappingProxyType(
+    {  # mutable-ok: frozen constant lookup table
+        "1024x1024": "1:1",
+        "1280x720": "16:9",
+        "1920x1080": "16:9",
+        "720x1280": "9:16",
+        "1080x1920": "9:16",
+        "1152x864": "4:3",
+        "864x1152": "3:4",
+        "1248x832": "3:2",
+        "832x1248": "2:3",
+        "1344x576": "21:9",
+    }
+)
+
+_RESERVED_REQUEST_KEYS = frozenset({"model", "prompt", "user", "size", "extra_body"})
+
+
+def _aspect_ratio_from_size(size: str) -> str:
+    mapped = _SIZE_TO_ASPECT_RATIO.get(size)
+    if mapped:
+        return mapped
+    width, _, height = size.partition("x")
+    try:
+        parsed_width, parsed_height = int(width), int(height)
+    except ValueError:
+        return size.replace("x", ":")
+    if parsed_width <= 0 or parsed_height <= 0:
+        return size.replace("x", ":")
+    divisor = gcd(parsed_width, parsed_height)
+    return f"{parsed_width // divisor}:{parsed_height // divisor}"
+
+
+class MinimaxImageGenerationConfig(BaseImageGenerationConfig):
+    def get_supported_openai_params(
+        self, model: str
+    ) -> list[OpenAIImageGenerationOptionalParams]:  # mutable-ok: BaseImageGenerationConfig contract returns list
+        return ["n", "response_format", "size"]
+
+    def get_complete_url(
+        self,
+        api_base: str | None,
+        api_key: str | None,
+        model: str,
+        optional_params: Mapping[str, Any],
+        litellm_params: Mapping[str, Any],
+        stream: bool | None = None,
+    ) -> str:
+        return f"{resolve_minimax_media_api_base(api_base)}/v1/image_generation"
+
+    def validate_environment(
+        self,
+        headers: Mapping[str, Any],
+        model: str,
+        messages: Sequence[AllMessageValues],
+        optional_params: Mapping[str, Any],
+        litellm_params: Mapping[str, Any],
+        api_key: str | None = None,
+        api_base: str | None = None,
+    ) -> dict:  # mutable-ok: BaseImageGenerationConfig contract returns dict
+        return minimax_bearer_headers(headers, api_key)
+
+    def map_openai_params(
+        self,
+        non_default_params: Mapping[str, Any],
+        optional_params: Mapping[str, Any],
+        model: str,
+        drop_params: bool,
+    ) -> dict:  # mutable-ok: BaseImageGenerationConfig contract returns dict
+        supported = self.get_supported_openai_params(model)
+        unsupported = tuple(
+            key
+            for key in non_default_params
+            if key != "size" and key not in supported and key not in MINIMAX_IMAGE_PASSTHROUGH_PARAMS
+        )
+        if unsupported and not drop_params:
+            raise ValueError(
+                f"Parameters {list(unsupported)} are not supported for model {model}. Supported "
+                f"parameters are {supported}. Set drop_params=True to drop unsupported parameters."
+            )
+        size = non_default_params.get("size")
+        aspect_ratio = _aspect_ratio_from_size(size) if isinstance(size, str) and size else None
+        return {
+            **optional_params,
+            **{key: value for key, value in non_default_params.items() if key != "size" and key not in unsupported},
+            **({"aspect_ratio": aspect_ratio} if aspect_ratio else EMPTY_MAP),
+        }
+
+    def transform_image_generation_request(
+        self,
+        model: str,
+        prompt: str,
+        optional_params: Mapping[str, Any],
+        litellm_params: Mapping[str, Any],
+        headers: Mapping[str, Any],
+    ) -> dict:  # mutable-ok: BaseImageGenerationConfig contract returns dict
+        extra_body = optional_params.get("extra_body")
+        params = {
+            **optional_params,
+            **(extra_body if isinstance(extra_body, dict) else EMPTY_MAP),
+        }
+        forwarded = drop_none_values(
+            {key: value for key, value in params.items() if key not in _RESERVED_REQUEST_KEYS and key != "image_url"}
+        )
+        image_url = params.get("image_url")
+        reference = (
+            ({"type": "character", "image_file": image_url.strip()},)  # mutable-ok: MiniMax wire shape is a JSON object
+            if "subject_reference" not in forwarded and isinstance(image_url, str) and image_url.strip()
+            else None
+        )
+        return dict(
+            drop_none_values(
+                {
+                    "model": strip_minimax_prefix(model),
+                    "prompt": prompt,
+                    **forwarded,
+                    "subject_reference": reference if reference is not None else forwarded.get("subject_reference"),
+                }
+            )
+        )
+
+    def transform_image_generation_response(
+        self,
+        model: str,
+        raw_response: httpx.Response,
+        model_response: ImageResponse,
+        logging_obj: LiteLLMLoggingObj,
+        request_data: Mapping[str, Any],
+        optional_params: Mapping[str, Any],
+        litellm_params: Mapping[str, Any],
+        encoding: Any,
+        api_key: str | None = None,
+        json_mode: bool | None = None,
+    ) -> ImageResponse:
+        self._raise_for_status(raw_response)
+        response_data = self._parse_json(raw_response)
+        self._raise_for_minimax_error(raw_response, response_data)
+        data = response_data.get("data") or EMPTY_MAP
+        urls = data.get("image_urls") or ()
+        b64_images = data.get("image_base64") or ()
+        images = [
+            *(ImageObject(url=url, b64_json=None) for url in urls if isinstance(url, str) and url),
+            *(ImageObject(url=None, b64_json=b64) for b64 in b64_images if isinstance(b64, str) and b64),
+        ]
+        if not images:
+            raise ValueError(f"MiniMax image generation returned no images: {response_data}")
+        model_response.data = images
+        return model_response
+
+    def _raise_for_status(self, raw_response: httpx.Response) -> None:
+        if raw_response.is_success:
+            return
+        raise self.get_error_class(
+            error_message=self._error_message_from_body(raw_response),
+            status_code=raw_response.status_code,
+            headers=raw_response.headers,
+        )
+
+    @staticmethod
+    def _error_message_from_body(raw_response: httpx.Response) -> str:
+        try:
+            body = raw_response.json()
+        except Exception:
+            return raw_response.text
+        if not isinstance(body, dict):
+            return raw_response.text
+        error = body.get("error")
+        message = error.get("message") if isinstance(error, dict) else None
+        if not (isinstance(message, str) and message):
+            base_resp = body.get("base_resp")
+            message = base_resp.get("status_msg") if isinstance(base_resp, dict) else None
+        return message if isinstance(message, str) and message else raw_response.text
+
+    @staticmethod
+    def _parse_json(raw_response: httpx.Response) -> Mapping[str, Any]:
+        try:
+            return raw_response.json()
+        except Exception as e:
+            raise ValueError(f"Error parsing MiniMax image generation response: {e}") from e
+
+    def _raise_for_minimax_error(self, raw_response: httpx.Response, response_data: Mapping[str, Any]) -> None:
+        base_resp = response_data.get("base_resp") or EMPTY_MAP
+        status_code = base_resp.get("status_code")
+        if status_code in (None, 0):
+            return
+        message = base_resp.get("status_msg") or "MiniMax API returned an error"
+        raise self.get_error_class(
+            error_message=f"{message} ({status_code})",
+            status_code=_V1_ERROR_HTTP_STATUS.get(status_code, 400),
+            headers=raw_response.headers,
+        )

litellm/llms/minimax/videos/__init__.py

@@ -0,0 +1,3 @@
+from .transformation import MinimaxVideoConfig
+
+__all__ = ("MinimaxVideoConfig",)