--- title: request_adapter ---

Translate OpenAI/vLLM-Omni requests into FastVideo's typed request API.

fastvideo.entrypoints.openai.request_adapter.MediaPolicy

MediaPolicy(roots: tuple[str, ...] = (), allow_private_urls: bool = False)

fastvideo.entrypoints.openai.request_adapter.MediaPolicyError

MediaPolicyError()

A media reference is not allowed by the server's media policy.

fastvideo.entrypoints.openai.request_adapter.discard

discard(path: str) -> None

fastvideo.entrypoints.openai.request_adapter.fetch_media

fetch_media(url: str, target: str, *, max_bytes: int, policy: MediaPolicy, transport: httpx.AsyncBaseTransport | None = None) -> httpx.Headers

Download url to target under policy; returns the final response headers.

fastvideo.entrypoints.openai.request_adapter.resolve_local_media

resolve_local_media(source: str, policy: MediaPolicy) -> str

Return the real path of a caller-supplied local file inside a configured root.

fastvideo.entrypoints.openai.request_adapter.get_media_policy

get_media_policy() -> MediaPolicy

Return the local-root and private-address rules for caller media references.

fastvideo.entrypoints.openai.request_adapter.get_media_reference_max_bytes

get_media_reference_max_bytes() -> int

Return the server's URL media-reference byte cap.

fastvideo.entrypoints.openai.request_adapter.explicit_request_updates

explicit_request_updates(request: GenerationRequest) -> dict[str, Any]

Project a GenerationRequest down to explicitly set fields only.

Returns a flat kwargs dict suitable for merging into a generator call. The projection uses _fastvideo_explicit_paths (populated during parse_config / raw binding) so schema defaults on the dataclass are not emitted — only paths the caller/operator actually wrote.

This is what makes ServeConfig.default_request work as an operator-pinned baseline rather than a full override: a YAML with just sampling.seed: 42 yields {"seed": 42}, not the full sampling config with its 15 schema defaults.

Precondition: the request must carry _fastvideo_explicit_paths — populated by :func:fastvideo.api.parser.parse_config or :func:fastvideo.api.compat.normalize_generation_request. Calling on a raw GenerationRequest() asserts.

fastvideo.entrypoints.openai.request_adapter.legacy_generate_call_to_request

legacy_generate_call_to_request(prompt: str | None, sampling_param: SamplingParam | None, *, mouse_cond: Any | None = None, keyboard_cond: Any | None = None, grid_sizes: Any | None = None, legacy_kwargs: Mapping[str, Any] | None = None) -> GenerationRequest

fastvideo.entrypoints.openai.request_adapter.request_to_sampling_param

request_to_sampling_param(request: GenerationRequest, *, model_path: str) -> SamplingParam

fastvideo.entrypoints.openai.request_adapter.GenerationRequest

GenerationRequest(prompt: str | list[str] | None = None, negative_prompt: str | None = None, inputs: InputConfig = InputConfig(), sampling: SamplingConfig = SamplingConfig(), runtime: RequestRuntimeConfig = RequestRuntimeConfig(), output: OutputConfig = OutputConfig(), stage_overrides: dict[str, Any] = dict(), state: ContinuationState | None = None, plan: GenerationPlan | None = None, extensions: dict[str, Any] = dict())

fastvideo.entrypoints.openai.request_adapter.FileImageReference

FileImageReference()

fastvideo.entrypoints.openai.request_adapter.FileVideoReference

FileVideoReference()

fastvideo.entrypoints.openai.request_adapter.UrlImageReference

UrlImageReference()

fastvideo.entrypoints.openai.request_adapter.UrlVideoReference

UrlVideoReference()

fastvideo.entrypoints.openai.request_adapter.VideoGenerationRequest

VideoGenerationRequest()

OpenAI/vLLM-Omni-compatible video generation request.

Model-specific parameters belong in extra_params. FastVideo keeps the legacy input_reference and reference_url fields for clients that predate vLLM-Omni's typed reference objects.

