How to Add Support for a New Video Generation API Provider to ViMax

You can add a new video generation API provider to ViMax by implementing a class that follows the VideoGenerator protocol in tools/protocols.py, registering it in tools/__init__.py, and referencing it via class_path in a YAML configuration file.

ViMax uses a modular, plug-in architecture that separates video generation logic from the core pipeline. This design allows you to integrate any third-party video API—whether it's a commercial service like Minimax, Veo, or a custom provider—without modifying existing pipeline code in pipelines/script2video_pipeline.py.

Understanding the VideoGenerator Protocol

All video generators in ViMax must satisfy the protocol defined in tools/protocols.py. The only strict requirement is an asynchronous method named generate_single_video that accepts specific parameters and returns a VideoOutput object. The RenderBackend factory in tools/render_backend.py dynamically instantiates your class based on a YAML config and automatically injects rate limiting if configured.

Step 1: Implement the Generator Class

Create a new Python file under the tools/ directory, such as tools/video_generator_myprovider_api.py. Your class does not need to inherit from any base class; it only needs to implement the required interface.

Required Method Signature

The generate_single_video method must have this exact signature:

async def generate_single_video(
    self,
    prompt: str,
    reference_image_paths: List[str],
    **kwargs,
) -> VideoOutput: ...

VideoOutput is imported from interfaces/video_output.py and serves as the standard return container for all generators.

Here is a complete implementation skeleton for a hypothetical provider:

# tools/video_generator_myprovider_api.py

import logging
import aiohttp
import asyncio
from typing import List, Optional
from interfaces.video_output import VideoOutput
from utils.image import image_path_to_b64

class VideoGeneratorMyProviderAPI:
    """Video generator for the MyProvider service."""

    def __init__(
        self,
        api_key: str,
        model: str = "myprovider-default-model",
        rate_limiter: Optional[RateLimiter] = None,
    ):
        self.api_key = api_key
        self.model = model
        self.rate_limiter = rate_limiter
        self.base_url = "https://api.myprovider.com/v1"

    async def generate_single_video(
        self,
        prompt: str,
        reference_image_paths: List[str],
        resolution: str = "1080p",
        aspect_ratio: str = "16:9",
        **kwargs,
    ) -> VideoOutput:
        """Create a video generation task, poll until completion, and return a VideoOutput."""
        # Select model variant based on reference image count

        model = self.model
        if len(reference_image_paths) == 1:
            model = f"{self.model}-first-frame"
        elif len(reference_image_paths) == 2:
            model = f"{self.model}-first-last-frame"

        payload = {
            "prompt": prompt,
            "model": model,
            "resolution": resolution,
            "aspect_ratio": aspect_ratio,
            "images": [image_path_to_b64(p, mime=True) for p in reference_image_paths],
        }

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

        if self.rate_limiter:
            await self.rate_limiter.acquire()

        # Create generation task via POST request

        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/videos/create", 
                json=payload, 
                headers=headers
            ) as resp:
                resp_json = await resp.json()
                task_id = resp_json["task_id"]
                logging.info(f"MyProvider task created: {task_id}")

        # Poll for task completion

        while True:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    f"{self.base_url}/videos/status/{task_id}", 
                    headers=headers
                ) as resp:
                    status_json = await resp.json()
            
            if status_json["status"] == "completed":
                video_url = status_json["video_url"]
                break
            if status_json["status"] == "failed":
                raise RuntimeError(f"MyProvider generation failed: {status_json}")
            
            await asyncio.sleep(2)

        return VideoOutput(fmt="url", ext="mp4", data=video_url)

Handling HTTP and Rate Limiting

Notice that the __init__ method accepts an optional rate_limiter parameter. When you configure rate limits in YAML, RenderBackend.from_config automatically constructs a RateLimiter instance and passes it to your class constructor. Always check if self.rate_limiter and call await self.rate_limiter.acquire() before making API requests.

Step 2: Register the Class in tools/init.py

To make your generator importable via a short module path, add it to tools/__init__.py:

# tools/__init__.py

from .video_generator_myprovider_api import VideoGeneratorMyProviderAPI

__all__ = [
    # ... existing exports

    "VideoGeneratorMyProviderAPI",
]

This registration step allows the YAML configuration parser to locate your class using dot-notation paths.

Step 3: Configure the Provider via YAML

ViMax uses YAML configuration files (such as configs/script2video.yaml) to wire components. Add a video_generator block that points to your new class:

# configs/script2video.yaml

video_generator:
  class_path: "tools.video_generator_myprovider_api.VideoGeneratorMyProviderAPI"
  init_args:
    api_key: "${MYPROVIDER_API_KEY}"  # Environment variables are supported

    model: "myprovider-high-res"
  max_requests_per_minute: 30  # Optional rate limiting

The RenderBackend factory reads this configuration, imports the class specified in class_path, and instantiates it with the supplied init_args.

Step 4: Integration with Pipelines

Once configured, your provider works automatically with existing pipelines. In pipelines/script2video_pipeline.py, the pipeline calls:

video_output = await self.video_generator.generate_single_video(
    prompt=prompt,
    reference_image_paths=ref_images,
    resolution="1080p",
    aspect_ratio="16:9",
)

Because self.video_generator is injected by RenderBackend based on the YAML config, no pipeline code changes are required when switching providers.

Step 5: Testing Your Implementation (Optional)

Create unit tests that mock aiohttp endpoints to verify polling logic and error handling. Follow the patterns in tests/test_provider_presets.py and tests/test_minimax_integration.py, ensuring your test asserts that generate_single_video returns a valid VideoOutput instance with the expected fmt and data attributes.

Summary

• Implement an async class with generate_single_video(prompt, reference_image_paths, **kwargs) -> VideoOutput in tools/.
• Export the class in tools/__init__.py to make it discoverable.
• Configure the provider in a YAML file using class_path under the video_generator key.
• Leverage optional rate limiting by accepting a rate_limiter parameter in __init__.
• Deploy without changing pipeline code, as RenderBackend handles dependency injection.

Frequently Asked Questions

Do I need to inherit from a base class when adding a new video generation API provider to ViMax?

No. ViMax uses structural subtyping (duck typing) rather than inheritance. Your class only needs to implement the generate_single_video method with the correct signature as defined in tools/protocols.py. The RenderBackend factory validates the interface at runtime based on the protocol definition.

How does rate limiting work with custom video generators?

When you specify max_requests_per_minute or max_requests_per_day in the YAML configuration, RenderBackend.from_config automatically creates a RateLimiter instance and injects it into your generator's __init__ method via the rate_limiter parameter. You simply call await self.rate_limiter.acquire() before making HTTP requests to block until the rate limit allows the next request.

Can I use synchronous HTTP clients like requests instead of aiohttp?

No, you must use async/await patterns compatible with the existing ViMax pipeline. The generate_single_video method is called within an async context in pipelines/script2video_pipeline.py. Use aiohttp for HTTP requests or asyncio compatible libraries to avoid blocking the event loop.

Where should API keys be stored for the new provider?

Store sensitive credentials in environment variables and reference them in the YAML configuration using the ${ENV_VAR_NAME} substitution syntax. The configuration loader resolves these variables before passing values to your class's __init__ method, keeping secrets out of version control.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}