How to Use Kling O3 Video API — Complete Tutorial 2026

pip install requests

import requests
import os
import time

# Get your API key from RapidAPI: https://rapidapi.com/nexaquency/api/kling-o3-video
RAPIDAPI_KEY = os.environ.get("RAPIDAPI_KEY", "your-rapidapi-key-here")
API_HOST = "kling-o3-video.p.rapidapi.com"

def generate_video(prompt: str, duration: int = 5, **kwargs) -> dict:
    """
    Generate a video using Kling O3 Video API via NexaAPI on RapidAPI.
    
    Args:
        prompt: Text description of the video to generate
        duration: Video duration in seconds (default: 5)
        **kwargs: Additional parameters (aspect_ratio, mode, etc.)
    
    Returns:
        dict with video URL or task ID for async generation
    """
    url = f"https://{API_HOST}/generate"
    
    headers = {
        "x-rapidapi-key": RAPIDAPI_KEY,
        "x-rapidapi-host": API_HOST,
        "Content-Type": "application/json"
    }
    
    payload = {
        "prompt": prompt,
        "duration": duration,
        **kwargs
    }
    
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()


def poll_video_status(task_id: str, max_wait: int = 300) -> dict:
    """
    Poll for async video generation completion.
    
    Args:
        task_id: The task ID returned from generate_video
        max_wait: Maximum seconds to wait (default: 300)
    
    Returns:
        dict with completed video URL
    """
    start_time = time.time()
    
    while time.time() - start_time < max_wait:
        response = requests.get(
            f"https://{API_HOST}/status/{task_id}",
            headers={
                "x-rapidapi-key": RAPIDAPI_KEY,
                "x-rapidapi-host": API_HOST
            }
        )
        response.raise_for_status()
        result = response.json()
        
        status = result.get("status")
        if status == "completed":
            return result
        elif status == "failed":
            raise Exception(f"Video generation failed: {result.get('error')}")
        
        print(f"Status: {status}. Waiting 10 seconds...")
        time.sleep(10)
    
    raise TimeoutError(f"Video generation timed out after {max_wait} seconds")


if __name__ == "__main__":
    print(f"Generating video with Kling O3 Video...")
    
    result = generate_video(
        prompt="a majestic eagle soaring over snow-capped mountains at golden hour, cinematic",
        duration=5,
        aspect_ratio="16:9"
    )
    
    # Handle async generation
    if "task_id" in result:
        print(f"Task ID: {result['task_id']}. Polling for completion...")
        final_result = poll_video_status(result["task_id"])
        video_url = final_result.get("url") or final_result.get("video_url")
    else:
        video_url = result.get("url") or result.get("video_url")
    
    print(f"Video URL: {video_url}")
    print(f"Cost: $0.033/video via NexaAPI")

import requests
import time
import os

RAPIDAPI_KEY = os.environ.get("RAPIDAPI_KEY")
API_HOST = "kling-o3-video.p.rapidapi.com"

def api_call_with_retry(endpoint: str, payload: dict, max_retries: int = 3) -> dict:
    """Make API call with exponential backoff retry."""
    for attempt in range(max_retries):
        try:
            response = requests.post(
                f"https://{API_HOST}/{endpoint}",
                json=payload,
                headers={
                    "x-rapidapi-key": RAPIDAPI_KEY,
                    "x-rapidapi-host": API_HOST,
                    "Content-Type": "application/json"
                },
                timeout=120  # 2 minute timeout for generation
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                wait = 2 ** attempt
                print(f"Rate limited. Waiting {wait}s...")
                time.sleep(wait)
            elif e.response.status_code >= 500:
                print(f"Server error. Retry {attempt+1}/{max_retries}")
                time.sleep(2)
            else:
                raise
        except requests.exceptions.Timeout:
            print(f"Timeout. Retry {attempt+1}/{max_retries}")
            time.sleep(5)
    raise Exception(f"Failed after {max_retries} retries")