fastvideo.entrypoints.openai.request_adapter.VideoGenerationRequest.validate_prompt

validate_prompt(value: str) -> str

fastvideo.entrypoints.openai.request_adapter.VideoGenerationRequest.resolve_video_params

resolve_video_params() -> VideoParams

Resolve top-level, nested, and size dimensions like vLLM-Omni.

fastvideo.entrypoints.openai.request_adapter.image_suffix

image_suffix(image_url: str) -> str

File suffix for a downloaded image; decoding inspects content, so this is only a hint.

fastvideo.entrypoints.openai.request_adapter.save_base64_image_to_path

save_base64_image_to_path(base64_data: str, target_path: str) -> str

fastvideo.entrypoints.openai.request_adapter.FastVideoArgs

FastVideoArgs(model_path: str, mode: ExecutionMode = ExecutionMode.INFERENCE, workload_type: WorkloadType = WorkloadType.T2V, distributed_executor_backend: str = 'mp', enable_streaming_ipc_queues: bool = False, ray_placement_group: PlacementGroup | None = None, ray_runtime_env: RuntimeEnv | None = None, inference_mode: bool = True, trust_remote_code: bool = False, revision: str | None = None, num_gpus: int = 1, tp_size: int = -1, sp_size: int = -1, hsdp_replicate_dim: int = 1, hsdp_shard_dim: int = -1, dist_timeout: int | None = None, pipeline_config: PipelineConfig = PipelineConfig(), preprocess_config: PreprocessConfig | None = None, lora_path: str | None = None, lora_nickname: str = 'default', lora_strength: float = 1.0, lora_target_modules: list[str] | None = None, output_type: str = 'pil', attention_backend: str | None = None, dit_cpu_offload: bool = True, use_fsdp_inference: bool = False, dit_layerwise_offload: bool = True, text_encoder_cpu_offload: bool = True, image_encoder_cpu_offload: bool = True, vae_cpu_offload: bool = True, pin_cpu_memory: bool = True, h3_sequential_load: bool | None = None, h3_encoder_split: bool | None = None, h3_encoder_nodes: int = -1, h3_encoder_workers: int = 0, video_decode_backend: str = 'h3-vae', taeh3_checkpoint: str | None = None, taeh3_chunk_size: int = 5, h3_vae_nvfp4_decoder: str | None = None, lazy_module_load: bool | None = None, vae_parallel_decode: bool = False, vae_parallel_encode: bool = False, vae_parallel_decode_strategy: str | None = None, enable_torch_compile: bool = False, regional_compile: bool = False, enable_torch_compile_text_encoder: bool = False, enable_torch_compile_vae: bool = False, enable_torch_compile_audio_vae: bool = False, torch_compile_kwargs: dict[str, Any] = dict(), torch_compile_kwargs_dit: dict[str, Any] = dict(), torch_compile_kwargs_text_encoder: dict[str, Any] = dict(), torch_compile_kwargs_vae: dict[str, Any] = dict(), torch_compile_kwargs_audio_vae: dict[str, Any] = dict(), inference_torch_compile: bool = False, h3_vsa_fp4: dict[str, Any] = dict(), disable_autocast: bool = False, VSA_sparsity: float | None = None, VSA_tile_size: int | None = None, moba_config_path: str | None = None, moba_config: dict[str, Any] = dict(), master_port: int | None = None, enable_stage_verification: bool = True, prompt_txt: str | None = None, ltx2_vae_tiling: bool | None = None, ltx2_vae_spatial_tile_size_in_pixels: int | None = None, ltx2_vae_spatial_tile_overlap_in_pixels: int | None = None, ltx2_vae_temporal_tile_size_in_frames: int | None = None, ltx2_vae_temporal_tile_overlap_in_frames: int | None = None, ltx2_initial_latent_path: str | None = None, ltx2_audio_latent_path: str | None = None, refine_enabled: bool | None = None, refine_upsampler_path: str | None = None, refine_transformer_path: str | None = None, refine_lora_path: str | None = None, refine_num_inference_steps: int | None = None, refine_guidance_scale: float | None = None, refine_add_noise: bool | None = None, refine_noise_path: str | None = None, refine_audio_noise_path: str | None = None, ltx2_refine_enabled: bool = False, ltx2_refine_upsampler_path: str | None = None, ltx2_refine_transformer_path: str | None = None, ltx2_refine_lora_path: str | None = None, ltx2_refine_num_inference_steps: int = 3, ltx2_refine_guidance_scale: float = 1.0, ltx2_refine_add_noise: bool = True, ltx2_refine_noise_path: str | None = None, ltx2_refine_audio_noise_path: str | None = None, ltx2_legacy_native_noise_order: bool = False, ltx2_use_distilled_sigmas: bool = True, model_paths: dict[str, str] = dict(), model_loaded: dict[str, bool] = (lambda: {'transformer': True, 'vae': True, 'upsampler': True})(), override_text_encoder_safetensors: str | None = None, override_text_encoder_quant: QuantizationMethods = None, transformer_quant: Any | None = None, override_transformer_cls_name: str | None = None, init_weights_from_safetensors: str = '', init_weights_from_safetensors_2: str = '', override_pipeline_cls_name: str | None = None, boundary_ratio: float | None = 0.875)

