scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-09 03:18:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 108

feat(fal_ai): add video and audio generation providers

Browse source 
Add fal.ai support for video (Kling v2.5/v3, ByteDance Seedance v2, Veo 3.1)
and audio (ElevenLabs, MiniMax, Stable Audio, Kokoro, Orpheus, and more for
TTS / music / SFX), built on fal.ai's async queue API.

- videos: FalAIVideoConfig handles submit -> poll status -> fetch result,
  with size->aspect-ratio mapping and resilient status polling (non-2xx and
  non-JSON responses raise/tolerate appropriately, queue URLs preserved).
- audio: FalAIAudioConfig implements the submit/poll/download cycle for
  text-to-speech, music, and sound-effects endpoints via speech/aspeech.
- wire fal_ai into main.speech dispatch and the shared HTTP handler.
- add FAL_AI_DEFAULT_API_BASE and FAL_AI_POLLING_TIMEOUT constants.
- register 26 fal.ai models in model_prices_and_context_window.json.
- add unit tests for video and audio request/response transformations.
This commit is contained in:
Nick L. 2026-06-22 14:32:29 -07:00
parent 8bc18388e3
commit b6574c27ac
22 changed files with 2337 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
litellm/constants.py
View file

@ -228,6 +228,13 @@ RUNWAYML_POLLING_TIMEOUT = int(
os.getenv("RUNWAYML_POLLING_TIMEOUT", 600)
) # 10 minutes default for image generation
FAL_AI_DEFAULT_API_BASE = str(
os.getenv("FAL_AI_DEFAULT_API_BASE", "https://queue.fal.run")
)
FAL_AI_POLLING_TIMEOUT = int(
os.getenv("FAL_AI_POLLING_TIMEOUT", 900)
) # 15 minutes default for video generation
########## Networking constants ##############################################################
_DEFAULT_TTL_FOR_HTTPX_CLIENTS = 3600 # 1 hour, re-use the same httpx client for 1 hour

17
litellm/cost_calculator.py
View file

