Kling AI API Tutorial: Build AI Video Generation into Your App

import requests
import time

API_KEY = "your-crazyrouter-key"
BASE_URL = "https://api.crazyrouter.com"
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

# Generate a video from text
response = requests.post(
    f"{BASE_URL}/kling/submit/video",
    headers=HEADERS,
    json={
        "prompt": "A barista carefully pouring latte art in a cozy coffee shop, warm morning light streaming through the window, close-up shot",
        "duration": 5,
        "mode": "standard",  # "standard" or "professional"
        "aspect_ratio": "16:9"
    }
)

task = response.json()
task_id = task["data"]["task_id"]
print(f"Task ID: {task_id}")

# Wait for completion
def wait_for_video(task_id, max_wait=300):
    start = time.time()
    while time.time() - start < max_wait:
        result = requests.get(
            f"{BASE_URL}/kling/fetch/{task_id}",
            headers=HEADERS
        ).json()

        status = result["data"]["status"]
        if status == "completed":
            return result["data"]
        elif status == "failed":
            raise Exception(f"Failed: {result['data'].get('error')}")

        print(f"  Status: {status}")
        time.sleep(10)

    raise TimeoutError("Video generation timed out")

result = wait_for_video(task_id)
print(f"Video URL: {result['video_url']}")

# Animate a product photo
response = requests.post(
    f"{BASE_URL}/kling/submit/video",
    headers=HEADERS,
    json={
        "prompt": "The person slowly smiles and turns their head to the right, hair gently moving in the breeze",
        "image_url": "https://example.com/portrait.jpg",
        "duration": 5,
        "mode": "professional",
        "aspect_ratio": "9:16"  # Portrait for social media
    }
)

# Generate a lip-synced video
response = requests.post(
    f"{BASE_URL}/kling/submit/lip-sync",
    headers=HEADERS,
    json={
        "video_url": "https://example.com/talking-head.mp4",
        "audio_url": "https://example.com/speech.mp3"
    }
)

# Text-to-video
curl -X POST https://api.crazyrouter.com/kling/submit/video \
  -H "Authorization: Bearer your-crazyrouter-key" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "Ocean waves crashing on a rocky shore at sunset, drone shot, 4K cinematic",
    "duration": 5,
    "mode": "professional",
    "aspect_ratio": "16:9"
  }'

# Check status
curl https://api.crazyrouter.com/kling/fetch/TASK_ID \
  -H "Authorization: Bearer your-crazyrouter-key"

[Subject] + [Action] + [Setting] + [Camera] + [Style]

"A chef flipping a pancake in a modern kitchen, medium shot, warm lighting, slow motion"

"Two friends laughing while walking through autumn leaves in a park, tracking shot, golden hour, cinematic"

"A cat stretching and yawning on a windowsill, close-up, soft natural light, cozy atmosphere"

class KlingVideoClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.crazyrouter.com"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def generate_video(self, prompt: str, duration: int = 5,
                       mode: str = "standard", image_url: str = None,
                       aspect_ratio: str = "16:9") -> dict:
        """Generate a video and wait for completion."""
        payload = {
            "prompt": prompt,
            "duration": duration,
            "mode": mode,
            "aspect_ratio": aspect_ratio
        }
        if image_url:
            payload["image_url"] = image_url

        response = requests.post(
            f"{self.base_url}/kling/submit/video",
            headers=self.headers,
            json=payload
        )
        response.raise_for_status()
        task_id = response.json()["data"]["task_id"]

        # Poll for result
        for _ in range(60):  # Max 10 minutes
            result = requests.get(
                f"{self.base_url}/kling/fetch/{task_id}",
                headers=self.headers
            ).json()

            if result["data"]["status"] == "completed":
                return result["data"]
            elif result["data"]["status"] == "failed":
                raise Exception(result["data"].get("error", "Unknown error"))

            time.sleep(10)

        raise TimeoutError("Generation timed out")

    def batch_generate(self, prompts: list, **kwargs) -> list:
        """Generate multiple videos concurrently."""
        import concurrent.futures

        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            futures = {
                executor.submit(self.generate_video, prompt, **kwargs): prompt
                for prompt in prompts
            }
            results = []
            for future in concurrent.futures.as_completed(futures):
                try:
                    result = future.result()
                    results.append({"prompt": futures[future], "result": result})
                except Exception as e:
                    results.append({"prompt": futures[future], "error": str(e)})
            return results


# Usage
kling = KlingVideoClient("your-crazyrouter-key")

# Single video
video = kling.generate_video(
    "A time-lapse of a flower blooming, macro shot, studio lighting",
    duration=5,
    mode="professional"
)
print(f"Video: {video['video_url']}")

# Batch generation
prompts = [
    "Sunrise over mountains, aerial drone shot",
    "Rain falling on a city street at night, neon reflections",
    "A butterfly landing on a flower, macro, slow motion"
]
results = kling.batch_generate(prompts, duration=5)
for r in results:
    print(f"{r['prompt'][:50]}... → {r.get('result', {}).get('video_url', r.get('error'))}")