fastvideo.entrypoints.openai.request_adapter.FastVideoArgs.is_output_rank

is_output_rank: bool

Whether this process may materialize user-facing outputs.

This is runtime state assigned by an SPMD executor, not a public configuration field, so dataclass serialization and typed config parsing deliberately ignore it.

fastvideo.entrypoints.openai.request_adapter.FastVideoArgs.add_cli_args

add_cli_args(parser: FlexibleArgumentParser) -> FlexibleArgumentParser

fastvideo.entrypoints.openai.request_adapter.FastVideoArgs.from_cli_args

from_cli_args(args: argparse.Namespace) -> FastVideoArgs

fastvideo.entrypoints.openai.request_adapter.FastVideoArgs.from_kwargs

from_kwargs(**kwargs: Any) -> FastVideoArgs

fastvideo.entrypoints.openai.request_adapter.FastVideoArgs.check_fastvideo_args

check_fastvideo_args() -> None

Validate inference arguments for consistency

fastvideo.entrypoints.openai.request_adapter.FastVideoArgs.finalize_device_offload_policy

finalize_device_offload_policy(device_id: int = 0) -> bool

Apply device-local memory policy, then resolve incompatible modes.

fastvideo.entrypoints.openai.request_adapter.FastVideoArgs.disable_offload_on_unified_memory

disable_offload_on_unified_memory(device_id: int = 0, *, offload_flag: str | None = None) -> bool

Disable host offload after a worker has selected its device.

CUDA's unified-memory probe reads runtime device properties and may initialize a CUDA context. Callers must therefore use this only inside a device-owning process, after selecting and binding device_id. Returning the classification lets direct component-loader callers apply the same policy to explicit per-call overrides. When offload_flag is given, the return value says whether this policy covers that component role.

fastvideo.entrypoints.openai.request_adapter.load_image

load_image(image: str | PIL.Image.Image, convert_method: Callable[[PIL.Image.Image], PIL.Image.Image] | None = None) -> PIL.Image.Image

Loads image to a PIL Image.

Parameters

image: `str` or `PIL.Image.Image`

The image to convert to the PIL Image format.

convert_method: Callable[[PIL.Image.Image], PIL.Image.Image], *optional*

A conversion method to apply to the image after loading it. When set to None the image will be converted "RGB".

Returns

PIL.Image.Image

PIL.Image.Image: A PIL Image.

fastvideo.entrypoints.openai.request_adapter.get_preset_selection

get_preset_selection(model_path: str) -> tuple[str | None, str | None]

Return (default_preset, model_family) for a model path.

Single-lookup variant of :func:get_default_preset + :func:get_model_family; callers that need both should prefer this to avoid walking the registry twice.

