scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-03 02:22:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 82

feat(xai): add grok imagine image generation, edit, and video download

Browse source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
hx 2026-09-27 17:48:31 +08:00
parent f4308bc124
commit 39b3d4346d
27 changed files with 2094 additions and 114 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
litellm/images/main.py
View file

@ -384,6 +384,7 @@ def image_generation(
litellm.LlmProviders.QWENCLOUD,
litellm.LlmProviders.QWEN_AI_PLATFORM,
litellm.LlmProviders.EDENAI,
litellm.LlmProviders.XAI,
):
if image_generation_config is None:
raise ValueError(f"image generation config is not supported for {custom_llm_provider}")
@ -393,7 +394,7 @@ def image_generation(
litellm_params_dict["api_base"] = _api_base
return llm_http_handler.image_generation_handler(
api_key=api_key,
api_key=api_key or dynamic_api_key,
model=model,
prompt=prompt,
image_generation_provider_config=image_generation_config,

3
litellm/llms/xai/image_edit/__init__.py Normal file
View file

@ -0,0 +1,3 @@
from .transformation import XAIImageEditConfig
__all__ = ["XAIImageEditConfig"] # mutable-ok: provider JSON body and base-class dict signature

251
litellm/llms/xai/image_edit/transformation.py Normal file
View file

@ -0,0 +1,251 @@
import base64
from io import BufferedReader, BytesIO
from typing import TYPE_CHECKING, Final
import httpx
from httpx._types import RequestFiles
from litellm.constants import XAI_API_BASE
from litellm.exceptions import AuthenticationError
from litellm.images.utils import ImageEditRequestUtils
from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig
from litellm.llms.xai.common_utils import XAIModelInfo
from litellm.secret_managers.main import get_secret_str
from litellm.types.images.main import ImageEditOptionalRequestParams
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import FileTypes, ImageObject, ImageResponse
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
_SIZE_TO_ASPECT_RATIO: Final = { # mutable-ok: provider JSON body and base-class dict signature
"1024x1024": "1:1",
"1792x1024": "16:9",
"1024x1792": "9:16",
"1536x1024": "3:2",
"1024x1536": "2:3",
"1280x720": "16:9",
"720x1280": "9:16",
"1920x1080": "16:9",
"1080x1920": "9:16",
}
_XAI_NATIVE_PARAMS: Final = frozenset({"aspect_ratio", "n", "resolution"})
def _read_seekable(image: BytesIO | BufferedReader) -> bytes:
current_pos: Final = image.tell()
image.seek(0)
data: Final = image.read()
image.seek(current_pos)
return data
class XAIImageEditConfig(BaseImageEditConfig):
def get_supported_openai_params(
self, model: str
) -> list: # mutable-ok: provider JSON body and base-class dict signature
return ["n", "response_format", "size", "user"] # mutable-ok: provider JSON body and base-class dict signature
def map_openai_params(
self,
image_edit_optional_params: ImageEditOptionalRequestParams,
model: str,
drop_params: bool,
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
supported: Final = frozenset(self.get_supported_openai_params(model))
allowed: Final = supported | _XAI_NATIVE_PARAMS
raw = image_edit_optional_params
incoming: Final = dict(raw) # mutable-ok: provider JSON body and base-class dict signature
unknown: Final = tuple(key for key in incoming if key not in allowed)
if unknown and not drop_params:
raise ValueError(
f"Parameter {unknown[0]} is not supported for model {model}. "
f"Supported parameters are {sorted(allowed)}. "
"Set drop_params=True to drop unsupported parameters."
)
pairs = ((key, value) for key, value in incoming.items() if key in allowed)
mapped: Final = dict(pairs) # mutable-ok: provider JSON body and base-class dict signature
size: Final = mapped.get("size")
aspect_ratio: Final = mapped.get("aspect_ratio") or (
_SIZE_TO_ASPECT_RATIO.get(str(size), "1:1") if size else None
)
n: Final = mapped.get("n")
resolution: Final = mapped.get("resolution")
aspect = {"aspect_ratio": aspect_ratio} if aspect_ratio is not None else None # mutable-ok: provider JSON body
count = {"n": int(n)} if n is not None else None # mutable-ok: provider JSON body
quality = {"resolution": resolution} if resolution is not None else None # mutable-ok: provider JSON body
aspect_ratio_field: Final = aspect or {} # mutable-ok: provider JSON body and base-class dict signature
n_field: Final = count or {} # mutable-ok: provider JSON body and base-class dict signature
resolution_field: Final = quality or {} # mutable-ok: provider JSON body and base-class dict signature
return { # mutable-ok: provider JSON body and base-class dict signature
**aspect_ratio_field,
**n_field,
**resolution_field,
} # mutable-ok: provider JSON body and base-class dict signature
def use_multipart_form_data(self) -> bool:
return False
def get_complete_url(
self,
model: str,
api_base: str | None,
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> str:
from litellm.llms.xai.oauth import XAIOAuthAuthenticator, should_use_xai_oauth
api_key: Final = litellm_params.get("api_key") if isinstance(litellm_params, dict) else None
resolved_base: Final = (
XAIOAuthAuthenticator().get_api_base()
if should_use_xai_oauth(litellm_params) and not XAIModelInfo.get_api_key(api_key)
else (api_base or get_secret_str("XAI_API_BASE") or get_secret_str("XAI_OAUTH_API_BASE") or XAI_API_BASE)
)
base: Final = (resolved_base or XAI_API_BASE).rstrip("/")
if base.endswith("/v1"):
return f"{base}/images/edits"
return f"{base}/v1/images/edits"
def validate_environment(
self,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
model: str,
api_key: str | None = None,
litellm_params: dict | None = None, # mutable-ok: provider JSON body and base-class dict signature
api_base: str | None = None,
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
from litellm.llms.xai.oauth import (
XAIOAuthAuthenticator,
XAIOAuthError,
should_use_xai_oauth,
)
params: Final = litellm_params or {} # mutable-ok: provider JSON body and base-class dict signature
dynamic_api_key: Final = XAIModelInfo.get_api_key(api_key)
if should_use_xai_oauth(params) and not dynamic_api_key:
try:
headers["Authorization"] = f"Bearer {XAIOAuthAuthenticator().get_access_token()}"
except XAIOAuthError as exc:
raise AuthenticationError(
model=model,
llm_provider="xai",
message=str(exc),
) from exc
else:
if not dynamic_api_key:
raise AuthenticationError(
model=model,
llm_provider="xai",
message=(
"Missing xAI credentials for image edit. Pass api_key / XAI_API_KEY, or set use_xai_oauth=True."
),
)
headers["Authorization"] = f"Bearer {dynamic_api_key}"
if "content-type" not in headers and "Content-Type" not in headers:
headers["Content-Type"] = "application/json"
return headers
def transform_image_edit_request(
self,
model: str,
prompt: str | None,
image: FileTypes | None,
image_edit_optional_request_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[dict, RequestFiles]: # mutable-ok: provider JSON body and base-class dict signature
if image is None:
raise ValueError("xAI image edit requires at least one reference image.")
image_payloads: Final = tuple(self._to_image_url(item) for item in self._as_image_list(image))
if not image_payloads:
raise ValueError("xAI image edit requires at least one reference image.")
n: Final = image_edit_optional_request_params.get("n")
prompt_body = {"prompt": prompt} if prompt is not None else None # mutable-ok: provider JSON body
many = {"images": list(image_payloads)} # mutable-ok: provider JSON body
one = image_payloads[0]
image_body = {"image": one} if len(image_payloads) == 1 else many # mutable-ok: provider JSON body
prompt_field: Final = prompt_body or {} # mutable-ok: provider JSON body and base-class dict signature
image_field: Final = image_body # mutable-ok: provider JSON body and base-class dict signature
request: Final[dict[str, object]] = { # mutable-ok: provider JSON body and base-class dict signature
"model": XAIModelInfo.get_base_model(model) or model,
**prompt_field,
**image_field,
**{ # mutable-ok: provider JSON body and base-class dict signature
key: image_edit_optional_request_params[key]
for key in ("aspect_ratio", "resolution")
if image_edit_optional_request_params.get(key) is not None
},
**({"n": int(n)} if n is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
}
return request, [] # mutable-ok: provider JSON body and base-class dict signature
def transform_image_edit_response(
self,
model: str,
raw_response: httpx.Response,
logging_obj: "LiteLLMLoggingObj",
) -> ImageResponse:
try:
response_data: Final = raw_response.json()
except Exception:
raise self.get_error_class(
error_message=raw_response.text,
status_code=raw_response.status_code,
headers=raw_response.headers,
)
images: Final = tuple(
ImageObject(
url=item.get("url"),
b64_json=item.get("b64_json") or item.get("b64"),
)
for item in response_data.get("data") or ()
if isinstance(item, dict)
)
if not images:
raise self.get_error_class(
error_message=f"xAI image edit returned no image data: {response_data}",
status_code=raw_response.status_code,
headers=raw_response.headers,
)
return ImageResponse(data=list(images)) # mutable-ok: provider JSON body and base-class dict signature
def _as_image_list(
self, image: FileTypes | list[FileTypes]
) -> tuple[FileTypes, ...]: # mutable-ok: provider JSON body and base-class dict signature
if isinstance(image, list):
return tuple(item for item in image if item is not None)
return (image,)
def _to_image_url(
self, image: FileTypes
) -> dict[str, str]: # mutable-ok: provider JSON body and base-class dict signature
if isinstance(image, str):
return {"url": image} # mutable-ok: provider JSON body and base-class dict signature
if isinstance(image, dict):
if image.get("url"):
return {"url": str(image["url"])} # mutable-ok: provider JSON body and base-class dict signature
if image.get("file_id"):
file_id = str(image["file_id"])
return {"file_id": file_id} # mutable-ok: provider JSON body and base-class dict signature
mime: Final = ImageEditRequestUtils.get_image_content_type(image)
encoded: Final = base64.b64encode(self._read_all_bytes(image)).decode("utf-8")
return {"url": f"data:{mime};base64,{encoded}"} # mutable-ok: provider JSON body and base-class dict signature
def _read_all_bytes(self, image: FileTypes) -> bytes:
if isinstance(image, bytes):
return image
if isinstance(image, bytearray):
return bytes(image)
if isinstance(image, (BytesIO, BufferedReader)):
return _read_seekable(image)
if hasattr(image, "read"):
raw: Final = image.read()
if isinstance(raw, str):
return raw.encode("utf-8")
return bytes(raw)
raise ValueError(f"Unsupported image input type for xAI image edit: {type(image)}")

11
litellm/llms/xai/image_generation/__init__.py Normal file
View file

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

200
litellm/llms/xai/image_generation/transformation.py Normal file
View file

@ -0,0 +1,200 @@
from typing import TYPE_CHECKING, Final
import httpx
from litellm.constants import XAI_API_BASE
from litellm.exceptions import AuthenticationError
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
)
from litellm.llms.xai.common_utils import XAIModelInfo
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import (
AllMessageValues,
OpenAIImageGenerationOptionalParams,
)
from litellm.types.utils import ImageObject, ImageResponse
if TYPE_CHECKING:
import tiktoken
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
_SIZE_TO_ASPECT_RATIO: Final = { # mutable-ok: provider JSON body and base-class dict signature
"1024x1024": "1:1",
"1792x1024": "16:9",
"1024x1792": "9:16",
"1536x1024": "3:2",
"1024x1536": "2:3",
"1280x720": "16:9",
"720x1280": "9:16",
"1920x1080": "16:9",
"1080x1920": "9:16",
}
_XAI_NATIVE_PARAMS: Final = frozenset({"aspect_ratio", "n"})
class XAIImageGenerationConfig(BaseImageGenerationConfig):
def get_supported_openai_params(
self, model: str
) -> list[OpenAIImageGenerationOptionalParams]: # mutable-ok: provider JSON body and base-class dict signature
return ["n", "response_format", "size", "user"] # mutable-ok: provider JSON body and base-class dict signature
def map_openai_params(
self,
non_default_params: dict, # mutable-ok: provider JSON body and base-class dict signature
optional_params: dict, # mutable-ok: provider JSON body and base-class dict signature
model: str,
drop_params: bool,
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
supported_params: Final = frozenset(self.get_supported_openai_params(model))
allowed: Final = supported_params | _XAI_NATIVE_PARAMS
unknown: Final = tuple(key for key in non_default_params if key not in optional_params and key not in allowed)
if unknown and not drop_params:
raise ValueError(
f"Parameter {unknown[0]} is not supported for model {model}. "
f"Supported parameters are {sorted(allowed)}. "
"Set drop_params=True to drop unsupported parameters."
)
pairs = ((k, v) for k, v in non_default_params.items() if k in allowed)
native: Final = dict(pairs) # mutable-ok: provider JSON body and base-class dict signature
merged: Final = {**optional_params, **native} # mutable-ok: provider JSON body and base-class dict signature
size: Final = merged.get("size")
aspect_ratio: Final = merged.get("aspect_ratio") or (
_SIZE_TO_ASPECT_RATIO.get(str(size), "1:1") if size else None
)
n: Final = merged.get("n")
aspect = {"aspect_ratio": aspect_ratio} if aspect_ratio is not None else None # mutable-ok: provider JSON body
count = {"n": int(n)} if n is not None else None # mutable-ok: provider JSON body
aspect_ratio_field: Final = aspect or {} # mutable-ok: provider JSON body and base-class dict signature
n_field: Final = count or {} # mutable-ok: provider JSON body and base-class dict signature
return {**aspect_ratio_field, **n_field} # mutable-ok: provider JSON body and base-class dict signature
def get_complete_url(
self,
api_base: str | None,
api_key: str | None,
model: str,
optional_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
stream: bool | None = None,
) -> str:
from litellm.llms.xai.oauth import XAIOAuthAuthenticator, should_use_xai_oauth
resolved_base: Final = (
XAIOAuthAuthenticator().get_api_base()
if should_use_xai_oauth(litellm_params) and not XAIModelInfo.get_api_key(api_key)
else (api_base or get_secret_str("XAI_API_BASE") or get_secret_str("XAI_OAUTH_API_BASE") or XAI_API_BASE)
)
base: Final = (resolved_base or XAI_API_BASE).rstrip("/")
if base.endswith("/v1"):
return f"{base}/images/generations"
return f"{base}/v1/images/generations"
def validate_environment(
self,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
model: str,
messages: list[AllMessageValues], # mutable-ok: provider JSON body and base-class dict signature
optional_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
api_key: str | None = None,
api_base: str | None = None,
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
from litellm.llms.xai.oauth import (
XAIOAuthAuthenticator,
XAIOAuthError,
should_use_xai_oauth,
)
dynamic_api_key: Final = XAIModelInfo.get_api_key(api_key)
if should_use_xai_oauth(litellm_params) and not dynamic_api_key:
try:
headers["Authorization"] = f"Bearer {XAIOAuthAuthenticator().get_access_token()}"
except XAIOAuthError as exc:
raise AuthenticationError(
model=model,
llm_provider="xai",
message=str(exc),
) from exc
else:
if not dynamic_api_key:
raise AuthenticationError(
model=model,
llm_provider="xai",
message=(
"Missing xAI credentials for image generation. "
"Pass api_key / XAI_API_KEY, or set use_xai_oauth=True."
),
)
headers["Authorization"] = f"Bearer {dynamic_api_key}"
if "content-type" not in headers and "Content-Type" not in headers:
headers["Content-Type"] = "application/json"
return headers
def transform_image_generation_request(
self,
model: str,
prompt: str,
optional_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
n: Final = optional_params.get("n")
requested_ratio = optional_params.get("aspect_ratio")
aspect = {"aspect_ratio": requested_ratio} if requested_ratio is not None else None # fmt: skip # mutable-ok: provider JSON body
count = {"n": int(n)} if n is not None else None # mutable-ok: provider JSON body
return { # mutable-ok: provider JSON body and base-class dict signature
"model": XAIModelInfo.get_base_model(model) or model,
"prompt": prompt,
**(aspect or {}), # mutable-ok: provider JSON body and base-class dict signature
**(count or {}), # mutable-ok: provider JSON body and base-class dict signature
}
def transform_image_generation_response(
self,
model: str,
raw_response: httpx.Response,
model_response: ImageResponse,
logging_obj: "LiteLLMLoggingObj",
request_data: dict, # mutable-ok: provider JSON body and base-class dict signature
optional_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
encoding: "tiktoken.Encoding | None",
api_key: str | None = None,
json_mode: bool | None = None,
) -> ImageResponse:
try:
response_data: Final = raw_response.json()
except Exception:
raise self.get_error_class(
error_message=raw_response.text,
status_code=raw_response.status_code,
headers=raw_response.headers,
)
logging_obj.post_call(
input=request_data.get("prompt", ""),
api_key=api_key,
additional_args={"complete_input_dict": request_data}, # mutable-ok: provider JSON body
original_response=response_data,
)
images: Final = tuple(
ImageObject(
url=item.get("url"),
b64_json=item.get("b64_json") or item.get("b64"),
)
for item in response_data.get("data") or ()
if isinstance(item, dict)
)
if not images:
raise self.get_error_class(
error_message=f"xAI image generation returned no image data: {response_data}",
status_code=raw_response.status_code,
headers=raw_response.headers,
)
model_response.data = list(images) # mutable-ok: provider JSON body and base-class dict signature
return model_response

3
litellm/llms/xai/videos/__init__.py Normal file
View file

@ -0,0 +1,3 @@
from .transformation import XAIVideoConfig
__all__ = ["XAIVideoConfig"] # mutable-ok: provider JSON body and base-class dict signature

420
litellm/llms/xai/videos/transformation.py Normal file
View file

@ -0,0 +1,420 @@
import time
from typing import TYPE_CHECKING, Final
import httpx
from httpx._types import RequestFiles
import litellm
from litellm.constants import XAI_API_BASE
from litellm.exceptions import AuthenticationError
from litellm.litellm_core_utils.url_utils import async_safe_get, encode_url_path_segment, safe_get
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.llms.xai.common_utils import XAIModelInfo
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
_DROPPED = frozenset(("seconds", "size", "input_reference", "user", "extra_headers", "model"))
_SIZE_TO_ASPECT_RATIO: Final = { # mutable-ok: provider JSON body and base-class dict signature
"1024x1024": "1:1",
"1792x1024": "16:9",
"1024x1792": "9:16",
"1280x720": "16:9",
"720x1280": "9:16",
"1920x1080": "16:9",
"1080x1920": "9:16",
}
def _duration_from_seconds(seconds: object) -> int:
try:
return int(seconds) if seconds is not None else 6
except (TypeError, ValueError):
return 6
_STATUS_MAP: Final = { # mutable-ok: provider JSON body and base-class dict signature
"done": "completed",
"completed": "completed",
"succeeded": "completed",
"failed": "failed",
"expired": "failed",
"pending": "processing",
"processing": "processing",
"in_progress": "processing",
}
class XAIVideoConfig(BaseVideoConfig):
def get_supported_openai_params(
self, model: str
) -> list: # mutable-ok: provider JSON body and base-class dict signature
return [ # mutable-ok: provider JSON body and base-class dict signature
"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: # mutable-ok: provider JSON body and base-class dict signature
raw = video_create_optional_params
incoming: Final = dict(raw) # mutable-ok: provider JSON body and base-class dict signature
size: Final = incoming.get("size")
seconds = incoming.get("seconds")
use_duration = "seconds" in incoming and "duration" not in incoming
duration_value = _duration_from_seconds(seconds)
duration = {"duration": duration_value} if use_duration else None # mutable-ok: provider JSON body
mapped_ratio = incoming.get("aspect_ratio") or _SIZE_TO_ASPECT_RATIO.get(str(size), "16:9")
use_ratio = bool(size) and "aspect_ratio" not in incoming
ratio = {"aspect_ratio": mapped_ratio} if use_ratio else None # mutable-ok: provider JSON body
image_ref = incoming.get("image") or incoming.get("input_reference")
use_image = bool(incoming.get("input_reference")) and "image" not in incoming
image = {"image": image_ref} if use_image else None # mutable-ok: provider JSON body
return { # mutable-ok: provider JSON body and base-class dict signature
**{ # mutable-ok: provider JSON body and base-class dict signature
key: value for key, value in incoming.items() if key not in _DROPPED
},
**(duration or {}), # mutable-ok: provider JSON body and base-class dict signature
**(ratio or {}), # mutable-ok: provider JSON body and base-class dict signature
**(image or {}), # mutable-ok: provider JSON body and base-class dict signature
}
def _resolve_api_base(
self,
api_base: str | None,
api_key: str | None,
litellm_params: GenericLiteLLMParams
| dict
| None, # mutable-ok: provider JSON body and base-class dict signature
) -> str:
from litellm.llms.xai.oauth import XAIOAuthAuthenticator, should_use_xai_oauth
params: Final = (
litellm_params.model_dump()
if isinstance(litellm_params, GenericLiteLLMParams)
else (litellm_params or {}) # mutable-ok: provider JSON body and base-class dict signature
)
if should_use_xai_oauth(params) and not XAIModelInfo.get_api_key(api_key):
return XAIOAuthAuthenticator().get_api_base().rstrip("/")
resolved: Final = (
api_base
or (params.get("api_base") if isinstance(params, dict) else None)
or get_secret_str("XAI_API_BASE")
or get_secret_str("XAI_OAUTH_API_BASE")
or XAI_API_BASE
)
return str(resolved).rstrip("/")
def _v1_root(self, api_base: str) -> str:
base: Final = api_base.rstrip("/")
if base.endswith("/v1"):
return base
return f"{base}/v1"
def validate_environment(
self,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
model: str,
api_key: str | None = None,
litellm_params: GenericLiteLLMParams | None = None,
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
from litellm.llms.xai.oauth import (
XAIOAuthAuthenticator,
XAIOAuthError,
should_use_xai_oauth,
)
dumped = litellm_params.model_dump() if litellm_params is not None else None
params: Final = dumped or {} # mutable-ok: provider JSON body and base-class dict signature
resolved_api_key: Final = api_key or (litellm_params.api_key if litellm_params else None)
dynamic_api_key: Final = XAIModelInfo.get_api_key(resolved_api_key)
if should_use_xai_oauth(params) and not dynamic_api_key:
try:
headers["Authorization"] = f"Bearer {XAIOAuthAuthenticator().get_access_token()}"
except XAIOAuthError as exc:
raise AuthenticationError(
model=model or "xai-video",
llm_provider="xai",
message=str(exc),
) from exc
else:
if not dynamic_api_key:
raise AuthenticationError(
model=model or "xai-video",
llm_provider="xai",
message=(
"Missing xAI credentials for video generation. "
"Pass api_key / XAI_API_KEY, or set use_xai_oauth=True."
),
)
headers["Authorization"] = f"Bearer {dynamic_api_key}"
if "content-type" not in headers and "Content-Type" not in headers:
headers["Content-Type"] = "application/json"
return headers
def get_complete_url(
self,
model: str,
api_base: str | None,
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> str:
resolved: Final = self._resolve_api_base(
api_base=api_base,
api_key=litellm_params.get("api_key") if litellm_params else None,
litellm_params=litellm_params,
)
if not model:
return self._v1_root(resolved)
return f"{self._v1_root(resolved)}/videos/generations"
def transform_video_create_request(
self,
model: str,
prompt: str,
api_base: str,
video_create_optional_request_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[dict, RequestFiles, str]: # mutable-ok: provider JSON body and base-class dict signature
copied: Final = { # mutable-ok: provider JSON body and base-class dict signature
key: video_create_optional_request_params[key]
for key in (
"image",
"images",
"duration",
"resolution_name",
"aspect_ratio",
"size",
)
if video_create_optional_request_params.get(key) is not None
}
prompt_body = {"prompt": prompt} if prompt else None # mutable-ok: provider JSON body
duration_body = {"duration": 6} if "duration" not in copied else None # mutable-ok: provider JSON body
return (
{ # mutable-ok: provider JSON body and base-class dict signature
"model": XAIModelInfo.get_base_model(model) or model,
**(prompt_body or {}), # mutable-ok: provider JSON body and base-class dict signature
**copied,
**(duration_body or {}), # mutable-ok: provider JSON body and base-class dict signature
},
[], # mutable-ok: provider JSON body and base-class dict signature
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, # mutable-ok: provider JSON body and base-class dict signature
) -> VideoObject:
response_data: Final = raw_response.json()
request_id: Final = response_data.get("request_id") or response_data.get("id")
if not request_id:
raise ValueError(f"xAI video generation response missing request_id: {response_data}")
usage: Final = response_data.get("usage") or {} # mutable-ok: provider JSON body and base-class dict signature
video_obj: Final = VideoObject(
id=str(request_id),
object="video",
status="processing",
created_at=int(time.time()),
model=XAIModelInfo.get_base_model(model) or model,
progress=0,
)
if custom_llm_provider:
video_obj.id = encode_video_id_with_provider(video_obj.id, custom_llm_provider, model)
usage_body = usage if isinstance(usage, dict) else None
video_obj.usage = usage_body or {} # mutable-ok: provider JSON body and base-class dict signature
video_obj._hidden_params["video_url"] = None
return video_obj
def _video_resource_url(self, api_base: str, video_id: str) -> str:
encoded_video_id: Final = encode_url_path_segment(
extract_original_video_id(video_id),
field_name="video_id",
)
return f"{self._v1_root(api_base)}/videos/{encoded_video_id}"
def transform_video_status_retrieve_request(
self,
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[str, dict]: # mutable-ok: provider JSON body and base-class dict signature
return self._video_resource_url(
api_base, video_id
), {} # mutable-ok: provider JSON body and base-class dict signature
def transform_video_status_retrieve_response(
self,
raw_response: httpx.Response,
logging_obj: "LiteLLMLoggingObj",
custom_llm_provider: str | None = None,
) -> VideoObject:
response_data: Final = raw_response.json()
status_raw: Final = str(response_data.get("status") or "processing").lower()
status: Final = _STATUS_MAP.get(status_raw, status_raw)
video_body = response_data.get("video")
video_meta: Final = video_body or {} # mutable-ok: provider JSON body and base-class dict signature
video_url: Final = video_meta.get("url") if isinstance(video_meta, dict) else None
seconds: Final = (
str(video_meta.get("duration"))
if isinstance(video_meta, dict) and video_meta.get("duration") is not None
else None
)
request_id: Final = (
response_data.get("request_id")
or response_data.get("id")
or (video_url.split("/")[-1].replace(".mp4", "") if video_url else "unknown")
)
video_obj: Final = VideoObject(
id=str(request_id),
object="video",
status=status,
created_at=response_data.get("created_at") or int(time.time()),
completed_at=int(time.time()) if status == "completed" else None,
model=response_data.get("model"),
progress=response_data.get("progress"),
seconds=seconds,
usage=response_data.get("usage")
if isinstance(response_data.get("usage"), dict)
else {}, # mutable-ok: provider JSON body and base-class dict signature
)
video_obj._hidden_params["video_url"] = video_url
if custom_llm_provider and video_obj.id and video_obj.id != "unknown":
video_obj.id = encode_video_id_with_provider(video_obj.id, custom_llm_provider, response_data.get("model"))
return video_obj
def transform_video_content_request(
self,
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
variant: str | None = None,
) -> tuple[str, dict]: # mutable-ok: provider JSON body and base-class dict signature
return self._video_resource_url(
api_base, video_id
), {} # mutable-ok: provider JSON body and base-class dict signature
def _video_cdn_url(self, raw_response: httpx.Response) -> str | None:
content_type: Final = (raw_response.headers.get("content-type") or "").lower()
if "application/json" not in content_type and raw_response.content[:1] != b"{":
return None
payload: Final = raw_response.json()
if not isinstance(payload, dict):
return None
video_meta: Final = payload.get("video") or {} # mutable-ok: provider JSON body and base-class dict signature
url: Final = video_meta.get("url") if isinstance(video_meta, dict) else None
if isinstance(url, str) and url:
return url
raise ValueError(f"xAI video not ready for download (status={payload.get('status')}): {payload}")
def transform_video_content_response(
self,
raw_response: httpx.Response,
logging_obj: "LiteLLMLoggingObj",
) -> bytes:
url: Final = self._video_cdn_url(raw_response)
if url is None:
return raw_response.content
httpx_client: Final[HTTPHandler] = _get_httpx_client()
video_response: Final = safe_get(httpx_client, url)
video_response.raise_for_status()
return video_response.content
async def async_transform_video_content_response(
self,
raw_response: httpx.Response,
logging_obj: "LiteLLMLoggingObj",
) -> bytes:
url: Final = self._video_cdn_url(raw_response)
if url is None:
return raw_response.content
async_httpx_client: Final[AsyncHTTPHandler] = get_async_httpx_client(
llm_provider=litellm.LlmProviders.XAI,
)
video_response: Final = await async_safe_get(async_httpx_client, url)
video_response.raise_for_status()
return video_response.content
def transform_video_remix_request(
self,
video_id: str,
prompt: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
extra_body: dict[str, object] | None = None, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[str, dict]: # mutable-ok: provider JSON body and base-class dict signature
raise NotImplementedError("Video remix is not supported by xAI Imagine API")
def transform_video_remix_response(
self,
raw_response: httpx.Response,
logging_obj: "LiteLLMLoggingObj",
custom_llm_provider: str | None = None,
) -> VideoObject:
raise NotImplementedError("Video remix is not supported by xAI Imagine API")
def transform_video_list_request(
self,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
after: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_query: dict[str, object] | None = None, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[str, dict]: # mutable-ok: provider JSON body and base-class dict signature
raise NotImplementedError("Video listing is not supported by xAI Imagine API")
def transform_video_list_response(
self,
raw_response: httpx.Response,
logging_obj: "LiteLLMLoggingObj",
custom_llm_provider: str | None = None,
) -> dict[str, str]: # mutable-ok: provider JSON body and base-class dict signature
raise NotImplementedError("Video listing is not supported by xAI Imagine API")
def transform_video_delete_request(
self,
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[str, dict]: # mutable-ok: provider JSON body and base-class dict signature
raise NotImplementedError("Video delete is not supported by xAI Imagine API")
def transform_video_delete_response(
self,
raw_response: httpx.Response,
logging_obj: "LiteLLMLoggingObj",
) -> VideoObject:
raise NotImplementedError("Video delete is not supported by xAI Imagine API")

36
litellm/model_prices_and_context_window_backup.json
View file

@ -63368,7 +63368,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63384,7 +63385,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63401,7 +63403,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63418,7 +63421,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63435,7 +63439,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63451,7 +63456,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63467,7 +63473,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63485,6 +63492,9 @@
"output_cost_per_second_480p": 0.05,
"output_cost_per_second_720p": 0.07,
"source": "https://docs.x.ai/docs/models/grok-imagine-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image",
@ -63503,6 +63513,9 @@
"output_cost_per_second_480p": 0.08,
"output_cost_per_second_720p": 0.14,
"source": "https://docs.x.ai/docs/models/grok-imagine-video-1.5",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image",
@ -63521,6 +63534,9 @@
"output_cost_per_second_480p": 0.08,
"output_cost_per_second_720p": 0.14,
"source": "https://docs.x.ai/docs/models/grok-imagine-video-1.5",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image",
@ -63539,6 +63555,9 @@
"output_cost_per_second_480p": 0.08,
"output_cost_per_second_720p": 0.14,
"source": "https://docs.x.ai/docs/models/grok-imagine-video-1.5",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image",
@ -63582,7 +63601,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",

6
litellm/provider_endpoints_support_backup.json
View file

@ -2360,7 +2360,8 @@
"messages": true,
"responses": true,
"embeddings": false,
"image_generations": false,
"image_generations": true,
"image_edits": true,
"audio_transcriptions": false,
"audio_speech": false,
"moderations": false,
@ -2368,7 +2369,8 @@
"rerank": false,
"a2a": true,
"interactions": true,
"realtime": true
"realtime": true,
"video_generations": true
}
},
"xinference": {

1
litellm/proxy/auth/auth_utils.py
View file

@ -1691,6 +1691,7 @@ _MODEL_ROUTING_HEADER_OR_QUERY_ROUTE_MARKERS: Final = (
"/batches",
"/skills",
"/evals",
"/videos",
)
_MODEL_ROUTING_QUERY_TARGET_MODEL_ROUTE_MARKERS: Final = (
"/files",

104
litellm/proxy/image_endpoints/endpoints.py
View file

@ -1,11 +1,13 @@
import asyncio
import io
from collections.abc import Sequence
from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import Final, get_type_hints
import orjson
from fastapi import APIRouter, Depends, File, HTTPException, Request, Response, UploadFile, status
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from fastapi.responses import ORJSONResponse
from starlette.datastructures import FormData, UploadFile
import litellm
from litellm.litellm_core_utils.prompt_templates.common_utils import (
@ -19,6 +21,7 @@ from litellm.proxy.common_request_processing import (
resolve_litellm_call_id,
)
from litellm.proxy.common_utils.http_parsing_utils import (
_is_form_content_type,
coerce_numeric_form_fields,
numeric_form_fields,
)
@ -39,6 +42,8 @@ IMAGE_EDIT_NUMERIC_FORM_FIELDS: Final = numeric_form_fields(get_type_hints(Image
IMAGE_ARRAY_FIELD: Final = "image[]"
MASK_ARRAY_FIELD: Final = "mask[]"
BRACKETED_FILE_FIELDS: Final = frozenset({IMAGE_ARRAY_FIELD, MASK_ARRAY_FIELD})
IMAGE_EDIT_FILE_FIELDS: Final = MappingProxyType({"image": IMAGE_ARRAY_FIELD, "mask": MASK_ARRAY_FIELD})
IMAGE_REFERENCE_PREFIXES: Final = ("http://", "https://", "data:image/")
async def uploadfile_to_bytesio(upload: UploadFile) -> io.BytesIO:
@ -63,6 +68,58 @@ async def batch_to_bytesio(
return [await uploadfile_to_bytesio(u) for u in uploads]
async def _image_edit_part(value: object, field: str) -> io.BytesIO | str:
if isinstance(value, UploadFile):
return await uploadfile_to_bytesio(value)
if isinstance(value, str) and value.startswith(IMAGE_REFERENCE_PREFIXES):
return value
raise HTTPException(
status_code=422,
detail=f"'{field}' must be a multipart file upload, an http(s) URL, or a data:image URI.",
)
def _image_edit_values(form: FormData | None, body: Mapping[str, object], name: str) -> tuple[object, ...]:
if form is not None:
return tuple(form.getlist(name))
raw: Final = body.get(name)
if raw is None:
return ()
return tuple(raw) if isinstance(raw, list) else (raw,)
async def _image_edit_field(
form: FormData | None, body: Mapping[str, object], field: str, alias: str
) -> list[io.BytesIO | str] | str | None:
values: Final = _image_edit_values(form, body, field)
alias_values: Final = _image_edit_values(form, body, alias)
if values and alias_values:
raise HTTPException(status_code=422, detail=f"Cannot specify both '{field}' and '{alias}'")
parts: Final = tuple([await _image_edit_part(value, field) for value in values or alias_values])
if not parts:
return None
if len(parts) == 1 and isinstance(parts[0], str):
return parts[0]
return list(parts) # mutable-ok: provider image edit handlers take a list of image parts
async def image_edit_assets(request: Request, body: Mapping[str, object]) -> Mapping[str, object]:
"""
Collect ``image`` / ``mask`` (or their ``[]`` aliases) from a multipart form or JSON body.
Each part is an uploaded file, an http(s) URL, or a ``data:image/`` URI; any other string is rejected so it can
never be treated as a filesystem path downstream.
"""
form: Final = await request.form() if _is_form_content_type(request.headers.get("content-type", "")) else None
return MappingProxyType(
{
field: parts
for field, alias in IMAGE_EDIT_FILE_FIELDS.items()
if (parts := await _image_edit_field(form, body, field, alias)) is not None
}
)
@router.post(
"/v1/images/generations",
dependencies=[Depends(user_api_key_auth)],
@ -247,10 +304,6 @@ async def image_edit_api(
request: Request,
fastapi_response: Response,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
image: list[UploadFile] | None = File(None),
image_array: list[UploadFile] | None = File(None, alias=IMAGE_ARRAY_FIELD),
mask: list[UploadFile] | None = File(None),
mask_array: list[UploadFile] | None = File(None, alias=MASK_ARRAY_FIELD),
model: str | None = None,
):
"""
@ -266,20 +319,6 @@ async def image_edit_api(
-F 'prompt=Create a studio ghibli image of this'
```
"""
if image is not None and image_array is not None:
raise HTTPException(status_code=422, detail="Cannot specify both 'image' and 'image[]'")
if mask is not None and mask_array is not None:
raise HTTPException(status_code=422, detail="Cannot specify both 'mask' and 'mask[]'")
if image is None and image_array is not None:
image = image_array
if mask is None and mask_array is not None:
mask = mask_array
# if image is None:
# raise HTTPException(status_code=422, detail="Field required: image")
# Note: Image is optional for some models (e.g., Bedrock Stability style-transfer)
# The validation will be done at the model level if image is truly required
from litellm.proxy.proxy_server import (
_read_request_body,
general_settings,
@ -298,27 +337,14 @@ async def image_edit_api(
#########################################################
# Read request body and convert UploadFiles to BytesIO
#########################################################
parsed_body: Final = coerce_numeric_form_fields(
parsed_body=await _read_request_body(request=request),
numeric_fields=IMAGE_EDIT_NUMERIC_FORM_FIELDS,
)
data: Final = {
key: value
for key, value in coerce_numeric_form_fields(
parsed_body=await _read_request_body(request=request),
numeric_fields=IMAGE_EDIT_NUMERIC_FORM_FIELDS,
).items()
if key not in BRACKETED_FILE_FIELDS
**{key: value for key, value in parsed_body.items() if key not in BRACKETED_FILE_FIELDS},
**await image_edit_assets(request, parsed_body),
}
image_files: Final = await batch_to_bytesio(image)
mask_files: Final = await batch_to_bytesio(mask)
if image_files:
data["image"] = image_files
if mask_files:
data["mask"] = mask_files
for _field in ("image", "mask"):
if _field in data and isinstance(data[_field], str):
raise HTTPException(
status_code=422,
detail=f"'{_field}' must be provided as a multipart file upload, not a string.",
)
# Ensure prompt exists in data (default to None for models that don't require it)
if "prompt" not in data:

56
litellm/proxy/video_endpoints/endpoints.py
View file

@ -17,9 +17,14 @@ from litellm.proxy.common_utils.openai_endpoint_utils import (
)
from litellm.proxy.image_endpoints.endpoints import batch_to_bytesio
from litellm.proxy.video_endpoints.utils import (
assert_video_owner,
encode_character_id_in_response,
extract_model_from_target_model_names,
get_custom_provider_from_data,
infer_video_provider_from_model,
resolve_video_request_model,
stamp_video_owner,
video_owner_from_key,
video_reference_to_id,
)
from litellm.types.videos.utils import (
@ -115,7 +120,19 @@ async def video_generation(
version=version,
)
else:
return generated
return _stamp_generated_video_owner(
generated, video_owner_from_key(user_api_key_dict.token, user_api_key_dict.api_key)
)
def _stamp_generated_video_owner(result: object, owner: str | None) -> object:
video_id: Final = getattr(result, "id", None)
if isinstance(video_id, str):
result.id = stamp_video_owner(video_id, owner) # mutable-ok: response object id is rewritten before return
return result
if isinstance(result, dict) and isinstance(result.get("id"), str):
result["id"] = stamp_video_owner(result["id"], owner) # mutable-ok: JSON response body
return result
@router.get(
@ -249,6 +266,8 @@ async def video_status(
version,
)
assert_video_owner(video_id, video_owner_from_key(user_api_key_dict.token, user_api_key_dict.api_key))
# Create data with video_id
data: Final[dict[str, object]] = {"video_id": video_id}
@ -256,23 +275,25 @@ async def video_status(
provider_from_id: Final = decoded.get("custom_llm_provider")
model_id_from_decoded: Final = decoded.get("model_id")
custom_llm_provider: Final = (
explicit_provider: Final = (
get_custom_llm_provider_from_request_headers(request=request)
or get_custom_llm_provider_from_request_query(request=request)
or await get_custom_llm_provider_from_request_body(request=request)
or provider_from_id
or "openai"
)
resolved_model: Final = resolve_video_request_model(
model_id_from_decoded=model_id_from_decoded,
query_model=request.query_params.get("model"),
llm_router=llm_router,
)
if resolved_model:
data["model"] = resolved_model
custom_llm_provider: Final = explicit_provider or infer_video_provider_from_model(resolved_model) or "openai"
if custom_llm_provider:
data["custom_llm_provider"] = custom_llm_provider
# Resolve model_name from model_id if available
# This allows the router to automatically inject litellm_params from the model config
if model_id_from_decoded and llm_router:
resolved_model: Final = llm_router.resolve_model_name_from_model_id(model_id_from_decoded)
if resolved_model:
data["model"] = resolved_model
# Process request using ProxyBaseLLMRequestProcessing
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
@ -350,6 +371,8 @@ async def video_content(
version,
)
assert_video_owner(video_id, video_owner_from_key(user_api_key_dict.token, user_api_key_dict.api_key))
# Create data with video_id
data: Final[dict[str, object]] = {"video_id": video_id}
@ -366,12 +389,13 @@ async def video_content(
if custom_llm_provider:
data["custom_llm_provider"] = custom_llm_provider
# Resolve model_name from model_id if available
# This allows the router to automatically inject litellm_params from the model config
if model_id_from_decoded and llm_router:
resolved_model: Final = llm_router.resolve_model_name_from_model_id(model_id_from_decoded)
if resolved_model:
data["model"] = resolved_model
resolved_content_model: Final = resolve_video_request_model(
model_id_from_decoded=model_id_from_decoded,
query_model=request.query_params.get("model"),
llm_router=llm_router,
)
if resolved_content_model:
data["model"] = resolved_content_model
# Process request using ProxyBaseLLMRequestProcessing
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:

84
litellm/proxy/video_endpoints/utils.py
View file

@ -1,16 +1,80 @@
from typing import Any, Final
from collections.abc import Mapping, Sequence
from typing import Final, Protocol
import orjson
from litellm.types.videos.utils import encode_character_id_with_provider
from litellm.proxy._types import ProxyException
from litellm.types.videos.utils import (
decode_video_id_with_provider,
encode_character_id_with_provider,
encode_video_id_with_provider,
)
def extract_model_from_target_model_names(target_model_names: Any) -> str | None:
if isinstance(target_model_names, str):
target_model_names = [m.strip() for m in target_model_names.split(",") if m.strip()]
elif not isinstance(target_model_names, list):
class VideoModelIdResolver(Protocol):
def resolve_model_name_from_model_id(self, model_id: str | None) -> str | None: ...
def video_owner_from_key(token: str | None, api_key: str | None) -> str | None:
return token or api_key
def assert_video_owner(video_id: str, owner: str | None) -> None:
recorded: Final = decode_video_id_with_provider(video_id).get("owner")
if recorded and recorded != owner:
raise ProxyException(
message="Video does not belong to this API key",
type="permission_error",
param="video_id",
code=403,
)
def stamp_video_owner(video_id: str, owner: str | None) -> str:
if not owner:
return video_id
decoded: Final = decode_video_id_with_provider(video_id)
provider: Final = decoded.get("custom_llm_provider")
raw_id: Final = decoded.get("video_id")
if not provider or not raw_id or decoded.get("owner"):
return video_id
return encode_video_id_with_provider(raw_id, provider, decoded.get("model_id"), owner)
def infer_video_provider_from_model(model: str | None) -> str | None:
if not isinstance(model, str) or not model:
return None
return target_model_names[0] if target_model_names else None
unprefixed: Final = model.split("/", 1)[-1]
if unprefixed.startswith("grok-imagine-video"):
return "xai"
return None
def resolve_video_request_model(
*,
model_id_from_decoded: str | None,
query_model: str | None,
llm_router: VideoModelIdResolver | None,
) -> str | None:
if model_id_from_decoded:
if llm_router is not None:
resolved: Final = llm_router.resolve_model_name_from_model_id(model_id_from_decoded)
if isinstance(resolved, str) and resolved:
return resolved
return model_id_from_decoded
if isinstance(query_model, str) and query_model:
return query_model
return None
def extract_model_from_target_model_names(target_model_names: object) -> str | None:
if isinstance(target_model_names, str):
names: Final = tuple(m.strip() for m in target_model_names.split(",") if m.strip())
return names[0] if names else None
if isinstance(target_model_names, Sequence) and not isinstance(target_model_names, (str, bytes)):
first: Final = target_model_names[0] if target_model_names else None
return first if isinstance(first, str) else None
return None
def video_reference_to_id(video_ref: object) -> str:
@ -25,9 +89,9 @@ def video_reference_to_id(video_ref: object) -> str:
return parsed_ref.get("id", "") if isinstance(parsed_ref, dict) else video_ref
def get_custom_provider_from_data(data: dict[str, Any]) -> str | None:
def get_custom_provider_from_data(data: Mapping[str, object]) -> str | None:
custom_llm_provider: Final = data.get("custom_llm_provider")
if custom_llm_provider:
if isinstance(custom_llm_provider, str) and custom_llm_provider:
return custom_llm_provider
extra_body = data.get("extra_body")
@ -47,7 +111,7 @@ def get_custom_provider_from_data(data: dict[str, Any]) -> str | None:
return None
def encode_character_id_in_response(response: Any, custom_llm_provider: str, model_id: str | None) -> Any:
def encode_character_id_in_response(response: object, custom_llm_provider: str, model_id: str | None) -> object:
if isinstance(response, dict) and response.get("id"):
response["id"] = encode_character_id_with_provider(
character_id=response["id"],

1
litellm/types/videos/main.py
View file

@ -102,6 +102,7 @@ class DecodedVideoId(TypedDict, total=False):
custom_llm_provider: str | None
model_id: str | None
video_id: str
owner: ReadOnly[str | None]
class CharacterObject(BaseModel):

10
litellm/types/videos/utils.py
View file

@ -35,7 +35,9 @@ def _add_base64_padding(value: str) -> str:
return value
def encode_video_id_with_provider(video_id: str, provider: str, model_id: str | None = None) -> str:
def encode_video_id_with_provider(
video_id: str, provider: str, model_id: str | None = None, owner: str | None = None
) -> str:
"""Encode provider and model_id into video_id using base64."""
if not provider or not video_id:
return video_id
@ -50,6 +52,8 @@ def encode_video_id_with_provider(video_id: str, provider: str, model_id: str |
# ID is not encoded (even if it starts with video_), so encode it
assembled_id = str(SpecialEnums.LITELLM_MANAGED_VIDEO_COMPLETE_STR.value).format(provider, model_id or "", video_id)
if owner:
assembled_id = f"{assembled_id};owner:{owner}"
base64_encoded_id: Final[str] = base64.b64encode(assembled_id.encode("utf-8")).decode("utf-8")
@ -89,20 +93,24 @@ def decode_video_id_with_provider(encoded_video_id: str) -> DecodedVideoId:
custom_llm_provider = None
model_id = None
decoded_video_id = encoded_video_id
owner = None
if len(parts) >= 3:
custom_llm_provider_part: Final = parts[0]
model_id_part: Final = parts[1]
video_id_part: Final = parts[2]
owner_part: Final = next((part for part in parts[3:] if part.startswith("owner:")), None)
custom_llm_provider = custom_llm_provider_part.replace("litellm:custom_llm_provider:", "")
model_id = model_id_part.replace("model_id:", "")
decoded_video_id = video_id_part.replace("video_id:", "")
owner = owner_part.removeprefix("owner:") if owner_part else None
return DecodedVideoId(
custom_llm_provider=custom_llm_provider,
model_id=model_id,
video_id=decoded_video_id,
owner=owner,
)
except Exception as e:
verbose_logger.debug("Error decoding video_id '%s': %s", encoded_video_id, e)

14
litellm/utils.py
View file

@ -9614,6 +9614,12 @@ class ProviderConfigManager:
return get_modelscope_image_generation_config(model)
elif LlmProviders.EDENAI == provider:
return litellm.EdenAIImageGenerationConfig()
elif LlmProviders.XAI == provider:
from litellm.llms.xai.image_generation import (
get_xai_image_generation_config,
)
return get_xai_image_generation_config(model)
return None
@staticmethod
@ -9645,6 +9651,10 @@ class ProviderConfigManager:
from litellm.llms.fal_ai.videos.transformation import FalAIVideoConfig
return FalAIVideoConfig()
elif LlmProviders.XAI == provider:
from litellm.llms.xai.videos.transformation import XAIVideoConfig
return XAIVideoConfig()
elif LlmProviders.HOSTED_VLLM == provider:
from litellm.llms.hosted_vllm.videos import get_hosted_vllm_video_config
@ -9783,6 +9793,10 @@ class ProviderConfigManager:
)
return get_openrouter_image_edit_config(model)
elif LlmProviders.XAI == provider:
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
return XAIImageEditConfig()
return None
@staticmethod

60
litellm/videos/main.py
View file

@ -30,6 +30,51 @@ from litellm.videos.utils import VideoGenerationRequestUtils
llm_http_handler: BaseLLMHTTPHandler = BaseLLMHTTPHandler()
def _litellm_provider_from_cost_entry(info: object) -> str | None:
if isinstance(info, dict):
catalog_provider: Final = info.get("litellm_provider")
if isinstance(catalog_provider, str) and catalog_provider:
return catalog_provider
return None
def _provider_from_prefixed_model(model: str) -> str | None:
if "/" not in model:
return None
try:
_, provider, _, _ = get_llm_provider(model=model)
except Exception:
return None
return provider
def _custom_llm_provider_from_model(model: str) -> str | None:
return (
_provider_from_prefixed_model(model)
or _litellm_provider_from_cost_entry(litellm.model_cost.get(model))
or _litellm_provider_from_cost_entry(litellm.model_cost.get(f"xai/{model}"))
or ("xai" if model.startswith("grok-imagine-video") else None)
)
def _provider_for_video_id(
video_id: str,
custom_llm_provider: str | None,
model: object | None = None,
) -> str:
if custom_llm_provider is not None:
return custom_llm_provider
decoded: Final = decode_video_id_with_provider(video_id)
from_id: Final = decoded.get("custom_llm_provider")
if from_id:
return from_id
if isinstance(model, str) and model:
from_model: Final = _custom_llm_provider_from_model(model)
if from_model:
return from_model
return "openai"
##### Video Generation #######################
@client
async def avideo_generation(
@ -317,10 +362,7 @@ def video_content(
litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None)
_is_async: Final = kwargs.pop("async_call", False) is True
# Try to decode provider from video_id if not explicitly provided
if custom_llm_provider is None:
decoded: Final = decode_video_id_with_provider(video_id)
custom_llm_provider = decoded.get("custom_llm_provider") or "openai"
custom_llm_provider = _provider_for_video_id(video_id, custom_llm_provider, kwargs.get("model"))
# get llm provider logic
litellm_params: Final = GenericLiteLLMParams(**kwargs)
@ -412,10 +454,7 @@ async def avideo_content(
loop: Final = asyncio.get_event_loop()
kwargs["async_call"] = True
# Try to decode provider from video_id if not explicitly provided
if custom_llm_provider is None:
decoded: Final = decode_video_id_with_provider(video_id)
custom_llm_provider = decoded.get("custom_llm_provider") or "openai"
custom_llm_provider = _provider_for_video_id(video_id, custom_llm_provider, kwargs.get("model"))
func: Final = partial(
video_content,
@ -1019,10 +1058,7 @@ def video_status(
response: Final = VideoObject(**mock_response)
return response
# Try to decode provider from video_id if not explicitly provided
if custom_llm_provider is None:
decoded: Final = decode_video_id_with_provider(video_id)
custom_llm_provider = decoded.get("custom_llm_provider") or "openai"
custom_llm_provider = _provider_for_video_id(video_id, custom_llm_provider, kwargs.get("model"))
# get llm provider logic
litellm_params: Final = GenericLiteLLMParams(**kwargs)

36
model_prices_and_context_window.json
View file

@ -63368,7 +63368,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63384,7 +63385,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63401,7 +63403,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63418,7 +63421,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63435,7 +63439,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63451,7 +63456,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63467,7 +63473,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",
@ -63485,6 +63492,9 @@
"output_cost_per_second_480p": 0.05,
"output_cost_per_second_720p": 0.07,
"source": "https://docs.x.ai/docs/models/grok-imagine-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image",
@ -63503,6 +63513,9 @@
"output_cost_per_second_480p": 0.08,
"output_cost_per_second_720p": 0.14,
"source": "https://docs.x.ai/docs/models/grok-imagine-video-1.5",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image",
@ -63521,6 +63534,9 @@
"output_cost_per_second_480p": 0.08,
"output_cost_per_second_720p": 0.14,
"source": "https://docs.x.ai/docs/models/grok-imagine-video-1.5",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image",
@ -63539,6 +63555,9 @@
"output_cost_per_second_480p": 0.08,
"output_cost_per_second_720p": 0.14,
"source": "https://docs.x.ai/docs/models/grok-imagine-video-1.5",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image",
@ -63582,7 +63601,8 @@
"mode": "image_generation",
"source": "https://docs.x.ai/docs/models",
"supported_endpoints": [
"/v1/images/generations"
"/v1/images/generations",
"/v1/images/edits"
],
"supported_modalities": [
"text",

6
provider_endpoints_support.json
View file

@ -2706,7 +2706,8 @@
"messages": true,
"responses": true,
"embeddings": false,
"image_generations": false,
"image_generations": true,
"image_edits": true,
"audio_transcriptions": false,
"audio_speech": false,
"moderations": false,
@ -2714,7 +2715,8 @@
"rerank": false,
"a2a": true,
"interactions": true,
"realtime": true
"realtime": true,
"video_generations": true
}
},
"xinference": {

19
tests/test_litellm/proxy/auth/test_auth_utils.py
View file

@ -1022,6 +1022,25 @@ def test_get_model_from_request_includes_fine_tuning_target_model_query():
)
def test_get_model_from_request_includes_video_query_model_for_plain_id():
result = get_model_from_request(
request_data={"video_id": "plain-xai-id"},
route="/v1/videos/{video_id}",
request_query_params={"model": "grok-imagine-video-1.5"},
)
assert result == "grok-imagine-video-1.5"
def test_get_model_from_request_includes_video_query_model_on_content_route():
result = get_model_from_request(
request_data={"video_id": "plain-xai-id"},
route="/v1/videos/{video_id}/content",
request_query_params={"model": "restricted-xai-model"},
request_headers={"x-litellm-model": "also-restricted"},
)
assert result == ["restricted-xai-model", "also-restricted"]
def test_get_model_from_request_extracts_video_id_model():
from litellm.types.videos.utils import encode_video_id_with_provider

48
tests/test_litellm/proxy/image_endpoints/test_endpoints.py
View file

@ -222,6 +222,54 @@ def test_image_edit_multipart_n_that_is_not_a_number_is_left_alone(monkeypatch):
assert captured["n"] == "two"
def test_image_edit_http_url_is_accepted(monkeypatch):
captured: Dict[str, Any] = {}
response = _image_edit_client(monkeypatch, captured).post(
"/v1/images/edits",
data={
"model": "grok-imagine-image",
"prompt": "make it night",
"image": "https://imgen.x.ai/source.jpeg",
},
)
assert response.status_code == 200
assert captured["image"] == "https://imgen.x.ai/source.jpeg"
def test_image_edit_data_uri_is_accepted(monkeypatch):
captured: Dict[str, Any] = {}
response = _image_edit_client(monkeypatch, captured).post(
"/v1/images/edits",
data={
"model": "grok-imagine-image",
"prompt": "make it red",
"image": "data:image/jpeg;base64,abc",
},
)
assert response.status_code == 200
assert captured["image"] == "data:image/jpeg;base64,abc"
def test_image_edit_plain_string_image_is_rejected(monkeypatch):
captured: Dict[str, Any] = {}
response = _image_edit_client(monkeypatch, captured).post(
"/v1/images/edits",
data={
"model": "grok-imagine-image",
"prompt": "make it red",
"image": "not-a-url-or-file",
},
)
assert response.status_code == 422
assert "multipart file" in response.json()["detail"]
@pytest.mark.asyncio
async def test_a_model_the_router_cannot_serve_answers_an_openai_typed_error(monkeypatch: pytest.MonkeyPatch):
"""A bare HTTPException carries no type or param, so the tail used to ship the

84
tests/test_litellm/proxy/video_endpoints/test_endpoints.py
View file

@ -309,6 +309,54 @@ async def test_status__header_provider_beats_decoded_id(harness):
assert data["model"] == "azure-sora"
@pytest.mark.asyncio
async def test_status__resolve_fail_keeps_decoded_model_id(harness):
encoded = encode_video_id_with_provider(
"9b444cea-aaaa-bbbb-cccc-dddddddddddd",
"xai",
"grok-imagine-video-1.5",
)
await call_status(harness, encoded)
harness.resolve_model.assert_called_once_with("grok-imagine-video-1.5")
assert harness.processor_data() == {
"video_id": encoded,
"custom_llm_provider": "xai",
"model": "grok-imagine-video-1.5",
}
@pytest.mark.asyncio
async def test_status__query_model_on_plain_id(harness):
await call_status(
harness,
"9b444cea-aaaa-bbbb-cccc-dddddddddddd",
query={"model": "grok-imagine-video-1.5"},
)
harness.resolve_model.assert_not_called()
assert harness.processor_data() == {
"video_id": "9b444cea-aaaa-bbbb-cccc-dddddddddddd",
"custom_llm_provider": "xai",
"model": "grok-imagine-video-1.5",
}
@pytest.mark.asyncio
async def test_status__query_model_grok_imagine_does_not_default_openai_before_inference(
harness,
):
await call_status(
harness,
"video_plain_xai",
query={"model": "grok-imagine-video"},
)
assert harness.processor_data()["custom_llm_provider"] == "xai"
assert harness.processor_data()["model"] == "grok-imagine-video"
# =========================================================================== #
# GET /v1/videos/{video_id}/content - video_content #
# =========================================================================== #
@ -365,6 +413,42 @@ async def test_content__model_encoded_id(harness):
}
@pytest.mark.asyncio
async def test_content__query_model_on_plain_id(harness):
harness.base_process.return_value = b"x"
await call_content(
harness,
"9b444cea-aaaa-bbbb-cccc-dddddddddddd",
query={"model": "grok-imagine-video-1.5"},
)
harness.resolve_model.assert_not_called()
assert harness.processor_data() == {
"video_id": "9b444cea-aaaa-bbbb-cccc-dddddddddddd",
"model": "grok-imagine-video-1.5",
}
@pytest.mark.asyncio
async def test_content__resolve_fail_keeps_decoded_model_id(harness):
harness.base_process.return_value = b"x"
encoded = encode_video_id_with_provider(
"9b444cea-aaaa-bbbb-cccc-dddddddddddd",
"xai",
"grok-imagine-video-1.5",
)
await call_content(harness, encoded)
harness.resolve_model.assert_called_once_with("grok-imagine-video-1.5")
assert harness.processor_data() == {
"video_id": encoded,
"custom_llm_provider": "xai",
"model": "grok-imagine-video-1.5",
}
# =========================================================================== #
# POST /v1/videos/edits - video_edit #
# =========================================================================== #

95
tests/test_litellm/proxy/video_endpoints/test_utils.py
View file

@ -13,21 +13,86 @@ is encode_character_id_with_provider, which runs for real; encoding assertions
are checked by the genuine decode round-trip.
"""
import pytest
from litellm.proxy._types import ProxyException
from litellm.proxy.video_endpoints.utils import (
assert_video_owner,
encode_character_id_in_response,
extract_model_from_target_model_names,
get_custom_provider_from_data,
infer_video_provider_from_model,
resolve_video_request_model,
stamp_video_owner,
video_reference_to_id,
)
from litellm.types.videos.utils import (
decode_character_id_with_provider,
encode_character_id_with_provider,
encode_video_id_with_provider,
)
# =========================================================================== #
# resolve_video_request_model
# =========================================================================== #
class _Resolver:
def __init__(self, mapping: dict[str, str | None]):
self.mapping = mapping
def resolve_model_name_from_model_id(self, model_id: str | None) -> str | None:
return self.mapping.get(model_id) if model_id else None
def test_resolve_video_request_model__router_hit():
assert (
resolve_video_request_model(
model_id_from_decoded="deployment-123",
query_model="ignored",
llm_router=_Resolver({"deployment-123": "azure-sora"}),
)
== "azure-sora"
)
def test_resolve_video_request_model__keeps_decoded_id_when_router_misses():
assert (
resolve_video_request_model(
model_id_from_decoded="grok-imagine-video-1.5",
query_model=None,
llm_router=_Resolver({}),
)
== "grok-imagine-video-1.5"
)
def test_resolve_video_request_model__query_model_on_plain_id():
assert (
resolve_video_request_model(
model_id_from_decoded=None,
query_model="grok-imagine-video-1.5",
llm_router=None,
)
== "grok-imagine-video-1.5"
)
@pytest.mark.parametrize(
"model,expected",
[
("grok-imagine-video", "xai"),
("grok-imagine-video-1.5", "xai"),
("xai/grok-imagine-video", "xai"),
("sora-2", None),
(None, None),
("", None),
],
)
def test_infer_video_provider_from_model(model, expected):
assert infer_video_provider_from_model(model) == expected
# =========================================================================== #
# extract_model_from_target_model_names
# =========================================================================== #
@ -103,12 +168,7 @@ def test_provider__falsy_top_level_falls_through_to_extra_body(falsy):
def test_provider__from_extra_body_dict():
assert (
get_custom_provider_from_data(
{"extra_body": {"custom_llm_provider": "bedrock"}}
)
== "bedrock"
)
assert get_custom_provider_from_data({"extra_body": {"custom_llm_provider": "bedrock"}}) == "bedrock"
def test_provider__from_extra_body_json_string():
@ -126,10 +186,7 @@ def test_provider__json_string_parsing_to_non_dict_is_none():
def test_provider__extra_body_provider_not_a_string_is_none():
assert (
get_custom_provider_from_data({"extra_body": {"custom_llm_provider": 123}})
is None
)
assert get_custom_provider_from_data({"extra_body": {"custom_llm_provider": 123}}) is None
@pytest.mark.parametrize(
@ -161,9 +218,7 @@ def test_encode__dict_with_id_mutates_in_place_and_preserves_other_keys():
assert out is response # same dict, mutated in place
assert out["object"] == "character" and out["name"] == "hero"
assert out["id"] == encode_character_id_with_provider(
"char_raw", "azure", "model-1"
)
assert out["id"] == encode_character_id_with_provider("char_raw", "azure", "model-1")
decoded = decode_character_id_with_provider(out["id"])
assert decoded["custom_llm_provider"] == "azure"
assert decoded["model_id"] == "model-1"
@ -201,6 +256,16 @@ def test_encode__object_non_str_or_empty_id_unchanged(bad_id):
assert resp.id == bad_id # untouched
def test_stamp_and_assert_video_owner_round_trip():
encoded = encode_video_id_with_provider("req_1", "xai", "grok-imagine-video")
stamped = stamp_video_owner(encoded, "key-hash")
assert stamped != encoded
assert_video_owner(stamped, "key-hash")
with pytest.raises(ProxyException):
assert_video_owner(stamped, "other-key")
assert_video_owner(encoded, "other-key")
def test_encode__object_without_id_attr_returned_unchanged():
resp = _Resp()
out = encode_character_id_in_response(resp, "azure", "model-1")

167
tests/unit/llms/xai/test_xai_image_edit.py Normal file
View file

@ -0,0 +1,167 @@
from unittest.mock import MagicMock, patch
import httpx
import pytest
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
def test_provider_config_manager_returns_xai_image_edit_config():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
config = ProviderConfigManager.get_provider_image_edit_config(
model="grok-imagine-image",
provider=LlmProviders.XAI,
)
assert isinstance(config, XAIImageEditConfig)
def test_get_complete_url_default():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
url = XAIImageEditConfig().get_complete_url(
model="grok-imagine-image",
api_base=None,
litellm_params={},
)
assert url.endswith("/v1/images/edits")
def test_uses_json_not_multipart():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
assert XAIImageEditConfig().use_multipart_form_data() is False
def test_map_size_to_aspect_ratio():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
mapped = XAIImageEditConfig().map_openai_params(
image_edit_optional_params={"size": "1024x1792", "n": 1},
model="grok-imagine-image",
drop_params=True,
)
assert mapped["aspect_ratio"] == "9:16"
assert mapped["n"] == 1
assert "size" not in mapped
def test_validate_environment_requires_credentials():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
with pytest.raises(Exception, match="Missing xAI credentials"):
XAIImageEditConfig().validate_environment(
headers={},
model="grok-imagine-image",
api_key=None,
litellm_params={},
)
def test_validate_environment_oauth_injects_bearer():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
with patch(
"litellm.llms.xai.oauth.XAIOAuthAuthenticator.get_access_token",
return_value="oauth-token",
):
headers = XAIImageEditConfig().validate_environment(
headers={},
model="grok-imagine-image",
api_key=None,
litellm_params={"use_xai_oauth": True},
)
assert headers["Authorization"] == "Bearer oauth-token"
def test_map_string_n_to_int():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
mapped = XAIImageEditConfig().map_openai_params(
image_edit_optional_params={"n": "1"},
model="grok-imagine-image",
drop_params=True,
)
assert mapped["n"] == 1
assert isinstance(mapped["n"], int)
def test_transform_string_n_to_int():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
data, _ = XAIImageEditConfig().transform_image_edit_request(
model="xai/grok-imagine-image",
prompt="make the cube red",
image="https://imgen.x.ai/source.jpeg",
image_edit_optional_request_params={"n": "1"},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert data["n"] == 1
assert isinstance(data["n"], int)
def test_transform_bytes_to_data_uri_and_response():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
config = XAIImageEditConfig()
data, files = config.transform_image_edit_request(
model="xai/grok-imagine-image",
prompt="make the cube red",
image=b"\xff\xd8\xfffakejpeg",
image_edit_optional_request_params={"aspect_ratio": "1:1"},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert files == []
assert data["model"] == "grok-imagine-image"
assert data["prompt"] == "make the cube red"
assert data["aspect_ratio"] == "1:1"
assert data["image"]["url"].startswith("data:image/jpeg;base64,")
raw = httpx.Response(
200,
json={"data": [{"url": "https://imgen.x.ai/edited.jpeg", "mime_type": "image/jpeg"}]},
)
response = config.transform_image_edit_response(
model="grok-imagine-image",
raw_response=raw,
logging_obj=MagicMock(),
)
assert response.data is not None
assert response.data[0].url == "https://imgen.x.ai/edited.jpeg"
def test_transform_http_url_passthrough():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
data, files = XAIImageEditConfig().transform_image_edit_request(
model="grok-imagine-image",
prompt="make it night",
image="https://imgen.x.ai/source.jpeg",
image_edit_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert files == []
assert data["image"] == {"url": "https://imgen.x.ai/source.jpeg"}
def test_transform_multiple_images_uses_images_array():
from litellm.llms.xai.image_edit.transformation import XAIImageEditConfig
data, _ = XAIImageEditConfig().transform_image_edit_request(
model="grok-imagine-image",
prompt="combine styles",
image=["https://imgen.x.ai/a.jpeg", "https://imgen.x.ai/b.jpeg"],
image_edit_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert "image" not in data
assert data["images"] == [
{"url": "https://imgen.x.ai/a.jpeg"},
{"url": "https://imgen.x.ai/b.jpeg"},
]

129
tests/unit/llms/xai/test_xai_image_generation.py Normal file
View file

@ -0,0 +1,129 @@
from unittest.mock import MagicMock, patch
import httpx
import pytest
from litellm.llms.xai.image_generation.transformation import XAIImageGenerationConfig
from litellm.types.utils import ImageResponse, LlmProviders
from litellm.utils import ProviderConfigManager
def test_provider_config_manager_returns_xai_image_config():
config = ProviderConfigManager.get_provider_image_generation_config(
model="grok-imagine-image",
provider=LlmProviders.XAI,
)
assert isinstance(config, XAIImageGenerationConfig)
def test_map_size_to_aspect_ratio():
mapped = XAIImageGenerationConfig().map_openai_params(
non_default_params={"size": "1024x1792"},
optional_params={},
model="grok-imagine-image",
drop_params=True,
)
assert mapped["aspect_ratio"] == "9:16"
assert "size" not in mapped
def test_get_complete_url_default():
url = XAIImageGenerationConfig().get_complete_url(
api_base=None,
api_key=None,
model="grok-imagine-image",
optional_params={},
litellm_params={},
)
assert url.endswith("/v1/images/generations")
def test_validate_environment_requires_credentials():
with pytest.raises(Exception, match="Missing xAI credentials"):
XAIImageGenerationConfig().validate_environment(
headers={},
model="grok-imagine-image",
messages=[],
optional_params={},
litellm_params={},
api_key=None,
)
def test_validate_environment_oauth_injects_bearer():
with patch(
"litellm.llms.xai.oauth.XAIOAuthAuthenticator.get_access_token",
return_value="oauth-token",
):
headers = XAIImageGenerationConfig().validate_environment(
headers={},
model="grok-imagine-image",
messages=[],
optional_params={},
litellm_params={"use_xai_oauth": True},
api_key=None,
)
assert headers["Authorization"] == "Bearer oauth-token"
def test_map_string_n_to_int():
mapped = XAIImageGenerationConfig().map_openai_params(
non_default_params={"n": "1"},
optional_params={},
model="grok-imagine-image",
drop_params=True,
)
assert mapped["n"] == 1
assert isinstance(mapped["n"], int)
def test_transform_string_n_to_int():
request = XAIImageGenerationConfig().transform_image_generation_request(
model="xai/grok-imagine-image",
prompt="a red apple",
optional_params={"n": "1"},
litellm_params={},
headers={},
)
assert request["n"] == 1
assert isinstance(request["n"], int)
def test_transform_request_and_response():
config = XAIImageGenerationConfig()
request = config.transform_image_generation_request(
model="xai/grok-imagine-image",
prompt="a red apple",
optional_params={"aspect_ratio": "1:1"},
litellm_params={},
headers={},
)
assert request == {
"model": "grok-imagine-image",
"prompt": "a red apple",
"aspect_ratio": "1:1",
}
raw = httpx.Response(
200,
json={
"data": [
{
"url": "https://imgen.x.ai/example.jpeg",
"mime_type": "image/jpeg",
}
]
},
)
response = config.transform_image_generation_response(
model="grok-imagine-image",
raw_response=raw,
model_response=ImageResponse(),
logging_obj=MagicMock(),
request_data=request,
optional_params={},
litellm_params={},
encoding=None,
)
assert response.data is not None
assert response.data[0].url == "https://imgen.x.ai/example.jpeg"

337
tests/unit/llms/xai/test_xai_video_generation.py Normal file
View file

@ -0,0 +1,337 @@
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from litellm.litellm_core_utils.url_utils import SSRFError
from litellm.llms.xai.videos.transformation import XAIVideoConfig
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
def test_provider_config_manager_returns_xai_video_config():
config = ProviderConfigManager.get_provider_video_config(
model="grok-imagine-video",
provider=LlmProviders.XAI,
)
assert isinstance(config, XAIVideoConfig)
def test_map_seconds_and_size():
mapped = XAIVideoConfig().map_openai_params(
video_create_optional_params={"seconds": "10", "size": "1280x720"},
model="grok-imagine-video",
drop_params=True,
)
assert mapped["duration"] == 10
assert mapped["aspect_ratio"] == "16:9"
def test_map_seconds_nine_is_not_clamped_to_six():
mapped = XAIVideoConfig().map_openai_params(
video_create_optional_params={"seconds": "9"},
model="grok-imagine-video-1.5",
drop_params=True,
)
assert mapped["duration"] == 9
assert "seconds" not in mapped
def test_get_complete_url_create_and_status_root():
config = XAIVideoConfig()
create_url = config.get_complete_url(
model="grok-imagine-video",
api_base="https://api.x.ai/v1",
litellm_params={},
)
assert create_url == "https://api.x.ai/v1/videos/generations"
status_root = config.get_complete_url(
model="",
api_base="https://api.x.ai/v1",
litellm_params={},
)
assert status_root == "https://api.x.ai/v1"
def test_validate_environment_oauth_injects_bearer():
with patch(
"litellm.llms.xai.oauth.XAIOAuthAuthenticator.get_access_token",
return_value="oauth-token",
):
headers = XAIVideoConfig().validate_environment(
headers={},
model="grok-imagine-video",
api_key=None,
litellm_params=GenericLiteLLMParams(use_xai_oauth=True),
)
assert headers["Authorization"] == "Bearer oauth-token"
def test_transform_create_and_status_response():
config = XAIVideoConfig()
data, files, api_base = config.transform_video_create_request(
model="xai/grok-imagine-video",
prompt="a cat walking",
api_base="https://api.x.ai/v1/videos/generations",
video_create_optional_request_params={"duration": 6},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert data["model"] == "grok-imagine-video"
assert data["prompt"] == "a cat walking"
assert data["duration"] == 6
assert files == []
created = config.transform_video_create_response(
model="grok-imagine-video",
raw_response=httpx.Response(200, json={"request_id": "req-123"}),
logging_obj=MagicMock(),
custom_llm_provider="xai",
)
assert created.status == "processing"
assert created.id
status_url, params = config.transform_video_status_retrieve_request(
video_id=created.id,
api_base="https://api.x.ai/v1",
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert status_url.endswith("/videos/req-123")
assert params == {}
status = config.transform_video_status_retrieve_response(
raw_response=httpx.Response(
200,
json={
"status": "done",
"request_id": "req-123",
"video": {"url": "https://vidgen.x.ai/x.mp4", "duration": 6},
"progress": 100,
"model": "grok-imagine-video",
},
),
logging_obj=MagicMock(),
custom_llm_provider="xai",
)
assert status.status == "completed"
assert status.seconds == "6"
assert status._hidden_params.get("video_url") == "https://vidgen.x.ai/x.mp4"
def test_validate_environment_requires_credentials():
with pytest.raises(Exception, match="Missing xAI credentials"):
XAIVideoConfig().validate_environment(
headers={},
model="grok-imagine-video",
api_key=None,
litellm_params=GenericLiteLLMParams(),
)
def test_content_request_is_get_status():
url, params = XAIVideoConfig().transform_video_content_request(
video_id="req-123",
api_base="https://api.x.ai/v1",
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert url == "https://api.x.ai/v1/videos/req-123"
assert params == {}
@pytest.mark.parametrize("video_id", ["..", ""])
@pytest.mark.parametrize(
"transform_name",
["transform_video_status_retrieve_request", "transform_video_content_request"],
)
def test_video_id_rejects_empty_and_dot_path_segments(video_id, transform_name):
transform = getattr(XAIVideoConfig(), transform_name)
with pytest.raises(ValueError, match="video_id"):
transform(
video_id=video_id,
api_base="https://api.x.ai/v1",
litellm_params=GenericLiteLLMParams(),
headers={},
)
@pytest.mark.parametrize(
"transform_name",
["transform_video_status_retrieve_request", "transform_video_content_request"],
)
def test_video_id_parent_path_is_one_encoded_segment(transform_name):
transform = getattr(XAIVideoConfig(), transform_name)
url, params = transform(
video_id="../models",
api_base="https://api.x.ai/v1",
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert url == "https://api.x.ai/v1/videos/..%2Fmodels"
assert "/videos/../" not in url
assert params == {}
def test_video_id_is_percent_encoded_as_one_segment():
url, params = XAIVideoConfig().transform_video_status_retrieve_request(
video_id="req-123?x=1#frag",
api_base="https://api.x.ai/v1",
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert url == "https://api.x.ai/v1/videos/req-123%3Fx%3D1%23frag"
assert params == {}
def test_content_response_fetches_cdn_via_shared_client():
status = httpx.Response(
200,
headers={"content-type": "application/json"},
json={"status": "done", "video": {"url": "https://vidgen.x.ai/x.mp4"}},
)
cdn = MagicMock()
video_resp = MagicMock()
video_resp.content = b"mp4-bytes"
video_resp.raise_for_status.return_value = None
with patch(
"litellm.llms.xai.videos.transformation._get_httpx_client",
return_value=cdn,
), patch(
"litellm.llms.xai.videos.transformation.safe_get",
return_value=video_resp,
) as safe_get:
body = XAIVideoConfig().transform_video_content_response(status, logging_obj=MagicMock())
assert body == b"mp4-bytes"
safe_get.assert_called_once_with(cdn, "https://vidgen.x.ai/x.mp4")
@pytest.mark.asyncio
async def test_async_content_response_does_not_use_sync_client():
status = httpx.Response(
200,
headers={"content-type": "application/json"},
json={"status": "done", "video": {"url": "https://vidgen.x.ai/x.mp4"}},
)
async_client = MagicMock()
video_resp = MagicMock()
video_resp.content = b"async-mp4"
video_resp.raise_for_status.return_value = None
with patch(
"litellm.llms.xai.videos.transformation.get_async_httpx_client",
return_value=async_client,
), patch(
"litellm.llms.xai.videos.transformation._get_httpx_client",
) as sync_client, patch(
"litellm.llms.xai.videos.transformation.async_safe_get",
new=AsyncMock(return_value=video_resp),
) as async_safe_get:
body = await XAIVideoConfig().async_transform_video_content_response(
status, logging_obj=MagicMock()
)
assert body == b"async-mp4"
sync_client.assert_not_called()
async_safe_get.assert_awaited_once_with(async_client, "https://vidgen.x.ai/x.mp4")
def test_content_response_raises_when_status_has_no_url():
status = httpx.Response(
200,
headers={"content-type": "application/json"},
json={"status": "pending"},
)
with pytest.raises(ValueError, match="not ready"):
XAIVideoConfig().transform_video_content_response(status, logging_obj=MagicMock())
def test_content_response_returns_raw_bytes_when_not_json():
raw = httpx.Response(
200,
headers={"content-type": "video/mp4"},
content=b"already-mp4",
)
assert XAIVideoConfig().transform_video_content_response(raw, logging_obj=MagicMock()) == b"already-mp4"
def test_content_response_rejects_internal_cdn_host():
status = httpx.Response(
200,
headers={"content-type": "application/json"},
json={"status": "done", "video": {"url": "http://127.0.0.1/secret.mp4"}},
)
def boom(*args, **kwargs):
raise AssertionError("unsafe CDN fetch must not run")
with patch(
"litellm.llms.xai.videos.transformation._get_httpx_client",
return_value=MagicMock(get=boom),
):
with pytest.raises((SSRFError, ValueError)):
XAIVideoConfig().transform_video_content_response(status, logging_obj=MagicMock())
def test_content_response_fetches_public_cdn_via_safe_get():
status = httpx.Response(
200,
headers={"content-type": "application/json"},
json={"status": "done", "video": {"url": "https://vidgen.x.ai/x.mp4"}},
)
video_resp = MagicMock()
video_resp.content = b"safe-mp4"
video_resp.raise_for_status.return_value = None
with patch(
"litellm.llms.xai.videos.transformation.safe_get",
return_value=video_resp,
create=True,
) as safe_get:
body = XAIVideoConfig().transform_video_content_response(status, logging_obj=MagicMock())
assert body == b"safe-mp4"
assert safe_get.call_count == 1
assert safe_get.call_args.args[1] == "https://vidgen.x.ai/x.mp4"
@pytest.mark.asyncio
async def test_async_content_response_rejects_internal_cdn_host():
status = httpx.Response(
200,
headers={"content-type": "application/json"},
json={"status": "done", "video": {"url": "http://127.0.0.1/secret.mp4"}},
)
async def boom(*args, **kwargs):
raise AssertionError("unsafe async CDN fetch must not run")
with patch(
"litellm.llms.xai.videos.transformation.get_async_httpx_client",
return_value=MagicMock(get=boom),
):
with pytest.raises((SSRFError, ValueError)):
await XAIVideoConfig().async_transform_video_content_response(
status, logging_obj=MagicMock()
)
@pytest.mark.asyncio
async def test_async_content_response_fetches_public_cdn_via_async_safe_get():
status = httpx.Response(
200,
headers={"content-type": "application/json"},
json={"status": "done", "video": {"url": "https://vidgen.x.ai/x.mp4"}},
)
video_resp = MagicMock()
video_resp.content = b"async-safe-mp4"
video_resp.raise_for_status.return_value = None
with patch(
"litellm.llms.xai.videos.transformation.async_safe_get",
new=AsyncMock(return_value=video_resp),
create=True,
) as async_safe_get:
body = await XAIVideoConfig().async_transform_video_content_response(
status, logging_obj=MagicMock()
)
assert body == b"async-safe-mp4"
async_safe_get.assert_awaited()
assert async_safe_get.call_args.args[1] == "https://vidgen.x.ai/x.mp4"

24
tests/unit/videos/test_main.py
View file

@ -178,6 +178,15 @@ def test_video_content__plain_id_defaults_to_openai(seams):
assert seams.kwargs_of("video_content_handler")["custom_llm_provider"] == "openai"
def test_video_content__plain_id_with_grok_model_uses_xai(seams):
videos_main.video_content(
video_id="9b444cea-aaaa-bbbb-cccc-dddddddddddd",
model="grok-imagine-video-1.5",
)
assert seams.kwargs_of("video_content_handler")["custom_llm_provider"] == "xai"
def test_video_remix__dispatch_and_provider_from_id(seams):
result = videos_main.video_remix(video_id=AZURE_VIDEO_ID, prompt="new colors")
@ -377,6 +386,21 @@ async def test_avideo_content__pre_decodes_provider_before_delegating():
assert sync.call_args.kwargs["custom_llm_provider"] == "azure"
@pytest.mark.asyncio
async def test_avideo_content__plain_id_with_grok_model_uses_xai():
sentinel = b"mp4-bytes"
with patch.object(
videos_main, "video_content", MagicMock(return_value=sentinel)
) as sync:
result = await videos_main.avideo_content(
video_id="9b444cea-aaaa-bbbb-cccc-dddddddddddd",
model="grok-imagine-video-1.5",
)
assert result is sentinel
assert sync.call_args.kwargs["custom_llm_provider"] == "xai"
# =========================================================================== #
# Credential passthrough - DB/YAML model-config credentials the router injects
# via kwargs must reach the provider call for EVERY video handler, carried in