AI Video Generation: API Integration & Self-Hosted Solutions

AI Video Generation: API Integration & Self-Hosted Solutions

A studio requested automatic creation of 10-second ad clips from product photos. Previously, designers edited manually — 2 clips per day. We integrated Kling 1.5 and Runway Gen-3 APIs, added a priority queue and post-processing. Throughput increased to 50 clips per day. This is a typical case: AI video generation from text or image prompts is no longer a toy but a working tool for marketing, production, gamedev, and education.

What Problems We Solve

• Unstable quality: different prompts yield different results. Our integration with Kling 1.5 allows specifying negative prompts, CFG scale, and pro mode for better control. For Runway, we pick a 1280:768 aspect ratio and 10-second duration — enough for short ads. Kling 1.5 offers 30% more control over Runway via negative prompts and cfg_scale.
• Generation latency: cloud API wait times can reach 5 minutes. We build async pipelines with WebSocket notifications and Redis queues. In a self-hosted setup (CogVideoX on A100), latency drops to 2 minutes for a 6-second clip.
• Cost at scale: each Kling request costs ~$0.10. At 5,000 clips/month, that's $500. Self-hosted CogVideoX is 2–3x cheaper than cloud APIs for large volumes — if traffic exceeds 10,000 requests/month, we recommend switching to your own GPUs, which can yield monthly savings of $800–$1,000.

Integrating Kling API

For one marketplace, we implemented text-to-video and image-to-video via Kling. Python code with httpx.AsyncClient and asyncio handles up to 100 parallel requests. Example KlingVideoGenerator class below.

import httpx
import asyncio
import json

class KlingVideoGenerator:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.klingai.com/v1"

    async def text_to_video(
        self,
        prompt: str,
        negative_prompt: str = "",
        duration: int = 5,  # 5 or 10 seconds
        aspect_ratio: str = "16:9",  # 16:9, 9:16, 1:1
        mode: str = "std",  # std (faster) or pro (quality)
        cfg_scale: float = 0.5
    ) -> str:
        """Create generation task, return task_id"""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.base_url}/videos/text2video",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={
                    "prompt": prompt,
                    "negative_prompt": negative_prompt,
                    "cfg_scale": cfg_scale,
                    "mode": mode,
                    "duration": str(duration),
                    "aspect_ratio": aspect_ratio
                }
            )
            return resp.json()["data"]["task_id"]

    async def image_to_video(
        self,
        image_url: str,
        prompt: str = "",
        duration: int = 5,
        motion_intensity: float = 0.5
    ) -> str:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.base_url}/videos/image2video",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={
                    "image_url": image_url,
                    "prompt": prompt,
                    "duration": str(duration),
                    "cfg_scale": motion_intensity
                }
            )
            return resp.json()["data"]["task_id"]

    async def wait_for_result(self, task_id: str, timeout: int = 300) -> str:
        """Poll until done, return video URL"""
        async with httpx.AsyncClient() as client:
            for _ in range(timeout // 5):
                await asyncio.sleep(5)
                resp = await client.get(
                    f"{self.base_url}/videos/text2video/{task_id}",
                    headers={"Authorization": f"Bearer {self.api_key}"}
                )
                data = resp.json()["data"]
                if data["task_status"] == "succeed":
                    return data["task_result"]["videos"][0]["url"]
                elif data["task_status"] == "failed":
                    raise RuntimeError(f"Generation failed: {data.get('task_status_msg')}")
        raise TimeoutError(f"Video generation timeout after {timeout}s")

Similarly, Runway Gen-3 is integrated. The difference is in the response format and support for image-to-video with prompt_image.

Why Choose Self-Hosted CogVideoX?

Note: when the monthly generation volume exceeds 10,000 minutes, cloud APIs become more expensive than a GPU server. CogVideoX-5b on A100 (80 GB) generates 49 frames in 2 minutes — ~6 seconds of video at 8 FPS. Resolution is 720p, but quality approaches Kling. You have full pipeline control, can fine-tune on your own data (LoRA), and pay no per-request fees.

from diffusers import CogVideoXPipeline
import torch

pipe = CogVideoXPipeline.from_pretrained(
    "THUDM/CogVideoX-5b",
    torch_dtype=torch.bfloat16
)
pipe.enable_model_cpu_offload()
pipe.vae.enable_tiling()

def generate_video_local(
    prompt: str,
    num_frames: int = 49,  # ~6 sec at 8 fps
    guidance_scale: float = 6.0
) -> str:
    video_frames = pipe(
        prompt=prompt,
        num_videos_per_prompt=1,
        num_inference_steps=50,
        num_frames=num_frames,
        guidance_scale=guidance_scale,
        generator=torch.Generator("cpu").manual_seed(42)
    ).frames[0]

    output_path = "/tmp/output_video.mp4"
    from diffusers.utils import export_to_video
    export_to_video(video_frames, output_path, fps=8)
    return output_path

How We Structure the Development Process?

1. Analytics and model selection — test Kling, Runway, Sora (if available), CogVideoX on your data. Collect p99 latency and cost metrics.
2. Architecture design — decide on pattern: API-only (fast) or self-hosted (cheap at scale). Document the scheme.
3. Integration implementation — write async clients, queue handlers, S3 storage. For self-hosted, deploy with Triton Inference Server and ONNX Runtime for acceleration.
4. Testing — A/B compare with manual editing. Measure FID/CLIP and subjective quality.
5. Deployment and monitoring — set up CI/CD, logging in Loki, alerts in Telegram.

Estimated Timelines

• Single API integration (Kling/Runway): 3 to 5 days.
• Platform with multiple providers, queue, and storage: 3 to 4 weeks.
• Self-hosted CogVideoX + optimization: 4 to 6 weeks.

What's Included in the Work

• API and architecture documentation
• Access to the code repository
• Team training (2–3 sessions)
• 3-month warranty on the integration
• Support during launch (first 2 weeks)

Platform Comparison

Applications by Niche

We have over 10 years of experience in AI and machine learning, having completed 50+ integration projects for video generation pipelines. Our team specializes in MLOps and API orchestration, ensuring reliable and scalable solutions. If you need AI video generation, contact us — we'll evaluate your project in 2 days and propose the optimal solution. Order a pilot project to see it in action.