fastvideo.entrypoints.openai.request_adapter.RequestAdaptationError

RequestAdaptationError()

The transport request cannot be represented by the loaded pipeline.

Source
class RequestAdaptationError(ValueError):
    """The transport request cannot be represented by the loaded pipeline."""

fastvideo.entrypoints.openai.request_adapter.discard_materialized_media

discard_materialized_media(paths: list[str]) -> None
Source
def discard_materialized_media(paths: list[str]) -> None:
    for path in paths:
        discard(path)

fastvideo.entrypoints.openai.request_adapter.prepare_reference_media

prepare_reference_media(request_id: str, request: VideoGenerationRequest, output_dir: str, *, max_bytes: int | None = None, policy: MediaPolicy | None = None) -> list[str]

Materialize media references and return the files created for this request.

Source
async def prepare_reference_media(
    request_id: str,
    request: VideoGenerationRequest,
    output_dir: str,
    *,
    max_bytes: int | None = None,
    policy: MediaPolicy | None = None,
) -> list[str]:
    """Materialize media references and return the files created for this request."""
    policy = get_media_policy() if policy is None else policy
    _image_sources(request)
    _video_sources(request)
    uploads_dir = os.path.join(os.path.abspath(output_dir), "uploads")
    created: list[str] = []

    def track(path: str) -> str:
        if path not in created:
            created.append(path)
        return path

    def is_remote(source: str) -> bool:
        return source.lower().startswith(("http://", "https://", "data:image"))

    async def materialize(source: str, index: int) -> str:
        source = source.strip()
        if not source:
            raise RequestAdaptationError("Image references must not be empty.")
        target = os.path.join(uploads_dir, f"{request_id}_{index}")
        if source.startswith("data:image"):
            try:
                local_path = track(await save_base64_image_to_path(source, target))
            except Exception as error:
                raise RequestAdaptationError(f"Unable to decode image reference: {error}") from error
        elif is_remote(source):
            local_path = track(await _materialize_image_url(source, target, max_bytes=max_bytes, policy=policy))
        else:
            local_path = _local_media(source, policy)
        try:
            await asyncio.to_thread(load_image, local_path)
        except Exception as error:
            raise RequestAdaptationError(f"Unable to decode image reference: {error}") from error
        return local_path

    try:
        index = 0
        for reference in _as_list(request.image_reference):
            if isinstance(reference, UrlImageReference):
                reference.image_url = await materialize(reference.image_url, index)
                index += 1
        if request.input_reference:
            request.input_reference = await materialize(request.input_reference, index)
            index += 1
        elif request.input_reference == "":
            request.input_reference = None
        if request.reference_url:
            request.reference_url = await materialize(request.reference_url, index)
        elif request.reference_url == "":
            request.reference_url = None

        for index, reference in enumerate(_as_list(request.video_reference)):
            if isinstance(reference, UrlVideoReference):
                remote = is_remote(reference.video_url)
                reference.video_url = await _materialize_video(reference.video_url,
                                                               os.path.join(uploads_dir, f"{request_id}_video_{index}"),
                                                               max_bytes=max_bytes,
                                                               policy=policy)
                if remote:
                    track(reference.video_url)
        if request.video_path:
            remote = is_remote(request.video_path)
            request.video_path = await _materialize_video(request.video_path,
                                                          os.path.join(uploads_dir, f"{request_id}_video_path"),
                                                          max_bytes=max_bytes,
                                                          policy=policy)
            if remote:
                track(request.video_path)
        if request.video_url:
            remote = is_remote(request.video_url)
            request.video_url = await _materialize_video(request.video_url,
                                                         os.path.join(uploads_dir, f"{request_id}_video_url"),
                                                         max_bytes=max_bytes,
                                                         policy=policy)
            if remote:
                track(request.video_url)
    except BaseException:
        # Includes cancellation, so a disconnected client leaves no downloaded inputs behind.
        discard_materialized_media(created)
        raise
    return created

