--- title: mlx_server ---

Serve native FastH3 MLX through the shared video-job API and playground.

fastvideo.entrypoints.openai.mlx_server.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.mlx_server.normalize_generation_request

normalize_generation_request(request: GenerationRequest | Mapping[str, Any]) -> GenerationRequest

fastvideo.entrypoints.openai.mlx_server.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.mlx_server.create_app

create_app(fastvideo_args: FastVideoArgs, output_dir: str = DEFAULT_OUTPUT_DIR, default_request: GenerationRequest | None = None, served_model_name: str | None = None, *, generator_factory: Callable[[], ServingGenerator] | None = None, video_request_validator: Callable[[VideoGenerationRequest], None] | None = None, runtime: str = 'cuda', media_reference_max_bytes: int = DEFAULT_MEDIA_REFERENCE_MAX_BYTES, media_policy: MediaPolicy | None = None) -> FastAPI

Build the FastAPI application with all routers mounted

fastvideo.entrypoints.openai.mlx_server.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.mlx_server.VideoGenerationRequest.validate_prompt

validate_prompt(value: str) -> str

fastvideo.entrypoints.openai.mlx_server.VideoGenerationRequest.resolve_video_params

resolve_video_params() -> VideoParams

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

fastvideo.entrypoints.openai.mlx_server.mlx_http_steps

mlx_http_steps(model_path: str) -> int
Source
def mlx_http_steps(model_path: str) -> int:
    return 9 if model_path == EIGHT_STEP_MODEL else 5

fastvideo.entrypoints.openai.mlx_server.mlx_num_steps

mlx_num_steps(num_inference_steps: int | None, *, model_path: str) -> int
Source
def mlx_num_steps(num_inference_steps: int | None, *, model_path: str) -> int:
    expected = mlx_http_steps(model_path)
    if num_inference_steps not in (None, expected):
        raise ValueError(f"This FastH3 MLX server uses {expected} sigma points "
                         f"({HTTP_STEPS_TO_FORWARDS[expected]} transformer forwards).")
    return HTTP_STEPS_TO_FORWARDS[expected]

fastvideo.entrypoints.openai.mlx_server.MLXGeneratorConfig

MLXGeneratorConfig()
Source
class MLXGeneratorConfig(BaseModel):
    model_config = ConfigDict(extra="forbid")
    model_path: Literal["FastVideo/FastVideo-Minimax-FastH3-Preview-v0.2",
                        "FastVideo/FastVideo-FastH3-8-Step-V2"] = PREVIEW_MODEL
    model_root: str
    mlx_checkpoint: str
    prompt_cache_dir: str = "outputs/h3_prompt_cache"
    vae_dtype: Literal["fp32", "fp16", "bf16"] = "fp32"
    vsa: bool = False
    vsa_sparsity: float = Field(default=0.9, ge=0.0, lt=1.0)
    vsa_tile_size: Literal[64, 256] = 64

fastvideo.entrypoints.openai.mlx_server.MLXServerConfig

MLXServerConfig()
Source
class MLXServerConfig(BaseModel):
    model_config = ConfigDict(extra="forbid")
    host: str = "127.0.0.1"
    port: int = Field(default=8000, ge=1, le=65535)
    output_dir: str = "outputs/mlx_fasth3"
    served_model_name: str = Field(default="fasth3", min_length=1)

fastvideo.entrypoints.openai.mlx_server.MLXServeConfig

MLXServeConfig()
Source
class MLXServeConfig(BaseModel):
    model_config = ConfigDict(extra="forbid")
    runtime: Literal["mlx"]
    generator: MLXGeneratorConfig
    server: MLXServerConfig = Field(default_factory=MLXServerConfig)
    default_request: dict[str, Any]

fastvideo.entrypoints.openai.mlx_server.validate_mlx_video_request

validate_mlx_video_request(request: VideoGenerationRequest, *, model_path: str = PREVIEW_MODEL) -> None

Reject unsupported inputs before fetching media or creating a job.

Source
def validate_mlx_video_request(request: VideoGenerationRequest, *, model_path: str = PREVIEW_MODEL) -> None:
    """Reject unsupported inputs before fetching media or creating a job."""
    allowed = {
        "model",
        "prompt",
        "seed",
        "size",
        "width",
        "height",
        "fps",
        "num_frames",
        "seconds",
        "video_params",
        "task",
        "guidance_scale",
        "num_inference_steps",
        "negative_prompt",
    }
    unsupported = request.model_fields_set - allowed
    if unsupported:
        raise ValueError("H3 MLX serving does not support: " + ", ".join(sorted(unsupported)))
    if request.task not in (None, "t2va"):
        raise ValueError("H3 MLX serving supports task=t2va only.")
    if request.guidance_scale not in (None, 1.0):
        raise ValueError("FastH3 MLX requires guidance_scale=1.")
    if request.negative_prompt not in (None, ""):
        raise ValueError("FastH3 MLX does not use a negative prompt.")
    expected = mlx_http_steps(model_path)
    if request.num_inference_steps not in (None, expected):
        raise ValueError(f"This FastH3 MLX server uses {expected} sigma points "
                         f"({HTTP_STEPS_TO_FORWARDS[expected]} transformer forwards).")
    if request.seed is not None and not 0 <= request.seed <= 2**32 - 1:
        raise ValueError("H3 MLX seed must be between 0 and 4294967295.")

fastvideo.entrypoints.openai.mlx_server.MLXH3Generator

MLXH3Generator(config: MLXGeneratorConfig)

Keep one pipeline on one MLX thread; preserve its phase-memory policy.