@ -491,6 +491,23 @@ def cost_per_token(
speech_model_info = litellm.get_model_info(
model=model_without_prefix, custom_llm_provider=custom_llm_provider
)
# Flat per-request audio pricing (e.g. fal.ai music generation):
# billed a fixed amount per call, independent of length or duration.
flat_audio_cost = speech_model_info.get("output_cost_per_audio")
if flat_audio_cost is not None:
return 0.0, float(flat_audio_cost)
# Per-second audio pricing: duration is decoded post-download and stashed
# on response._hidden_params["audio_output_duration"] by the provider
# transform, mirroring the transcription "audio_transcription_duration".
output_cost_per_second = speech_model_info.get("output_cost_per_second")
if output_cost_per_second is not None:
audio_output_duration = 0.0
if response is not None:
_hidden = getattr(response, "_hidden_params", {}) or {}
audio_output_duration = (
_hidden.get("audio_output_duration", 0.0) or 0.0
)
return 0.0, output_cost_per_second * float(audio_output_duration)
cost_metric = select_cost_metric_for_model(speech_model_info)
prompt_cost: float = 0.0
completion_cost: float = 0.0

9
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -6808,6 +6808,8 @@ class BaseLLMHTTPHandler:
params=data,
)
response.raise_for_status()
# Transform the response using the provider config
return video_content_provider_config.transform_video_content_response(
raw_response=response,
@ -6886,6 +6888,8 @@ class BaseLLMHTTPHandler:
params=data,
)
response.raise_for_status()
# Transform the response using the provider config
return await video_content_provider_config.async_transform_video_content_response(
raw_response=response,
@ -8052,6 +8056,8 @@ class BaseLLMHTTPHandler:
headers=headers,
)
response.raise_for_status()
return (
video_status_provider_config.transform_video_status_retrieve_response(
raw_response=response,
@ -8143,6 +8149,9 @@ class BaseLLMHTTPHandler:
url=url,
headers=headers,
)
response.raise_for_status()
return (
video_status_provider_config.transform_video_status_retrieve_response(
raw_response=response,

2
litellm/llms/fal_ai/__init__.py
View file

@ -11,6 +11,7 @@ from .image_generation import (
FalAIStableDiffusionConfig,
get_fal_ai_image_generation_config,
)
from .videos import FalAIVideoConfig
__all__ = [
"cost_calculator",
@ -23,5 +24,6 @@ __all__ = [
"FalAIFluxProV11UltraConfig",
"FalAIFluxSchnellConfig",
"FalAIStableDiffusionConfig",
"FalAIVideoConfig",
"get_fal_ai_image_generation_config",
]

3
litellm/llms/fal_ai/audio/__init__.py Normal file
View file

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

282
litellm/llms/fal_ai/audio/transformation.py Normal file
View file

@ -0,0 +1,282 @@
import time
from typing import TYPE_CHECKING, Any, Coroutine, Dict, Optional, Tuple, Union
import httpx
import litellm
from litellm._logging import verbose_logger
from litellm.constants import FAL_AI_DEFAULT_API_BASE, FAL_AI_POLLING_TIMEOUT
from litellm.litellm_core_utils.audio_utils.utils import calculate_request_duration
from litellm.llms.base_llm.text_to_speech.transformation import (
BaseTextToSpeechConfig,
TextToSpeechRequestData,
)
from litellm.llms.custom_httpx.http_handler import HTTPHandler, _get_httpx_client
from litellm.llms.fal_ai.utils import normalize_fal_model_id
from litellm.secret_managers.main import get_secret_str
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
from litellm.types.llms.openai import (
HttpxBinaryResponseContent as _HttpxBinaryResponseContent,
)
LiteLLMLoggingObj = _LiteLLMLoggingObj
HttpxBinaryResponseContent = _HttpxBinaryResponseContent
else:
LiteLLMLoggingObj = Any
HttpxBinaryResponseContent = Any
_TERMINAL_OK = "COMPLETED"
_TERMINAL_FAIL = {"FAILED", "CANCELLED"}
_POLL_INTERVAL_SECS = 1.5
class FalAIAudioConfig(BaseTextToSpeechConfig):
"""
fal.ai audio (TTS / music / SFX) via its queue API: submit goes through the
shared BaseLLMHTTPHandler, then transform_text_to_speech_response polls the
queue and downloads the rendered audio.
"""
def __init__(self) -> None:
super().__init__()
self._polling_timeout_secs: float = float(FAL_AI_POLLING_TIMEOUT)
def get_supported_openai_params(self, model: str) -> list:
return [
"input",
"voice",
"response_format",
"speed",
"extra_headers",
"extra_body",
]
def map_openai_params(
self,
model: str,
optional_params: Dict,
voice: Optional[Union[str, Dict]] = None,
drop_params: bool = False,
kwargs: Dict = {},
) -> Tuple[Optional[str], Dict]:
return (voice if isinstance(voice, str) else None), dict(optional_params)
def validate_environment(
self,
headers: dict,
model: str,
api_key: Optional[str] = None,
api_base: Optional[str] = None,
) -> dict:
resolved_key = (
api_key
or litellm.api_key
or get_secret_str("FAL_AI_API_KEY")
or get_secret_str("FAL_KEY")
)
if not resolved_key:
raise ValueError(
"fal.ai API key is required. Set FAL_AI_API_KEY (or FAL_KEY) "
"environment variable or pass api_key parameter."
)
headers.update(
{
"Authorization": f"Key {resolved_key}",
"Content-Type": "application/json",
}
)
return headers
def get_complete_url(
self,
model: str,
api_base: Optional[str],
litellm_params: dict,
) -> str:
base = api_base or get_secret_str("FAL_AI_API_BASE") or FAL_AI_DEFAULT_API_BASE
model_id = normalize_fal_model_id(model)
return f"{base.rstrip('/')}/{model_id}"
def transform_text_to_speech_request(
self,
model: str,
input: str,
voice: Optional[str],
optional_params: Dict,
litellm_params: Dict,
headers: dict,
) -> TextToSpeechRequestData:
body: Dict[str, Any] = {"text": input, "prompt": input}
if voice is not None:
body["voice"] = voice
for key, value in optional_params.items():
if key in ("response_format", "speed", "extra_headers", "extra_body"):
continue
body[key] = value
extra_body = optional_params.get("extra_body")
if isinstance(extra_body, dict):
body.update(extra_body)
return TextToSpeechRequestData(dict_body=body, headers={})
def dispatch_text_to_speech(
self,
model: str,
input: str,
voice: Optional[Union[str, Dict]],
optional_params: Dict,
litellm_params_dict: Dict,
logging_obj: "LiteLLMLoggingObj",
timeout: Union[float, httpx.Timeout],
extra_headers: Optional[Dict[str, Any]],
base_llm_http_handler: Any,
aspeech: bool,
api_base: Optional[str],
api_key: Optional[str],
**kwargs: Any,
) -> Union[
"HttpxBinaryResponseContent",
Coroutine[Any, Any, "HttpxBinaryResponseContent"],
]:
api_base = (
api_base
or litellm_params_dict.get("api_base")
or litellm.api_base
or get_secret_str("FAL_AI_API_BASE")
or FAL_AI_DEFAULT_API_BASE
)
api_key = (
api_key
or litellm_params_dict.get("api_key")
or litellm.api_key
or get_secret_str("FAL_AI_API_KEY")
or get_secret_str("FAL_KEY")
)
litellm_params_dict.update({"api_key": api_key, "api_base": api_base})
self._polling_timeout_secs = self._resolve_polling_timeout(timeout)
merged_params = dict(optional_params)
if "extra_body" not in merged_params and kwargs.get("extra_body") is not None:
merged_params["extra_body"] = kwargs["extra_body"]
voice_param = voice if isinstance(voice, str) else None
return base_llm_http_handler.text_to_speech_handler(
model=model,
input=input,
voice=voice_param,
text_to_speech_provider_config=self,
text_to_speech_optional_params=merged_params,
custom_llm_provider="fal_ai",
litellm_params=litellm_params_dict,
logging_obj=logging_obj,
timeout=timeout,
extra_headers=extra_headers,
client=None,
_is_async=aspeech,
)
def transform_text_to_speech_response(
self,
model: str,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> "HttpxBinaryResponseContent":
from litellm.types.llms.openai import HttpxBinaryResponseContent
submit_payload = raw_response.json()
status_url, response_url = self._queue_urls(submit_payload)
headers = self._poll_headers(raw_response)
client = _get_httpx_client()
verbose_logger.debug(
"fal.ai audio polling: rid=%s", submit_payload.get("request_id")
)
self._poll_until_complete_sync(
status_url=status_url,
headers=headers,
client=client,
timeout_secs=self._polling_timeout_secs,
)
result_resp = client.get(url=response_url, headers=headers)
result_resp.raise_for_status()
audio_url = self._extract_audio_url(result_resp.json())
binary_resp = client.get(url=audio_url)
binary_resp.raise_for_status()
result = HttpxBinaryResponseContent(response=binary_resp)
duration = calculate_request_duration(binary_resp.content)
if duration is not None:
result._hidden_params = {"audio_output_duration": duration}
return result
@staticmethod
def _resolve_polling_timeout(timeout: Union[float, httpx.Timeout]) -> float:
candidate: Any = timeout
if isinstance(timeout, httpx.Timeout):
candidate = timeout.read or timeout.connect
try:
value = float(candidate)
except (TypeError, ValueError):
return float(FAL_AI_POLLING_TIMEOUT)
return value if value > 0 else float(FAL_AI_POLLING_TIMEOUT)
@staticmethod
def _queue_urls(submit_payload: Dict[str, Any]) -> Tuple[str, str]:
status_url = submit_payload.get("status_url")
response_url = submit_payload.get("response_url")
if not status_url or not response_url:
raise ValueError(
"fal.ai queue submit response missing status_url/response_url"
)
return status_url, response_url
@staticmethod
def _poll_headers(raw_response: httpx.Response) -> Dict[str, str]:
authorization = raw_response.request.headers.get("Authorization", "")
return {"Authorization": authorization} if authorization else {}
def _poll_until_complete_sync(
self,
status_url: str,
headers: Dict[str, str],
client: HTTPHandler,
timeout_secs: float,
) -> None:
deadline = time.monotonic() + timeout_secs
while True:
if time.monotonic() > deadline:
raise TimeoutError(
f"fal.ai audio job did not complete within {timeout_secs}s"
)
resp = client.get(url=status_url, headers=headers)
resp.raise_for_status()
status = (resp.json().get("status") or "").upper()
if status == _TERMINAL_OK:
return
if status in _TERMINAL_FAIL:
raise RuntimeError(f"fal.ai audio job ended with status={status}")
time.sleep(_POLL_INTERVAL_SECS)
@staticmethod
def _extract_audio_url(result_payload: Dict[str, Any]) -> str:
error_payload = result_payload.get("error")
if error_payload:
raise ValueError(f"fal.ai audio generation failed: {error_payload}")
audio = result_payload.get("audio")
if isinstance(audio, dict) and isinstance(audio.get("url"), str):
return audio["url"]
audio_url = result_payload.get("audio_url")
if isinstance(audio_url, str):
return audio_url
audio_file = result_payload.get("audio_file")
if isinstance(audio_file, dict) and isinstance(audio_file.get("url"), str):
return audio_file["url"]
raise ValueError(
"fal.ai audio result missing audio url; got keys: "
f"{list(result_payload.keys())}"
)

8
litellm/llms/fal_ai/utils.py Normal file
View file

@ -0,0 +1,8 @@
def normalize_fal_model_id(model: str) -> str:
stripped = model
if stripped.startswith("fal_ai/"):
stripped = stripped[len("fal_ai/") :]
stripped = stripped.strip("/")
if not stripped:
raise ValueError("fal.ai model id is empty after stripping provider prefix")
return stripped

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

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

444
litellm/llms/fal_ai/videos/transformation.py Normal file
View file

@ -0,0 +1,444 @@
from json import JSONDecodeError
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
import httpx
from httpx._types import RequestFiles
import litellm
from litellm.constants import FAL_AI_DEFAULT_API_BASE
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
from litellm.llms.custom_httpx.http_handler import (
AsyncHTTPHandler,
HTTPHandler,
_get_httpx_client,
get_async_httpx_client,
)
from litellm.llms.fal_ai.utils import normalize_fal_model_id as _normalize_fal_model_id
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 (
decode_video_id_with_provider,
encode_video_id_with_provider,
extract_original_video_id,
)
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
LiteLLMLoggingObj = _LiteLLMLoggingObj
else:
LiteLLMLoggingObj = Any
_FAL_AI_STATUS_MAP = {
"IN_QUEUE": "queued",
"IN_PROGRESS": "in_progress",
"COMPLETED": "completed",
"FAILED": "failed",
"CANCELLED": "failed",
}
_SIZE_TO_ASPECT_RATIO = {
"1280x720": "16:9",
"1920x1080": "16:9",
"720x1280": "9:16",
"1080x1920": "9:16",
"1024x1024": "1:1",
"1280x1280": "1:1",
}
class FalAIVideoConfig(BaseVideoConfig):
"""
fal.ai uses a queue API: POST to /{model_id}, then poll
/{model_id}/requests/{id}/status and GET /{model_id}/requests/{id} for the
result. Video models return {"video": {"url": ...}}.
"""
def get_supported_openai_params(self, model: str) -> list:
return [
"model",
"prompt",
"input_reference",
"seconds",
"size",
"user",
"extra_headers",
"extra_body",
]
@staticmethod
def _image_url_field_for_model(model: str) -> str:
# Kling v3 image-to-video requires `start_image_url`; Seedance uses `image_url`.
normalized = model.lower()
if "kling-video/v3" in normalized:
return "start_image_url"
return "image_url"
def map_openai_params(
self,
video_create_optional_params: VideoCreateOptionalRequestParams,
model: str,
drop_params: bool,
) -> Dict:
mapped: Dict[str, Any] = {}
seconds = video_create_optional_params.get("seconds")
if seconds is not None:
mapped["duration"] = str(seconds)
size = video_create_optional_params.get("size")
if isinstance(size, str):
aspect = _SIZE_TO_ASPECT_RATIO.get(size)
if aspect is not None:
mapped["aspect_ratio"] = aspect
elif "x" in size:
mapped["aspect_ratio"] = size.replace("x", ":")
input_reference = video_create_optional_params.get("input_reference")
if isinstance(input_reference, str) and input_reference:
mapped[self._image_url_field_for_model(model)] = input_reference
supported = self.get_supported_openai_params(model)
for key, value in video_create_optional_params.items():
if key not in supported:
mapped[key] = value
extra_body = video_create_optional_params.get("extra_body")
if isinstance(extra_body, dict):
mapped.update(extra_body)
mapped.pop("extra_body", None)
return mapped
def validate_environment(
self,
headers: dict,
model: str,
api_key: Optional[str] = None,
litellm_params: Optional[GenericLiteLLMParams] = None,
) -> dict:
if litellm_params and litellm_params.api_key:
api_key = api_key or litellm_params.api_key
resolved_key = (
api_key
or litellm.api_key
or get_secret_str("FAL_AI_API_KEY")
or get_secret_str("FAL_KEY")
)
if not resolved_key:
raise ValueError(
"fal.ai API key is required. Set FAL_AI_API_KEY (or FAL_KEY) "
"environment variable or pass api_key parameter."
)
headers.update(
{
"Authorization": f"Key {resolved_key}",
"Content-Type": "application/json",
}
)
return headers
def get_complete_url(
self,
model: str,
api_base: Optional[str],
litellm_params: dict,
) -> str:
base = api_base or get_secret_str("FAL_AI_API_BASE") or FAL_AI_DEFAULT_API_BASE
return base.rstrip("/")
def transform_video_create_request(
self,
model: str,
prompt: str,
api_base: str,
video_create_optional_request_params: Dict,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[Dict, RequestFiles, str]:
model_id = _normalize_fal_model_id(model)
request_data: Dict[str, Any] = {"prompt": prompt}
request_data.update(video_create_optional_request_params)
request_data.pop("model", None)
return request_data, [], f"{api_base}/{model_id}"
def transform_video_create_response(
self,
model: str,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
custom_llm_provider: Optional[str] = None,
request_data: Optional[Dict] = None,
) -> VideoObject:
response_data = raw_response.json()
model_id = _normalize_fal_model_id(model)
video_data: Dict[str, Any] = {
"id": response_data.get("request_id", ""),
"object": "video",
"status": _FAL_AI_STATUS_MAP.get(
response_data.get("status", "IN_QUEUE").upper(), "queued"
),
"model": model,
}
if request_data:
if "duration" in request_data:
video_data["seconds"] = str(request_data["duration"])
if "aspect_ratio" in request_data:
video_data["size"] = str(request_data["aspect_ratio"]).replace(":", "x")
video_obj = VideoObject(**video_data) # type: ignore[arg-type]
if custom_llm_provider and video_obj.id:
video_obj.id = encode_video_id_with_provider(
video_obj.id,
custom_llm_provider,
model_id,
)
usage: Dict[str, Any] = {}
if video_obj.seconds:
try:
usage["duration_seconds"] = float(video_obj.seconds)
except (ValueError, TypeError):
pass
video_obj.usage = usage
return video_obj
def transform_video_status_retrieve_request(
self,
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
original_id, model_id = self._extract_request_and_model_id(video_id)
encoded = encode_url_path_segment(original_id, field_name="video_id")
return f"{api_base}/{model_id}/requests/{encoded}/status", {}
def transform_video_status_retrieve_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
custom_llm_provider: Optional[str] = None,
) -> VideoObject:
self._raise_for_status(raw_response)
try:
response_data = raw_response.json()
except (ValueError, JSONDecodeError):
return VideoObject(id="", object="video", status="in_progress")
status_raw = response_data.get("status", "IN_QUEUE")
error_payload = response_data.get("error")
status = _FAL_AI_STATUS_MAP.get(status_raw.upper(), "queued")
if error_payload:
status = "failed"
video_data: Dict[str, Any] = {
"id": response_data.get("request_id", ""),
"object": "video",
"status": status,
}
if "queue_position" in response_data:
video_data["progress"] = response_data["queue_position"]
if status == "failed":
video_data["error"] = {
"code": "failed",
"message": str(error_payload or "Video generation failed"),
}
video_obj = VideoObject(**video_data) # type: ignore[arg-type]
if custom_llm_provider and video_obj.id:
model_id = self._model_id_from_request_url(raw_response)
video_obj.id = encode_video_id_with_provider(
video_obj.id, custom_llm_provider, model_id
)
return video_obj
@staticmethod
def _model_id_from_request_url(raw_response: httpx.Response) -> Optional[str]:
request = getattr(raw_response, "request", None)
if request is None:
return None
path = request.url.path
head = path.split("/requests/", 1)[0].strip("/")
return head or None
def transform_video_content_request(
self,
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
variant: Optional[str] = None,
) -> Tuple[str, Dict]:
original_id, model_id = self._extract_request_and_model_id(video_id)
encoded = encode_url_path_segment(original_id, field_name="video_id")
return f"{api_base}/{model_id}/requests/{encoded}", {}
def transform_video_content_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> bytes:
self._raise_for_status(raw_response)
video_url = self._extract_video_url(raw_response.json())
httpx_client: HTTPHandler = _get_httpx_client()
video_response = httpx_client.get(video_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:
self._raise_for_status(raw_response)
video_url = self._extract_video_url(raw_response.json())
async_client: AsyncHTTPHandler = get_async_httpx_client(
llm_provider=litellm.LlmProviders.FAL_AI,
)
video_response = await async_client.get(video_url)
video_response.raise_for_status()
return video_response.content
@staticmethod
def _extract_video_url(response_data: Dict[str, Any]) -> str:
error_payload = response_data.get("error")
if error_payload:
raise ValueError(
f"fal.ai video generation failed: {error_payload}"
)
video = response_data.get("video")
if isinstance(video, dict):
url = video.get("url")
if isinstance(url, str) and url:
return url
top_level = response_data.get("url")
if isinstance(top_level, str) and top_level:
return top_level
raise ValueError(
"Video URL not found in fal.ai response. The job may still be processing."
)
@staticmethod
def _extract_request_and_model_id(video_id: str) -> Tuple[str, str]:
# Queue URLs are always rebuilt from api_base + model_id + request id, never
# taken from the (caller-supplied, only base64-encoded) video_id. Trusting an
# embedded URL would let a forged id redirect fal-authenticated requests to an
# arbitrary host and leak the API key.
decoded = decode_video_id_with_provider(video_id)
original_id = decoded.get("video_id") or extract_original_video_id(video_id)
model_id = decoded.get("model_id")
if not model_id:
raise ValueError(
"fal.ai video status/content lookup requires a model id encoded "
"in the video_id. Use the id returned by video creation."
)
return original_id, model_id
def transform_video_remix_request(
self,
video_id: str,
prompt: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
extra_body: Optional[Dict[str, Any]] = None,
) -> Tuple[str, Dict]:
raise NotImplementedError(
"Video remix is not supported by the fal.ai queue API"
)
def transform_video_remix_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
custom_llm_provider: Optional[str] = None,
) -> VideoObject:
raise NotImplementedError(
"Video remix is not supported by the fal.ai queue API"
)
def transform_video_list_request(
self,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
after: Optional[str] = None,
limit: Optional[int] = None,
order: Optional[str] = None,
extra_query: Optional[Dict[str, Any]] = None,
) -> Tuple[str, Dict]:
raise NotImplementedError(
"Video listing is not supported by the fal.ai queue API"
)
def transform_video_list_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
custom_llm_provider: Optional[str] = None,
) -> Dict[str, str]:
raise NotImplementedError(
"Video listing is not supported by the fal.ai queue API"
)
def transform_video_delete_request(
self,
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
# fal cancels jobs via PUT /requests/{id}/cancel, not the DELETE the shared handler issues.
raise NotImplementedError(
"Video delete/cancel is not supported by the fal.ai queue API via LiteLLM"
)
def transform_video_delete_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> VideoObject:
raise NotImplementedError(
"Video delete/cancel is not supported by the fal.ai queue API via LiteLLM"
)
def get_error_class(
self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers]
) -> BaseLLMException:
raise BaseLLMException(
status_code=status_code,
message=error_message,
headers=headers,
)
def _raise_for_status(self, raw_response: httpx.Response) -> None:
if raw_response.is_success:
return
raise self.get_error_class(
error_message=raw_response.text,
status_code=raw_response.status_code,
headers=raw_response.headers,
)

25
litellm/main.py
View file

@ -7419,6 +7419,31 @@ def speech(
api_key=api_key,
**kwargs,
)
elif custom_llm_provider == "fal_ai":
from litellm.llms.fal_ai.audio.transformation import FalAIAudioConfig
if text_to_speech_provider_config is None:
text_to_speech_provider_config = FalAIAudioConfig()
fal_ai_audio_config = cast(
FalAIAudioConfig, text_to_speech_provider_config
)
response = fal_ai_audio_config.dispatch_text_to_speech(
model=model,
input=input,
voice=voice,
optional_params=optional_params,
litellm_params_dict=litellm_params_dict,
logging_obj=logging_obj,
timeout=timeout,
extra_headers=extra_headers,
base_llm_http_handler=base_llm_http_handler,
aspeech=aspeech or False,
api_base=api_base,
api_key=api_key,
**kwargs,
)
if response is None:
raise Exception(

379
litellm/model_prices_and_context_window_backup.json
View file

@ -1810,6 +1810,35 @@
"supports_native_structured_output": true,
"supports_output_config": true
},
"jp.anthropic.claude-sonnet-4-6": {
"cache_creation_input_token_cost": 4.125e-06,
"cache_read_input_token_cost": 3.3e-07,
"input_cost_per_token": 3.3e-06,
"litellm_provider": "bedrock_converse",
"max_input_tokens": 1000000,
"max_output_tokens": 64000,
"max_tokens": 64000,
"mode": "chat",
"output_cost_per_token": 1.65e-05,
"search_context_cost_per_query": {
"search_context_size_high": 0.01,
"search_context_size_low": 0.01,
"search_context_size_medium": 0.01
},
"supports_assistant_prefill": true,
"supports_computer_use": true,
"supports_function_calling": true,
"supports_pdf_input": true,
"supports_prompt_caching": true,
"supports_reasoning": true,
"supports_response_schema": true,
"supports_max_reasoning_effort": true,
"supports_tool_choice": true,
"supports_vision": true,
"tool_use_system_prompt_tokens": 346,
"supports_native_structured_output": true,
"supports_minimal_reasoning_effort": true
},
"anthropic.claude-sonnet-4-20250514-v1:0": {
"cache_creation_input_token_cost": 3.75e-06,
"cache_read_input_token_cost": 3e-07,
@ -14436,6 +14465,356 @@
"/v1/images/generations"
]
},
"fal_ai/fal-ai/kling-video/v2.5-turbo/pro/text-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.07,
"source": "https://fal.ai/pricing",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/fal-ai/veo3.1/fast": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.1,
"source": "https://fal.ai/models/fal-ai/veo3.1/fast",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"video"
],
"metadata": {
"comment": "Listed price is for 720p/1080p video without audio"
}
},
"fal_ai/fal-ai/kling-video/v3/pro/text-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.112,
"source": "https://fal.ai/models/fal-ai/kling-video/v3/pro/text-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/fal-ai/kling-video/v3/standard/text-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.084,
"source": "https://fal.ai/models/fal-ai/kling-video/v3/standard/text-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/bytedance/seedance-2.0/text-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.3034,
"source": "https://fal.ai/models/bytedance/seedance-2.0/text-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/fal-ai/kling-video/v3/pro/image-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.112,
"source": "https://fal.ai/models/fal-ai/kling-video/v3/pro/image-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/fal-ai/kling-video/v3/standard/image-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.084,
"source": "https://fal.ai/models/fal-ai/kling-video/v3/standard/image-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/bytedance/seedance-2.0/image-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.3034,
"source": "https://fal.ai/models/bytedance/seedance-2.0/image-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/fal-ai/elevenlabs/tts/eleven-v3": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 0.0001,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/elevenlabs/tts/turbo-v2.5": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 5e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/elevenlabs/tts/multilingual-v2": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 0.0001,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/minimax/speech-2.8-hd": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 0.0001,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/minimax/speech-2.8-turbo": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 6e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/kokoro/american-english": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 2e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/orpheus-tts": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 5e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/dia-tts": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 4e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/inworld-tts": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 1e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/elevenlabs/music": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_second": 0.013333,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/lyria3/pro": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_audio": 0.08,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/minimax-music/v2.6": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_audio": 0.15,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/stable-audio-25/text-to-audio": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_audio": 0.2,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/elevenlabs/sound-effects/v2": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_second": 0.002,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/mmaudio-v2/text-to-audio": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_second": 0.001,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/stable-audio-3/medium/text-to-audio": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_audio": 0.0376,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"featherless_ai/featherless-ai/Qwerky-72B": {
"litellm_provider": "featherless_ai",
"max_input_tokens": 32768,

2
litellm/types/utils.py
View file

@ -261,6 +261,7 @@ class ModelInfoBase(ProviderSpecificModelInfo, total=False):
output_cost_per_reasoning_token: Optional[float]
output_cost_per_video_per_second: Optional[float] # only for vertex ai models
output_cost_per_audio_per_second: Optional[float] # only for vertex ai models
output_cost_per_audio: Optional[float] # flat per-request audio generation
output_cost_per_second: Optional[float] # for OpenAI Speech models
output_cost_per_second_1080p: Optional[
float
@ -3143,6 +3144,7 @@ class CustomPricingLiteLLMParams(BaseModel):
output_cost_per_reasoning_token: Optional[float] = None
output_cost_per_video_per_second: Optional[float] = None
output_cost_per_audio_per_second: Optional[float] = None
output_cost_per_audio: Optional[float] = None
search_context_cost_per_query: Optional[Dict[str, Any]] = None
citation_cost_per_token: Optional[float] = None
tiered_pricing: Optional[List[Dict[str, Any]]] = None

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

@ -36,7 +36,9 @@ def _add_base64_padding(value: str) -> str:
def encode_video_id_with_provider(
video_id: str, provider: str, model_id: Optional[str] = None
video_id: str,
provider: str,
model_id: Optional[str] = None,
) -> str:
"""Encode provider and model_id into video_id using base64."""
if not provider or not video_id:
@ -96,16 +98,15 @@ def decode_video_id_with_provider(encoded_video_id: str) -> DecodedVideoId:
model_id = None
decoded_video_id = encoded_video_id
if len(parts) >= 3:
custom_llm_provider_part = parts[0]
model_id_part = parts[1]
video_id_part = parts[2]
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:", "")
for part in parts:
if part.startswith("litellm:custom_llm_provider:"):
custom_llm_provider = part.replace(
"litellm:custom_llm_provider:", ""
)
elif part.startswith("model_id:"):
model_id = part.replace("model_id:", "")
elif part.startswith("video_id:"):
decoded_video_id = part.replace("video_id:", "")
return DecodedVideoId(
custom_llm_provider=custom_llm_provider,

9
litellm/utils.py
View file

@ -6152,6 +6152,7 @@ def _get_model_info_helper(
"output_cost_per_token_above_512k_tokens", None
),
output_cost_per_second=_model_info.get("output_cost_per_second", None),
output_cost_per_audio=_model_info.get("output_cost_per_audio", None),
output_cost_per_second_1080p=_model_info.get(
"output_cost_per_second_1080p", None
),
@ -9508,6 +9509,10 @@ class ProviderConfigManager:
from litellm.llms.runwayml.videos.transformation import RunwayMLVideoConfig
return RunwayMLVideoConfig()
elif LlmProviders.FAL_AI == provider:
from litellm.llms.fal_ai.videos.transformation import FalAIVideoConfig
return FalAIVideoConfig()
return None
@staticmethod
@ -9789,6 +9794,10 @@ class ProviderConfigManager:
)
return AWSPollyTextToSpeechConfig()
elif litellm.LlmProviders.FAL_AI == provider:
from litellm.llms.fal_ai.audio.transformation import FalAIAudioConfig
return FalAIAudioConfig()
return None
@staticmethod

350
model_prices_and_context_window.json
View file

@ -14444,6 +14444,356 @@
"/v1/images/generations"
]
},
"fal_ai/fal-ai/kling-video/v2.5-turbo/pro/text-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.07,
"source": "https://fal.ai/pricing",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/fal-ai/veo3.1/fast": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.1,
"source": "https://fal.ai/models/fal-ai/veo3.1/fast",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"video"
],
"metadata": {
"comment": "Listed price is for 720p/1080p video without audio"
}
},
"fal_ai/fal-ai/kling-video/v3/pro/text-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.112,
"source": "https://fal.ai/models/fal-ai/kling-video/v3/pro/text-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/fal-ai/kling-video/v3/standard/text-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.084,
"source": "https://fal.ai/models/fal-ai/kling-video/v3/standard/text-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/bytedance/seedance-2.0/text-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.3034,
"source": "https://fal.ai/models/bytedance/seedance-2.0/text-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/fal-ai/kling-video/v3/pro/image-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.112,
"source": "https://fal.ai/models/fal-ai/kling-video/v3/pro/image-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/fal-ai/kling-video/v3/standard/image-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.084,
"source": "https://fal.ai/models/fal-ai/kling-video/v3/standard/image-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/bytedance/seedance-2.0/image-to-video": {
"litellm_provider": "fal_ai",
"mode": "video_generation",
"output_cost_per_video_per_second": 0.3034,
"source": "https://fal.ai/models/bytedance/seedance-2.0/image-to-video",
"supported_endpoints": [
"/v1/videos"
],
"supported_modalities": [
"text",
"image"
],
"supported_output_modalities": [
"video"
]
},
"fal_ai/fal-ai/elevenlabs/tts/eleven-v3": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 0.0001,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/elevenlabs/tts/turbo-v2.5": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 5e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/elevenlabs/tts/multilingual-v2": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 0.0001,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/minimax/speech-2.8-hd": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 0.0001,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/minimax/speech-2.8-turbo": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 6e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/kokoro/american-english": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 2e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/orpheus-tts": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 5e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/dia-tts": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 4e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/inworld-tts": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"input_cost_per_character": 1e-05,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/elevenlabs/music": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_second": 0.013333,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/lyria3/pro": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_audio": 0.08,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/minimax-music/v2.6": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_audio": 0.15,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/stable-audio-25/text-to-audio": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_audio": 0.2,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/elevenlabs/sound-effects/v2": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_second": 0.002,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/mmaudio-v2/text-to-audio": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_second": 0.001,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"fal_ai/fal-ai/stable-audio-3/medium/text-to-audio": {
"litellm_provider": "fal_ai",
"mode": "audio_speech",
"output_cost_per_audio": 0.0376,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
]
},
"featherless_ai/featherless-ai/Qwerky-72B": {
"litellm_provider": "featherless_ai",
"max_input_tokens": 32768,

0
tests/test_litellm/llms/fal_ai/__init__.py Normal file
View file

0
tests/test_litellm/llms/fal_ai/audio/__init__.py Normal file
View file

63
tests/test_litellm/llms/fal_ai/audio/test_falai_audio_cost.py Normal file
View file

@ -0,0 +1,63 @@
import types
import pytest
import litellm
from litellm.cost_calculator import cost_per_token
ELEVEN_V3 = "fal_ai/fal-ai/elevenlabs/tts/eleven-v3"
ELEVEN_MUSIC = "fal_ai/fal-ai/elevenlabs/music"
LYRIA3_PRO = "fal_ai/fal-ai/lyria3/pro"
@pytest.fixture(autouse=True)
def _local_cost_map(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
yield
def test_char_priced_tts_uses_prompt_characters():
prompt_cost, completion_cost = cost_per_token(
model="fal_ai/fal-ai/elevenlabs/tts/eleven-v3",
custom_llm_provider="fal_ai",
call_type="speech",
prompt_characters=1000,
)
assert prompt_cost == pytest.approx(0.0001 * 1000)
assert completion_cost == 0.0
def test_flat_priced_music_returns_fixed_audio_cost():
prompt_cost, completion_cost = cost_per_token(
model="fal_ai/fal-ai/lyria3/pro",
custom_llm_provider="fal_ai",
call_type="speech",
)
assert prompt_cost == 0.0
assert completion_cost == pytest.approx(0.08)
def test_per_second_music_multiplies_decoded_duration():
response = types.SimpleNamespace(
_hidden_params={"audio_output_duration": 30.0}
)
prompt_cost, completion_cost = cost_per_token(
model="fal_ai/fal-ai/elevenlabs/music",
custom_llm_provider="fal_ai",
call_type="speech",
response=response,
)
assert prompt_cost == 0.0
assert completion_cost == pytest.approx(0.013333 * 30.0)
def test_per_second_music_falls_back_to_zero_without_duration():
prompt_cost, completion_cost = cost_per_token(
model="fal_ai/fal-ai/elevenlabs/music",
custom_llm_provider="fal_ai",
call_type="speech",
response=None,
)
assert prompt_cost == 0.0
assert completion_cost == 0.0

324
tests/test_litellm/llms/fal_ai/audio/test_falai_audio_transformation.py Normal file
View file

@ -0,0 +1,324 @@
from unittest.mock import MagicMock
import httpx
import pytest
from litellm.llms.fal_ai.audio.transformation import FalAIAudioConfig
from litellm.llms.fal_ai.utils import normalize_fal_model_id
ELEVEN_V3 = "fal_ai/fal-ai/elevenlabs/tts/eleven-v3"
ELEVEN_V3_ID = "fal-ai/elevenlabs/tts/eleven-v3"
FAL_API_BASE = "https://queue.fal.run"
SUBMIT_PAYLOAD = {
"request_id": "test-rid",
"status_url": f"{FAL_API_BASE}/fal-ai/elevenlabs/requests/test-rid/status",
"response_url": f"{FAL_API_BASE}/fal-ai/elevenlabs/requests/test-rid",
}
RESULT_PAYLOAD = {
"audio": {
"url": "https://v3b.fal.media/files/x/output.mp3",
"content_type": "audio/mpeg",
}
}
def _resp(json_payload=None, content=b"", status_code=200, request=None):
resp = MagicMock(spec=httpx.Response)
resp.status_code = status_code
resp.json.return_value = json_payload
resp.content = content
resp.request = request
resp.raise_for_status = MagicMock()
return resp
class TestFalAIAudioBasics:
def setup_method(self):
self.config = FalAIAudioConfig()
def test_validate_environment_uses_fal_ai_api_key(self, monkeypatch):
monkeypatch.setenv("FAL_AI_API_KEY", "key-123")
headers = self.config.validate_environment(headers={}, model=ELEVEN_V3)
assert headers["Authorization"] == "Key key-123"
assert headers["Content-Type"] == "application/json"
def test_validate_environment_falls_back_to_fal_key(self, monkeypatch):
monkeypatch.delenv("FAL_AI_API_KEY", raising=False)
monkeypatch.setenv("FAL_KEY", "fallback")
headers = self.config.validate_environment(headers={}, model=ELEVEN_V3)
assert headers["Authorization"] == "Key fallback"
def test_validate_environment_raises_when_missing(self, monkeypatch):
monkeypatch.delenv("FAL_AI_API_KEY", raising=False)
monkeypatch.delenv("FAL_KEY", raising=False)
with pytest.raises(ValueError, match="fal.ai API key is required"):
self.config.validate_environment(headers={}, model=ELEVEN_V3)
def test_get_complete_url_default(self, monkeypatch):
monkeypatch.delenv("FAL_AI_API_BASE", raising=False)
assert self.config.get_complete_url(
model=ELEVEN_V3, api_base=None, litellm_params={}
) == f"{FAL_API_BASE}/{ELEVEN_V3_ID}"
def test_get_complete_url_strips_trailing_slash(self):
assert self.config.get_complete_url(
model=ELEVEN_V3,
api_base="https://custom.example.com/",
litellm_params={},
) == f"https://custom.example.com/{ELEVEN_V3_ID}"
def test_normalize_model_id_strips_prefix(self):
assert normalize_fal_model_id(ELEVEN_V3) == ELEVEN_V3_ID
assert normalize_fal_model_id(ELEVEN_V3_ID) == ELEVEN_V3_ID
def test_normalize_model_id_rejects_empty(self):
with pytest.raises(ValueError, match="empty after stripping"):
normalize_fal_model_id("fal_ai/")
def test_transform_request_carries_text_voice_and_extras(self):
request = self.config.transform_text_to_speech_request(
model=ELEVEN_V3,
input="hello",
voice="Aria",
optional_params={
"stability": 0.5,
"extra_body": {"language_code": "en"},
"response_format": "mp3",
},
litellm_params={},
headers={},
)
body = request["dict_body"]
assert body["text"] == "hello"
assert body["prompt"] == "hello"
assert body["voice"] == "Aria"
assert body["stability"] == 0.5
assert body["language_code"] == "en"
assert "response_format" not in body
assert "extra_body" not in body
def test_extract_audio_url_supports_known_shapes(self):
assert self.config._extract_audio_url({"audio": {"url": "x"}}) == "x"
assert self.config._extract_audio_url({"audio_url": "y"}) == "y"
assert self.config._extract_audio_url({"audio_file": {"url": "z"}}) == "z"
def test_extract_audio_url_raises_when_missing(self):
with pytest.raises(ValueError, match="missing audio url"):
self.config._extract_audio_url({"other": "shape"})
def test_extract_audio_url_raises_on_error_payload(self):
with pytest.raises(ValueError, match="audio generation failed"):
self.config._extract_audio_url({"error": "boom"})
class TestFalAIAudioResponsePolling:
def setup_method(self):
self.config = FalAIAudioConfig()
def _submit_response(self):
request = httpx.Request(
"POST",
f"{FAL_API_BASE}/{ELEVEN_V3_ID}",
headers={"Authorization": "Key key-123"},
)
return _resp(json_payload=SUBMIT_PAYLOAD, request=request)
def test_response_polls_until_complete_then_downloads(self, monkeypatch):
binary_payload = b"audio-bytes-12345"
binary_resp = _resp(content=binary_payload)
result_resp = _resp(json_payload=RESULT_PAYLOAD)
in_progress = _resp(json_payload={"status": "IN_PROGRESS"})
completed = _resp(json_payload={"status": "COMPLETED"})
client = MagicMock()
client.get.side_effect = [in_progress, completed, result_resp, binary_resp]
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation._get_httpx_client",
lambda: client,
)
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation.time.sleep", lambda _s: None
)
out = self.config.transform_text_to_speech_response(
model=ELEVEN_V3,
raw_response=self._submit_response(),
logging_obj=MagicMock(),
)
assert out.response.content == binary_payload
get_urls = [c.kwargs.get("url") or c.args[0] for c in client.get.call_args_list]
assert get_urls[0] == SUBMIT_PAYLOAD["status_url"]
assert get_urls[1] == SUBMIT_PAYLOAD["status_url"]
assert get_urls[2] == SUBMIT_PAYLOAD["response_url"]
assert get_urls[3] == RESULT_PAYLOAD["audio"]["url"]
def test_response_forwards_authorization_to_poll(self, monkeypatch):
client = MagicMock()
client.get.side_effect = [
_resp(json_payload={"status": "COMPLETED"}),
_resp(json_payload=RESULT_PAYLOAD),
_resp(content=b"audio"),
]
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation._get_httpx_client",
lambda: client,
)
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation.time.sleep", lambda _s: None
)
self.config.transform_text_to_speech_response(
model=ELEVEN_V3,
raw_response=self._submit_response(),
logging_obj=MagicMock(),
)
status_call = client.get.call_args_list[0]
assert status_call.kwargs["headers"]["Authorization"] == "Key key-123"
def test_response_raises_on_failed_status(self, monkeypatch):
client = MagicMock()
client.get.side_effect = [_resp(json_payload={"status": "FAILED"})]
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation._get_httpx_client",
lambda: client,
)
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation.time.sleep", lambda _s: None
)
with pytest.raises(RuntimeError, match="status=FAILED"):
self.config.transform_text_to_speech_response(
model=ELEVEN_V3,
raw_response=self._submit_response(),
logging_obj=MagicMock(),
)
def test_response_raises_when_submit_payload_missing_urls(self):
request = httpx.Request("POST", f"{FAL_API_BASE}/{ELEVEN_V3_ID}")
bad_submit = _resp(json_payload={"request_id": "x"}, request=request)
with pytest.raises(ValueError, match="missing status_url/response_url"):
self.config.transform_text_to_speech_response(
model=ELEVEN_V3,
raw_response=bad_submit,
logging_obj=MagicMock(),
)
def test_response_stashes_decoded_duration(self, monkeypatch):
client = MagicMock()
client.get.side_effect = [
_resp(json_payload={"status": "COMPLETED"}),
_resp(json_payload=RESULT_PAYLOAD),
_resp(content=b"RIFFfake-wav-bytes"),
]
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation._get_httpx_client",
lambda: client,
)
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation.time.sleep", lambda _s: None
)
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation.calculate_request_duration",
lambda _content: 12.5,
)
out = self.config.transform_text_to_speech_response(
model=ELEVEN_V3,
raw_response=self._submit_response(),
logging_obj=MagicMock(),
)
assert out._hidden_params["audio_output_duration"] == 12.5
def test_response_omits_duration_when_undeterminable(self, monkeypatch):
client = MagicMock()
client.get.side_effect = [
_resp(json_payload={"status": "COMPLETED"}),
_resp(json_payload=RESULT_PAYLOAD),
_resp(content=b"audio"),
]
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation._get_httpx_client",
lambda: client,
)
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation.time.sleep", lambda _s: None
)
monkeypatch.setattr(
"litellm.llms.fal_ai.audio.transformation.calculate_request_duration",
lambda _content: None,
)
out = self.config.transform_text_to_speech_response(
model=ELEVEN_V3,
raw_response=self._submit_response(),
logging_obj=MagicMock(),
)
assert "audio_output_duration" not in out._hidden_params
CHAR_PRICED_MODELS = [
"fal_ai/fal-ai/elevenlabs/tts/eleven-v3",
"fal_ai/fal-ai/elevenlabs/tts/turbo-v2.5",
"fal_ai/fal-ai/elevenlabs/tts/multilingual-v2",
"fal_ai/fal-ai/minimax/speech-2.8-hd",
"fal_ai/fal-ai/minimax/speech-2.8-turbo",
"fal_ai/fal-ai/kokoro/american-english",
"fal_ai/fal-ai/orpheus-tts",
"fal_ai/fal-ai/dia-tts",
"fal_ai/fal-ai/inworld-tts",
]
PER_SECOND_MODELS = [
"fal_ai/fal-ai/elevenlabs/music",
"fal_ai/fal-ai/elevenlabs/sound-effects/v2",
"fal_ai/fal-ai/mmaudio-v2/text-to-audio",
]
FLAT_MODELS = [
"fal_ai/fal-ai/lyria3/pro",
"fal_ai/fal-ai/minimax-music/v2.6",
"fal_ai/fal-ai/stable-audio-25/text-to-audio",
"fal_ai/fal-ai/stable-audio-3/medium/text-to-audio",
]
def _backup_entry(model_id):
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
entry = GetModelCostMap.load_local_model_cost_map().get(model_id)
assert entry is not None, f"{model_id} missing from local backup model cost map"
assert entry["litellm_provider"] == "fal_ai"
assert entry["mode"] == "audio_speech"
assert "/v1/audio/speech" in entry["supported_endpoints"]
assert entry["supported_output_modalities"] == ["audio"]
return entry
@pytest.mark.parametrize("model_id", CHAR_PRICED_MODELS)
def test_tts_models_priced_per_character(model_id):
entry = _backup_entry(model_id)
assert isinstance(entry["input_cost_per_character"], (int, float))
assert "output_cost_per_second" not in entry
@pytest.mark.parametrize("model_id", PER_SECOND_MODELS)
def test_music_models_priced_per_second(model_id):
entry = _backup_entry(model_id)
assert isinstance(entry["output_cost_per_second"], (int, float))
@pytest.mark.parametrize("model_id", FLAT_MODELS)
def test_music_models_priced_flat_per_audio(model_id):
entry = _backup_entry(model_id)
assert isinstance(entry["output_cost_per_audio"], (int, float))
def test_provider_config_manager_returns_fal_ai_audio_config():
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
config = ProviderConfigManager.get_provider_text_to_speech_config(
model=ELEVEN_V3, provider=LlmProviders.FAL_AI
)
assert isinstance(config, FalAIAudioConfig)