fastvideo.entrypoints.openai.request_adapter.validate_served_model_name

validate_served_model_name(model: str | None, args: FastVideoArgs, served_model_name: str) -> None

Reject a request model id that is not the one this server loaded.

Source
def validate_served_model_name(
    model: str | None,
    args: FastVideoArgs,
    served_model_name: str,
) -> None:
    """Reject a request model id that is not the one this server loaded."""
    if model is None:
        return
    allowed_models = {args.lora_nickname} if args.lora_path else {served_model_name}
    if model not in allowed_models:
        choices = ", ".join(sorted(allowed_models))
        raise RequestAdaptationError(f"Model mismatch: request specifies {model!r}; this server provides {choices}.")

fastvideo.entrypoints.openai.request_adapter.validate_model_and_lora

validate_model_and_lora(request: VideoGenerationRequest, args: FastVideoArgs, served_model_name: str) -> None

Validate vLLM-style model and LoRA selectors against startup state.

FastVideo's published FastH3 adapters include dense replacement tensors in addition to low-rank factors. Those tensors are applied while the model is loaded and cannot be swapped safely between concurrent requests. The API accepts vLLM's selector shape, but it must identify the startup adapter.

Source
def validate_model_and_lora(
    request: VideoGenerationRequest,
    args: FastVideoArgs,
    served_model_name: str,
) -> None:
    """Validate vLLM-style model and LoRA selectors against startup state.

    FastVideo's published FastH3 adapters include dense replacement tensors in
    addition to low-rank factors. Those tensors are applied while the model is
    loaded and cannot be swapped safely between concurrent requests. The API
    accepts vLLM's selector shape, but it must identify the startup adapter.
    """
    validate_served_model_name(request.model, args, served_model_name)

    if request.lora is None:
        return
    if not args.lora_path:
        raise RequestAdaptationError(
            "This server has no startup LoRA. Configure generator.pipeline.components.lora_path before using "
            "the request lora selector.")

    body = request.lora
    name = next((body[key] for key in ("name", "lora_name", "adapter") if body.get(key) is not None), None)
    path = next((body[key] for key in ("path", "lora_path", "local_path") if body.get(key) is not None), None)
    scale = next((body[key] for key in ("scale", "lora_scale") if body.get(key) is not None), None)
    if name is None and path is None:
        raise RequestAdaptationError("lora must provide a name or path")
    if name is not None and str(name) != args.lora_nickname:
        raise RequestAdaptationError(f"Requested LoRA {name!r} is not the startup adapter {args.lora_nickname!r}.")
    if path is not None and str(path) != args.lora_path:
        raise RequestAdaptationError(
            f"Requested LoRA path {path!r} does not match the startup adapter {args.lora_path!r}.")
    if scale is not None:
        try:
            scale_value = float(scale)
        except (TypeError, ValueError) as error:
            raise RequestAdaptationError(f"Invalid LoRA scale {scale!r}") from error
        if not math.isclose(scale_value, args.lora_strength, rel_tol=0.0, abs_tol=1e-8):
            raise RequestAdaptationError(
                f"Requested LoRA scale {scale_value:g} does not match startup strength {args.lora_strength:g}.")

fastvideo.entrypoints.openai.request_adapter.build_generation_request

build_generation_request(request_id: str, request: VideoGenerationRequest, args: FastVideoArgs, *, served_model_name: str, output_dir: str, default_request: GenerationRequest | None = None) -> GenerationRequest

Build one tracked FastVideo request using explicit-field precedence.