Source
class MLXH3Generator:
    """Keep one pipeline on one MLX thread; preserve its phase-memory policy."""

    def __init__(self, config: MLXGeneratorConfig) -> None:
        self._model_path = config.model_path
        self._vsa = config.vsa
        self._vsa_sparsity = config.vsa_sparsity
        self._vsa_tile_size = config.vsa_tile_size
        self._worker = ThreadPoolExecutor(max_workers=1, thread_name_prefix="h3-mlx")
        try:
            self._pipeline = self._worker.submit(self._load, config).result()
        except BaseException:
            self._worker.shutdown(wait=True)
            raise

    @staticmethod
    def _load(config: MLXGeneratorConfig):
        if platform.system() != "Darwin" or platform.machine() != "arm64":
            raise RuntimeError("H3 MLX serving requires an Apple Silicon Mac.")
        if shutil.which("ffmpeg") is None:
            raise RuntimeError("Install ffmpeg before starting the H3 MLX server.")
        from fastvideo.mlx_runtime.minimax_h3_pipeline import MiniMaxH3MLXPipeline

        return MiniMaxH3MLXPipeline(
            model_root=Path(config.model_root).expanduser(),
            mlx_dit_checkpoint=Path(config.mlx_checkpoint).expanduser(),
            prompt_cache_dir=Path(config.prompt_cache_dir).expanduser(),
            vae_dtype=config.vae_dtype,
        )

    def generate(self, request: GenerationRequest) -> dict[str, Any]:
        return self._worker.submit(self._generate, request).result()

    def _generate(self, request: GenerationRequest) -> dict[str, Any]:
        started = time.perf_counter()
        generate_kwargs: dict[str, Any] = {
            "output_path": request.output.output_path,
            "width": request.sampling.width,
            "height": request.sampling.height,
            "num_frames": request.sampling.num_frames,
            "seed": request.sampling.seed,
            "num_steps": mlx_num_steps(request.sampling.num_inference_steps, model_path=self._model_path),
        }
        if self._vsa:
            generate_kwargs.update(
                vsa=True,
                vsa_sparsity=self._vsa_sparsity,
                vsa_tile_size=self._vsa_tile_size,
            )
        result = self._pipeline.generate(request.prompt, **generate_kwargs)
        # Do not retain decoded frames/waveforms or label phase peaks as total RAM.
        return {"video_path": str(result.video_path), "generation_time": time.perf_counter() - started}

    def shutdown(self) -> None:

        def release():
            self._pipeline = None
            from fastvideo.mlx_runtime.minimax_h3_pipeline import _cleanup_mlx

            _cleanup_mlx()

        try:
            self._worker.submit(release).result()
        finally:
            self._worker.shutdown(wait=True)

fastvideo.entrypoints.openai.mlx_server.MLXH3Generator.generate

generate(request: GenerationRequest) -> dict[str, Any]
Source
def generate(self, request: GenerationRequest) -> dict[str, Any]:
    return self._worker.submit(self._generate, request).result()

fastvideo.entrypoints.openai.mlx_server.MLXH3Generator.shutdown

shutdown() -> None
Source
def shutdown(self) -> None:

    def release():
        self._pipeline = None
        from fastvideo.mlx_runtime.minimax_h3_pipeline import _cleanup_mlx

        _cleanup_mlx()

    try:
        self._worker.submit(release).result()
    finally:
        self._worker.shutdown(wait=True)

fastvideo.entrypoints.openai.mlx_server.load_config

load_config(path: str) -> MLXServeConfig
Source
def load_config(path: str) -> MLXServeConfig:
    with open(path, encoding="utf-8") as source:
        return MLXServeConfig.model_validate(yaml.safe_load(source))

fastvideo.entrypoints.openai.mlx_server.create_mlx_app

create_mlx_app(config: MLXServeConfig)
Source
def create_mlx_app(config: MLXServeConfig):
    request = normalize_generation_request(config.default_request)
    explicit = explicit_request_updates(request)
    supported = {
        "width", "height", "num_frames", "fps", "seed", "num_inference_steps", "guidance_scale", "negative_prompt"
    }
    if set(explicit) - supported:
        raise ValueError("MLX default_request contains unsupported fields: " +
                         ", ".join(sorted(set(explicit) - supported)))
    required = {"width", "height", "num_frames", "fps", "seed", "num_inference_steps", "guidance_scale"}
    if required - set(explicit):
        raise ValueError("MLX default_request must set: " + ", ".join(sorted(required - set(explicit))))
    validate_mlx_video_request(VideoGenerationRequest(prompt="validate config", **explicit),
                               model_path=config.generator.model_path)
    # Transport admission uses the registered H3 family, not CUDA engine options.
    args = SimpleNamespace(model_path=config.generator.model_path,
                           lora_path=None,
                           lora_nickname="default",
                           lora_strength=1.0,
                           override_pipeline_cls_name=None)
    from fastvideo.entrypoints.openai.request_adapter import build_generation_request

    build_generation_request("config-check",
                             VideoGenerationRequest(prompt="validate config"),
                             args,
                             served_model_name=config.server.served_model_name,
                             output_dir=config.server.output_dir,
                             default_request=request)

    def video_request_validator(request: VideoGenerationRequest) -> None:
        validate_mlx_video_request(request, model_path=config.generator.model_path)

    return create_app(
        args,
        config.server.output_dir,
        request,
        config.server.served_model_name,
        generator_factory=lambda: MLXH3Generator(config.generator),
        video_request_validator=video_request_validator,
        runtime="mlx",
    )

fastvideo.entrypoints.openai.mlx_server.main

main() -> None
Source
def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config",
                        required=True,
                        help="H3 MLX serving YAML; paths are relative to the working directory")
    args = parser.parse_args()
    config = load_config(args.config)
    uvicorn.run(create_mlx_app(config), host=config.server.host, port=config.server.port)