0
tests/test_litellm/llms/fal_ai/videos/__init__.py Normal file
View file

397
tests/test_litellm/llms/fal_ai/videos/test_falai_video_transformation.py Normal file
View file

@ -0,0 +1,397 @@
from unittest.mock import Mock
import httpx
import pytest
from litellm.llms.fal_ai.videos.transformation import FalAIVideoConfig
from litellm.types.router import GenericLiteLLMParams
from litellm.types.videos.main import VideoObject
from litellm.types.videos.utils import (
decode_video_id_with_provider,
encode_video_id_with_provider,
)
SORA_2_MODEL = "fal_ai/fal-ai/sora-2/text-to-video"
KLING_MODEL = "fal_ai/fal-ai/kling-video/v2.5-turbo/pro/text-to-video"
KLING_MODEL_ID = "fal-ai/kling-video/v2.5-turbo/pro/text-to-video"
FAL_API_BASE = "https://queue.fal.run"
def _fal_status_response(payload, request_id="abc-123", status_code=200):
request = httpx.Request(
"GET", f"{FAL_API_BASE}/{KLING_MODEL_ID}/requests/{request_id}/status"
)
return httpx.Response(status_code, json=payload, request=request)
class TestFalAIVideoTransformation:
def setup_method(self):
self.config = FalAIVideoConfig()
self.mock_logging_obj = Mock()
def test_validate_environment_uses_fal_ai_api_key(self, monkeypatch):
monkeypatch.setenv("FAL_AI_API_KEY", "test-key-123")
headers = self.config.validate_environment(
headers={},
model=SORA_2_MODEL,
)
assert headers["Authorization"] == "Key test-key-123"
assert headers["Content-Type"] == "application/json"
def test_validate_environment_falls_back_to_fal_key(self, monkeypatch):
monkeypatch.delenv("FAL_AI_API_KEY", raising=False)
monkeypatch.setenv("FAL_KEY", "fallback-key")
headers = self.config.validate_environment(headers={}, model=SORA_2_MODEL)
assert headers["Authorization"] == "Key fallback-key"
def test_validate_environment_raises_when_missing(self, monkeypatch):
monkeypatch.delenv("FAL_AI_API_KEY", raising=False)
monkeypatch.delenv("FAL_KEY", raising=False)
with pytest.raises(ValueError, match="fal.ai API key is required"):
self.config.validate_environment(headers={}, model=SORA_2_MODEL)
def test_get_complete_url_uses_default_base(self, monkeypatch):
monkeypatch.delenv("FAL_AI_API_BASE", raising=False)
url = self.config.get_complete_url(
model=SORA_2_MODEL, api_base=None, litellm_params={}
)
assert url == FAL_API_BASE
def test_get_complete_url_strips_trailing_slash(self):
url = self.config.get_complete_url(
model=SORA_2_MODEL,
api_base="https://custom.example.com/",
litellm_params={},
)
assert url == "https://custom.example.com"
def test_map_openai_params_converts_seconds_and_size(self):
params = self.config.map_openai_params(
video_create_optional_params={"seconds": 5, "size": "1280x720"},
model=KLING_MODEL,
drop_params=False,
)
assert params["duration"] == "5"
assert params["aspect_ratio"] == "16:9"
def test_map_openai_params_falls_back_to_colon_replacement(self):
params = self.config.map_openai_params(
video_create_optional_params={"size": "640x480"},
model=KLING_MODEL,
drop_params=False,
)
assert params["aspect_ratio"] == "640:480"
def test_map_openai_params_unpacks_extra_body(self):
params = self.config.map_openai_params(
video_create_optional_params={
"extra_body": {"negative_prompt": "blurry", "cfg_scale": 0.5}
},
model=KLING_MODEL,
drop_params=False,
)
assert params["negative_prompt"] == "blurry"
assert params["cfg_scale"] == 0.5
assert "extra_body" not in params
def test_transform_video_create_request_builds_queue_url(self):
data, files, url = self.config.transform_video_create_request(
model=KLING_MODEL,
prompt="A demo video",
api_base=FAL_API_BASE,
video_create_optional_request_params={
"duration": "5",
"aspect_ratio": "16:9",
},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert url == f"{FAL_API_BASE}/{KLING_MODEL_ID}"
assert data["prompt"] == "A demo video"
assert data["duration"] == "5"
assert data["aspect_ratio"] == "16:9"
assert "model" not in data
assert files == []
def test_transform_video_create_response_encodes_model_into_video_id(self):
mock_response = Mock(spec=httpx.Response)
mock_response.json.return_value = {
"request_id": "abc-123",
"status": "IN_QUEUE",
}
video_obj = self.config.transform_video_create_response(
model=KLING_MODEL,
raw_response=mock_response,
logging_obj=self.mock_logging_obj,
custom_llm_provider="fal_ai",
request_data={"duration": "5", "aspect_ratio": "16:9"},
)
assert isinstance(video_obj, VideoObject)
assert video_obj.status == "queued"
assert video_obj.id.startswith("video_")
decoded = decode_video_id_with_provider(video_obj.id)
assert decoded.get("video_id") == "abc-123"
assert decoded.get("custom_llm_provider") == "fal_ai"
assert decoded.get("model_id") == KLING_MODEL_ID
assert video_obj.seconds == "5"
assert video_obj.size == "16x9"
def test_transform_video_status_retrieve_request_builds_status_url(self):
encoded_id = encode_video_id_with_provider("abc-123", "fal_ai", KLING_MODEL_ID)
url, params = self.config.transform_video_status_retrieve_request(
video_id=encoded_id,
api_base=FAL_API_BASE,
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert url == f"{FAL_API_BASE}/{KLING_MODEL_ID}/requests/abc-123/status"
assert params == {}
def test_transform_video_status_retrieve_request_reconstructs_from_model_id(self):
encoded_id = encode_video_id_with_provider("abc-123", "fal_ai", KLING_MODEL_ID)
url, params = self.config.transform_video_status_retrieve_request(
video_id=encoded_id,
api_base="https://attacker.example.com",
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert url == f"https://attacker.example.com/{KLING_MODEL_ID}/requests/abc-123/status"
assert params == {}
def test_transform_video_status_request_url_path_segment_is_encoded(self):
encoded_id = encode_video_id_with_provider(
"../../../etc/passwd", "fal_ai", KLING_MODEL_ID
)
url, _ = self.config.transform_video_status_retrieve_request(
video_id=encoded_id,
api_base=FAL_API_BASE,
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert "/requests/..%2F..%2F..%2Fetc%2Fpasswd/status" in url
def test_transform_video_status_response_maps_in_progress(self):
mock_response = _fal_status_response(
{
"request_id": "abc-123",
"status": "IN_PROGRESS",
"queue_position": 2,
}
)
status_obj = self.config.transform_video_status_retrieve_response(
raw_response=mock_response,
logging_obj=self.mock_logging_obj,
custom_llm_provider="fal_ai",
)
assert status_obj.status == "in_progress"
assert status_obj.progress == 2
def test_transform_video_status_response_maps_failed_with_error(self):
mock_response = _fal_status_response(
{
"request_id": "abc-123",
"status": "FAILED",
"error": "model timed out",
}
)
status_obj = self.config.transform_video_status_retrieve_response(
raw_response=mock_response,
logging_obj=self.mock_logging_obj,
custom_llm_provider="fal_ai",
)
assert status_obj.status == "failed"
assert status_obj.error is not None
assert status_obj.error["message"] == "model timed out"
def test_transform_video_status_response_tolerates_non_json_body(self):
mock_response = Mock(spec=httpx.Response)
mock_response.json.side_effect = ValueError(
"Expecting value: line 1 column 1 (char 0)"
)
status_obj = self.config.transform_video_status_retrieve_response(
raw_response=mock_response,
logging_obj=self.mock_logging_obj,
custom_llm_provider="fal_ai",
)
assert status_obj.status == "in_progress"
def test_transform_video_content_request_builds_result_url(self):
encoded_id = encode_video_id_with_provider("abc-123", "fal_ai", KLING_MODEL_ID)
url, params = self.config.transform_video_content_request(
video_id=encoded_id,
api_base=FAL_API_BASE,
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert url == f"{FAL_API_BASE}/{KLING_MODEL_ID}/requests/abc-123"
assert params == {}
def test_transform_video_content_request_reconstructs_from_model_id(self):
encoded_id = encode_video_id_with_provider("abc-123", "fal_ai", KLING_MODEL_ID)
url, params = self.config.transform_video_content_request(
video_id=encoded_id,
api_base="https://attacker.example.com",
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert url == f"https://attacker.example.com/{KLING_MODEL_ID}/requests/abc-123"
assert params == {}
def test_extract_video_url_handles_video_object(self):
url = self.config._extract_video_url(
{"video": {"url": "https://cdn.example.com/v.mp4"}}
)
assert url == "https://cdn.example.com/v.mp4"
def test_extract_video_url_handles_top_level_url(self):
url = self.config._extract_video_url({"url": "https://cdn.example.com/v.mp4"})
assert url == "https://cdn.example.com/v.mp4"
def test_extract_video_url_raises_when_missing(self):
with pytest.raises(ValueError, match="Video URL not found"):
self.config._extract_video_url({"status": "IN_PROGRESS"})
def test_status_request_requires_model_id_in_video_id(self):
plain_id = encode_video_id_with_provider("abc-123", "fal_ai", None)
with pytest.raises(ValueError, match="model id encoded"):
self.config.transform_video_status_retrieve_request(
video_id=plain_id,
api_base=FAL_API_BASE,
litellm_params=GenericLiteLLMParams(),
headers={},
)
def test_transform_video_delete_request_raises_not_implemented(self):
encoded_id = encode_video_id_with_provider("abc-123", "fal_ai", KLING_MODEL_ID)
with pytest.raises(NotImplementedError, match="delete/cancel is not supported"):
self.config.transform_video_delete_request(
video_id=encoded_id,
api_base=FAL_API_BASE,
litellm_params=GenericLiteLLMParams(),
headers={},
)
def test_transform_video_delete_response_raises_not_implemented(self):
mock_response = Mock(spec=httpx.Response)
with pytest.raises(NotImplementedError, match="delete/cancel is not supported"):
self.config.transform_video_delete_response(
raw_response=mock_response,
logging_obj=self.mock_logging_obj,
)
def test_remix_and_list_raise_not_implemented(self):
with pytest.raises(NotImplementedError):
self.config.transform_video_remix_request(
video_id="x",
prompt="p",
api_base=FAL_API_BASE,
litellm_params=GenericLiteLLMParams(),
headers={},
)
with pytest.raises(NotImplementedError):
self.config.transform_video_list_request(
api_base=FAL_API_BASE,
litellm_params=GenericLiteLLMParams(),
headers={},
)
def test_full_video_workflow(self):
config = FalAIVideoConfig()
mock_logging_obj = Mock()
data, _, url = config.transform_video_create_request(
model=KLING_MODEL,
prompt="A high quality demo of LiteLLM video gateway",
api_base=FAL_API_BASE,
video_create_optional_request_params={
"duration": "5",
"aspect_ratio": "16:9",
},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert url.endswith(KLING_MODEL_ID)
create_response = Mock(spec=httpx.Response)
create_response.json.return_value = {
"request_id": "queued-id-1",
"status": "IN_QUEUE",
}
video_obj = config.transform_video_create_response(
model=KLING_MODEL,
raw_response=create_response,
logging_obj=mock_logging_obj,
custom_llm_provider="fal_ai",
request_data=data,
)
assert video_obj.status == "queued"
assert video_obj.id.startswith("video_")
status_url, _ = config.transform_video_status_retrieve_request(
video_id=video_obj.id,
api_base=FAL_API_BASE,
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert status_url.endswith("/requests/queued-id-1/status")
completed_response = _fal_status_response(
{
"request_id": "queued-id-1",
"status": "COMPLETED",
},
request_id="queued-id-1",
)
completed_obj = config.transform_video_status_retrieve_response(
raw_response=completed_response,
logging_obj=mock_logging_obj,
custom_llm_provider="fal_ai",
)
assert completed_obj.status == "completed"
def test_provider_config_manager_returns_fal_ai_video_config():
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
config = ProviderConfigManager.get_provider_video_config(
model=SORA_2_MODEL, provider=LlmProviders.FAL_AI
)
assert isinstance(config, FalAIVideoConfig)
@pytest.mark.parametrize(
"model_id,expected_modalities",
[
("fal_ai/fal-ai/kling-video/v3/standard/text-to-video", ("text",)),
("fal_ai/fal-ai/kling-video/v3/pro/text-to-video", ("text",)),
("fal_ai/bytedance/seedance-2.0/text-to-video", ("text",)),
("fal_ai/fal-ai/veo3.1/fast", ("text",)),
("fal_ai/fal-ai/kling-video/v3/standard/image-to-video", ("text", "image")),
("fal_ai/fal-ai/kling-video/v3/pro/image-to-video", ("text", "image")),
("fal_ai/bytedance/seedance-2.0/image-to-video", ("text", "image")),
],
)
def test_fal_ai_video_model_registered_with_video_endpoint(
model_id: str, expected_modalities: tuple
):
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
backup = GetModelCostMap.load_local_model_cost_map()
entry = backup.get(model_id)
assert entry is not None, f"{model_id} missing from local backup model cost map"
assert entry["litellm_provider"] == "fal_ai"
assert entry["mode"] == "video_generation"
assert "/v1/videos" in entry["supported_endpoints"]
assert tuple(entry["supported_modalities"]) == expected_modalities
assert entry["supported_output_modalities"] == ["video"]
assert isinstance(entry["output_cost_per_video_per_second"], (int, float))

1
tests/test_litellm/test_utils.py
View file

@ -593,6 +593,7 @@ def validate_model_cost_values(model_data, exceptions=None):
"output_cost_per_pixel",
"input_cost_per_second",
"output_cost_per_second",
"output_cost_per_audio",
"output_cost_per_second_1080p",
"input_cost_per_query",
"input_cost_per_request",