Source
def build_generation_request(
    request_id: str,
    request: VideoGenerationRequest,
    args: FastVideoArgs,
    *,
    served_model_name: str,
    output_dir: str,
    default_request: GenerationRequest | None = None,
) -> GenerationRequest:
    """Build one tracked FastVideo request using explicit-field precedence."""
    validate_model_and_lora(request, args, served_model_name)
    workload_type = getattr(args, "workload_type", "")
    workload_value = str(getattr(workload_type, "value", workload_type) or "")
    is_audio = workload_value.endswith("2a")
    kwargs: dict[str, Any] = {}
    if default_request is not None:
        kwargs.update(explicit_request_updates(default_request))

    body_set = request.model_fields_set
    nested_set = request.video_params.model_fields_set if request.video_params is not None else set()
    if "size" in body_set and request.size is not None:
        width, height = request.size.split("x", 1)
        kwargs["width"], kwargs["height"] = int(width), int(height)
    else:
        if "width" in body_set and request.width is not None:
            kwargs["width"] = request.width
        elif "video_params" in body_set and "width" in nested_set and request.video_params.width is not None:
            kwargs["width"] = request.video_params.width
        if "height" in body_set and request.height is not None:
            kwargs["height"] = request.height
        elif "video_params" in body_set and "height" in nested_set and request.video_params.height is not None:
            kwargs["height"] = request.video_params.height

    fps_explicit = ("fps" in body_set
                    and request.fps is not None) or ("video_params" in body_set and "fps" in nested_set
                                                     and request.video_params.fps is not None)
    if fps_explicit:
        fps = request.fps if "fps" in body_set else request.video_params.fps
        if fps is not None:
            kwargs["fps"] = fps
    kwargs.setdefault("fps", 24)

    frames_explicit = ("num_frames" in body_set
                       and request.num_frames is not None) or ("video_params" in body_set and "num_frames" in nested_set
                                                               and request.video_params.num_frames is not None)
    if frames_explicit:
        num_frames = request.num_frames if "num_frames" in body_set else request.video_params.num_frames
        if num_frames is not None:
            kwargs["num_frames"] = num_frames
    elif not is_audio and "seconds" in body_set and request.seconds is not None:
        kwargs["num_frames"] = int(request.seconds) * int(kwargs["fps"])

    direct_fields = (
        "seed",
        "num_inference_steps",
        "guidance_scale",
        "guidance_scale_2",
        "true_cfg_scale",
        "negative_prompt",
        "enable_teacache",
        "max_sequence_length",
        "boundary_ratio",
        "audio_start_in_s",
        "audio_end_in_s",
    )
    for name in direct_fields:
        if name in body_set:
            value = getattr(request, name)
            if value is not None:
                kwargs[name] = value
    if is_audio and "seconds" in body_set and request.seconds is not None and "audio_end_in_s" not in body_set:
        start = float(kwargs.get("audio_start_in_s") or 0.0)
        kwargs["audio_start_in_s"] = start
        kwargs["audio_end_in_s"] = start + float(int(request.seconds))
    if is_audio:
        if {"size", "width", "height", "video_params"} & body_set:
            raise RequestAdaptationError("Audio generation does not accept size, width, height, or video_params.")
        start = float(kwargs.get("audio_start_in_s") or 0.0)
        end = kwargs.get("audio_end_in_s")
        if end is not None and float(end) <= start:
            raise RequestAdaptationError(f"audio_end_in_s ({end}) must be greater than audio_start_in_s ({start}).")
    if "n" in body_set or "num_outputs_per_prompt" in body_set:
        kwargs["num_videos_per_prompt"] = request.resolved_num_outputs

    try:
        _, model_family = get_preset_selection(args.model_path)
    except (RuntimeError, ValueError):
        model_family = None
    if request.resolved_num_outputs != 1:
        raise RequestAdaptationError("FastVideo serving currently supports exactly one video output per request.")
    if "short_edge" in body_set and request.short_edge is not None and request.aspect_ratio is None:
        raise RequestAdaptationError("short_edge requires aspect_ratio.")
    _apply_aspect_ratio(kwargs, request, model_family=model_family)
    _apply_reference_inputs(kwargs, request, args, model_family=model_family)
    if model_family == "mmaudio" and not workload_value.endswith("t2a") and not kwargs.get("video_path"):
        raise RequestAdaptationError("MMAudio video-to-audio generation requires video_reference or video_path.")

    extension_fields = ("flow_shift", "sound_duration", "start_time_seconds")
    for name in extension_fields:
        if name in body_set and getattr(request, name) is not None:
            kwargs[name] = getattr(request, name)
    if "generate_sound" in body_set and request.generate_sound and model_family != "minimax_h3":
        kwargs["generate_sound"] = True
    if "enable_frame_interpolation" in body_set and request.enable_frame_interpolation:
        kwargs["enable_frame_interpolation"] = True
        for name in (
                "frame_interpolation_exp",
                "frame_interpolation_scale",
                "frame_interpolation_model_path",
        ):
            kwargs[name] = getattr(request, name)
    if request.extra_params:
        unknown_extra_params = sorted(set(request.extra_params) - set(REQUEST_BATCH_EXTRA_PASSTHROUGH_FIELDS))
        if unknown_extra_params:
            raise RequestAdaptationError("Unsupported extra_params fields: " + ", ".join(unknown_extra_params))
        kwargs.update(request.extra_params)

    width = kwargs.get("width")
    height = kwargs.get("height")
    if width is not None and (not isinstance(width, int) or width <= 0):
        raise RequestAdaptationError(f"width must be a positive integer, got {width!r}")
    if height is not None and (not isinstance(height, int) or height <= 0):
        raise RequestAdaptationError(f"height must be a positive integer, got {height!r}")

    if model_family == "minimax_h3":
        from fastvideo.pipelines.basic.minimax_h3.packing import (
            MINIMAX_H3_CANVAS_MULTIPLE,
            MINIMAX_H3_MAX_PIXELS,
        )
        from fastvideo.pipelines.basic.minimax_h3.stages.minimax_h3_input_preparation import (
            resolve_target_num_frames, )

        if kwargs["fps"] != 24:
            raise RequestAdaptationError(f"MiniMax-H3 requires fps=24, got {kwargs['fps']}.")
        if width is None or height is None:
            raise RequestAdaptationError("MiniMax-H3 requires both width and height.")
        if width % MINIMAX_H3_CANVAS_MULTIPLE or height % MINIMAX_H3_CANVAS_MULTIPLE:
            raise RequestAdaptationError("MiniMax-H3 width and height must be positive multiples of "
                                         f"{MINIMAX_H3_CANVAS_MULTIPLE}, got {width}x{height}.")
        if width * height > MINIMAX_H3_MAX_PIXELS:
            raise RequestAdaptationError(
                f"MiniMax-H3 canvas exceeds the {MINIMAX_H3_MAX_PIXELS}-pixel limit: {width}x{height}.")
        try:
            requested_num_frames = kwargs.get("num_frames")
            aligned_num_frames = resolve_target_num_frames(requested_num_frames)
        except (TypeError, ValueError) as error:
            raise RequestAdaptationError(str(error)) from error
        if frames_explicit and aligned_num_frames != requested_num_frames:
            raise RequestAdaptationError("MiniMax-H3 num_frames must be on the causal-VAE grid (17 * n + 5); "
                                         f"got {requested_num_frames}, next valid value is {aligned_num_frames}.")
        kwargs["num_frames"] = aligned_num_frames

    media_dir = "audio" if is_audio else "videos"
    extension = ".wav" if is_audio else ".mp4"
    output_path = os.path.join(os.path.abspath(output_dir), media_dir, f"{request_id}{extension}")
    os.makedirs(os.path.dirname(output_path), exist_ok=True)
    kwargs.update({
        "output_path": output_path,
        "save_video": True,
        "return_frames": False,
    })
    generation_request = legacy_generate_call_to_request(request.prompt, None, legacy_kwargs=kwargs)
    try:
        # Resolve once at admission time so unsupported model-specific fields
        # are a deterministic 400, rather than an asynchronous failed job.
        request_to_sampling_param(generation_request, model_path=args.model_path)
    except (TypeError, ValueError) as error:
        raise RequestAdaptationError(str(error)) from error
    return generation_request