Kling AI Camera Control

Overview

Add controlled camera movements to text-to-video and image-to-video generation using the camera_control parameter. Supports pan, tilt, zoom, roll, and dolly. Available on v1.6+ models.

Supports: POST /v1/videos/text2video and POST /v1/videos/image2video

Camera Control Parameters

"camera_control": {
    "type": "simple",     # "simple" = one movement axis
    "config": {
        "horizontal": 0,  # Pan: -10 (left) to 10 (right)
        "vertical": 0,    # Tilt: -10 (down) to 10 (up)
        "zoom": 0,         # Zoom: -10 (out) to 10 (in)
        "roll": 0,         # Roll: -10 (CCW) to 10 (CW)
        "pan": 0,          # Dolly: -10 (left) to 10 (right)
        "tilt": 0,         # Dolly: -10 (down) to 10 (up)
    }
}

Rule: For type: "simple", only ONE field in config can be non-zero.

Cinematic Shot Examples

Slow Pan Right

response = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
    "model_name": "kling-v2-6",
    "prompt": "A medieval castle on a cliff at sunrise, fog in the valley below",
    "duration": "5",
    "mode": "professional",
    "camera_control": {
        "type": "simple",
        "config": {"horizontal": 5, "vertical": 0, "zoom": 0, "roll": 0, "pan": 0, "tilt": 0}
    },
})

Dramatic Zoom In

response = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
    "model_name": "kling-v2-6",
    "prompt": "Close-up of a tiger's face, intense eyes, jungle background",
    "duration": "5",
    "mode": "professional",
    "camera_control": {
        "type": "simple",
        "config": {"horizontal": 0, "vertical": 0, "zoom": 7, "roll": 0, "pan": 0, "tilt": 0}
    },
})

Tilt Up Reveal

response = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
    "model_name": "kling-v2-6",
    "prompt": "A skyscraper, starting from the base looking up, glass and steel",
    "duration": "5",
    "mode": "standard",
    "camera_control": {
        "type": "simple",
        "config": {"horizontal"

Kling AI CI Integration

Overview

Automate video generation in CI/CD pipelines. Common use cases: generate product demos on release, create marketing videos from prompts in a YAML file, regression-test video quality across model versions.

GitHub Actions Workflow

# .github/workflows/generate-videos.yml
name: Generate Videos
on:
  workflow_dispatch:
    inputs:
      prompt:
        description: "Video prompt"
        required: true
      model:
        description: "Model version"
        default: "kling-v2-master"

jobs:
  generate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install dependencies
        run: pip install PyJWT requests

      - name: Generate video
        env:
          KLING_ACCESS_KEY: ${{ secrets.KLING_ACCESS_KEY }}
          KLING_SECRET_KEY: ${{ secrets.KLING_SECRET_KEY }}
        run: |
          python3 scripts/generate-video.py \
            --prompt "${{ inputs.prompt }}" \
            --model "${{ inputs.model }}" \
            --output output/

      - name: Upload artifact
        uses: actions/upload-artifact@v4
        with:
          name: generated-video
          path: output/*.mp4
          retention-days: 7

CI Generation Script

#!/usr/bin/env python3
"""scripts/generate-video.py -- CI-friendly video generation."""
import argparse
import jwt
import time
import os
import requests
import sys

BASE = "https://api.klingai.com/v1"

def get_headers():
    ak, sk = os.environ["KLING_ACCESS_KEY"], os.environ["KLING_SECRET_KEY"]
    token = jwt.encode(
        {"iss": ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5},
        sk, algorithm="HS256", headers={"alg": "HS256", "typ": "JWT"}
    )
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--prompt", required=True)
    parser.add_argument("--model", default="kling-v2-master")
    parser.add_argument("--duration", default="5")
    parser.add_argument("--mode", default="standard")
    parser.add_argument("--output", default="output/")
    parser.add_argument("--timeout", type=int, default=600)
    args = parser.parse_args()

    os.makedirs(args.output, exist_ok=True)

    # Submit
    r = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
        "model_name": args.model,
        "prompt": args.prompt,
        "dura

Kling AI Common Errors

Overview

Complete error reference for the Kling AI API. Covers HTTP status codes, task-level failures, JWT issues, and generation-specific problems with tested solutions.

HTTP Error Codes

Task-Level Failures

When HTTP returns 200 but task_status is "failed":

JWT Authentication Errors

Problem: 401 on every request

# WRONG — missing headers parameter
token = jwt.encode(payload, sk, algorit

Kling AI Compliance Review

Overview

Security and compliance assessment framework for Kling AI integrations. Covers data handling, credential management, content policy, privacy, and regulatory considerations.

Data Flow Assessment

User Prompt → [Your App] → [Kling AI API] → [Kling GPU Cluster]
                                                     ↓
[Your CDN] ← download ← [Kling CDN (temporary URL)] ← Generated Video

Data Residency

Security Checklist

Credential Security

• [ ] AK/SK stored in secrets manager (not env files, not code)
• [ ] Keys rotated quarterly
• [ ] Separate keys per environment
• [ ] JWT tokens never logged
• [ ] Access key prefix logged (first 8 chars only)

# Safe logging pattern
def safe_log_key(access_key: str) -> str:
    return access_key[:8] + "..." + access_key[-4:]

Network Security

• [ ] All API calls over HTTPS (enforced by base URL)
• [ ] Webhook endpoints use HTTPS with valid TLS cert
• [ ] Network egress rules allow api.klingai.com:443
• [ ] No API keys in query strings (Bearer token in header only)

Input Validation

• [ ] Prompt length validated (<= 2500 chars)
• [ ] Image URLs validated before sending
• [ ] User input sanitized against injection
• [ ] Content policy pre-filtering active

Output Handling

• [ ] Kling CDN URLs treated as temporary
• [ ] Videos downloaded and stored on your infrastructure
• [ ] Generated content scanned before serving to end users
• [ ] Video metadata stripped of sensitive info before public delivery

Privacy Assessment

GDPR Considerations

class GDPRCompliantClient:
    """Kling client with GDPR data ha

Kling AI Content Policy

Overview

Kling AI enforces content policies server-side. Tasks with policy-violating prompts return task_status: "failed" with a content policy message. This skill covers pre-submission filtering to avoid wasted credits and API calls.

Restricted Content Categories

Kling AI prohibits prompts that generate:

Pre-Submission Prompt Filter

import re

class PromptFilter:
    """Filter prompts before sending to Kling AI to save credits."""

    BLOCKED_PATTERNS = [
        r"\b(nude|naked|explicit|nsfw|porn)\b",
        r"\b(gore|dismember|torture|mutilat)\b",
        r"\b(bomb|terroris|weapon|firearm)\b",
        r"\b(suicide|self.harm|kill.yourself)\b",
        r"\b(deepfake|impersonat)\b",
    ]

    BLOCKED_TERMS = {
        "blood splatter", "graphic violence", "child abuse",
        "drug manufacturing", "hate speech",
    }

    def __init__(self):
        self._patterns = [re.compile(p, re.IGNORECASE) for p in self.BLOCKED_PATTERNS]

    def check(self, prompt: str) -> tuple[bool, str]:
        """Returns (is_safe, reason)."""
        lower = prompt.lower()

        for term in self.BLOCKED_TERMS:
            if term in lower:
                return False, f"Blocked term: '{term}'"

        for pattern in self._patterns:
            match = pattern.search(prompt)
            if match:
                return False, f"Blocked pattern: '{match.group()}'"

        if len(prompt) > 2500:
            return False, "Prompt exceeds 2500 character limit"

        if len(prompt.strip()) < 5:
            return False, "Prompt too short"

        return True, "OK"

    def sanitize(self, prompt: str) -> str:
        """Remove problematic terms and return cleaned prompt."""
        for pattern in self._patterns:
            prompt = pattern.sub(&qu

Kling AI Cost Controls

Overview

Prevent unexpected spending with per-request cost estimation, daily budget enforcement, threshold alerts, and usage dashboards. Credits are consumed per task based on duration, mode, and audio.

Credit Cost Reference

Budget Guard

import time
from dataclasses import dataclass, field

@dataclass
class BudgetGuard:
    """Enforce daily credit budget with alerting."""
    daily_limit: int = 1000
    alert_threshold: float = 0.8  # alert at 80%
    _used: int = 0
    _reset_time: float = field(default_factory=time.time)
    _alerts_sent: set = field(default_factory=set)

    def _check_reset(self):
        if time.time() - self._reset_time > 86400:
            self._used = 0
            self._reset_time = time.time()
            self._alerts_sent.clear()

    def estimate_credits(self, duration: int = 5, mode: str = "standard",
                         audio: bool = False) -> int:
        base = {(5, "standard"): 10, (5, "professional"): 35,
                (10, "standard"): 20, (10, "professional"): 70}
        credits = base.get((duration, mode), 10)
        if audio:
            credits *= 5
        return credits

    def check(self, credits_needed: int) -> bool:
        self._check_reset()

        # Check alert threshold
        usage_pct = (self._used + credits_needed) / self.daily_limit
        if usage_pct >= self.alert_threshold and "80pct" not in self._alerts_sent:
            self._alerts_sent.add("80pct")
            self._on_alert(f"Budget at {usage_pct:.0%} ({self._used + credits_needed}/{self.daily_limit})")

        if self._used + credits_needed > self.daily_limit:
            raise RuntimeError(
                f"Daily budget exceeded: {self._used} + {credits_needed} > {self.daily_limit} credits"
            )
        return True

    def record(self, credits: int):
        self._used += credits

    def _on_alert(self, message: str):
        """Override for custom alerting (Slack, email, PagerDuty)."""
        print(f"ALERT: {message}")

    @property
    def remaining(self) -> int:
        self._check_reset()
        return max(0, self.daily_limit - self._used)

    @property
    def usage_report(self) -> dict:
        self._check_reset()
        return 

Kling AI Debug Bundle

Overview

Structured logging, request tracing, and diagnostic tools for Kling AI API integrations. Captures request/response pairs, task lifecycle events, and timing metrics for every call to https://api.klingai.com/v1.

Debug-Enabled Client

import jwt, time, os, requests, logging, json
from datetime import datetime

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("kling.debug")

class KlingDebugClient:
    """Kling AI client with full request/response logging."""

    BASE = "https://api.klingai.com/v1"

    def __init__(self):
        self.ak = os.environ["KLING_ACCESS_KEY"]
        self.sk = os.environ["KLING_SECRET_KEY"]
        self._request_log = []

    def _get_headers(self):
        token = jwt.encode(
            {"iss": self.ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5},
            self.sk, algorithm="HS256", headers={"alg": "HS256", "typ": "JWT"}
        )
        return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

    def _traced_request(self, method, path, body=None):
        """Execute request with full tracing."""
        url = f"{self.BASE}{path}"
        start = time.monotonic()
        trace = {
            "timestamp": datetime.utcnow().isoformat(),
            "method": method,
            "path": path,
            "request_body": body,
        }

        try:
            if method == "POST":
                r = requests.post(url, headers=self._get_headers(), json=body, timeout=30)
            else:
                r = requests.get(url, headers=self._get_headers(), timeout=30)

            trace["status_code"] = r.status_code
            trace["response_body"] = r.json() if r.content else None
            trace["duration_ms"] = round((time.monotonic() - start) * 1000)

            logger.debug(f"{method} {path} -> {r.status_code} ({trace['duration_ms']}ms)")

            if r.status_code >= 400:
                logger.error(f"API error: {r.status_code} -- {r.text[:300]}")

            r.raise_for_status()
            return r.json()

        except Exception as e:
            trace["error"] = str(e)
            trace["duration_ms"] = round((time.monotonic() - start) * 1000)
            logger.exception(f"Request failed: {path}")
            raise
        finally:
            self._request_log.append(trace)

    def text_to_video(self, prompt, **kwargs):
        body = {
            "model_name": kwargs.get(

Kling AI Hello World

Overview

Generate your first AI video in under 20 lines of code. This skill walks through the complete create-poll-download cycle using the Kling AI REST API.

Base URL: https://api.klingai.com/v1

Prerequisites

• Completed klingai-install-auth setup
• Python 3.8+ with requests and PyJWT
• At least 10 credits in your Kling AI account

Minimal Example — Python

import jwt, time, os, requests

# --- Auth ---
def get_token():
    ak = os.environ["KLING_ACCESS_KEY"]
    sk = os.environ["KLING_SECRET_KEY"]
    payload = {"iss": ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5}
    return jwt.encode(payload, sk, algorithm="HS256",
                      headers={"alg": "HS256", "typ": "JWT"})

BASE = "https://api.klingai.com/v1"
HEADERS = {"Authorization": f"Bearer {get_token()}", "Content-Type": "application/json"}

# --- Step 1: Create task ---
task = requests.post(f"{BASE}/videos/text2video", headers=HEADERS, json={
    "model_name": "kling-v2-master",
    "prompt": "A golden retriever running through autumn leaves in slow motion, cinematic lighting",
    "duration": "5",
    "aspect_ratio": "16:9",
    "mode": "standard",
}).json()

task_id = task["data"]["task_id"]
print(f"Task created: {task_id}")

# --- Step 2: Poll until complete ---
import time as t
while True:
    t.sleep(10)
    status = requests.get(f"{BASE}/videos/text2video/{task_id}", headers=HEADERS).json()
    state = status["data"]["task_status"]
    print(f"Status: {state}")
    if state == "succeed":
        video_url = status["data"]["task_result"]["videos"][0]["url"]
        print(f"Video ready: {video_url}")
        break
    elif state == "failed":
        print(f"Failed: {status['data']['task_status_msg']}")
        break

Minimal Example — Node.js

import jwt from "jsonwebtoken";

const BASE = "https://api.klingai.com/v1";

function getHeaders() {
  const token = jwt.sign(
    { iss: process.env.KLING_ACCESS_KEY, exp: Math.floor(Date.now() / 1000) + 1800,
      nbf: Math.floor(Date.now() / 1000) - 5 },
    process.env.KLING_SECRET_KEY,
    { algorithm: "HS256", header: { typ: "JWT" } }
  );
  return { Authorization: `Bearer ${token}`, "Content-Type": "application/json" };
}

// Create task
const res = await fetch(`${BASE}/videos/text2vi

Kling AI Image-to-Video

Overview

Animate static images using the /v1/videos/image2video endpoint. Supports motion prompts, camera control, dynamic masks (motion brush), static masks, and tail images for start-to-end transitions.

Endpoint: POST https://api.klingai.com/v1/videos/image2video

Request Parameters

Basic Image-to-Video

import jwt, time, os, requests

BASE = "https://api.klingai.com/v1"

def get_headers():
    ak, sk = os.environ["KLING_ACCESS_KEY"], os.environ["KLING_SECRET_KEY"]
    token = jwt.encode(
        {"iss": ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5},
        sk, algorithm="HS256", headers={"alg": "HS256", "typ": "JWT"}
    )
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# Animate a landscape photo
response = requests.post(f"{BASE}/videos/image2video", headers=get_headers(), json={
    "model_name": "kling-v2-1",
    "image": "https://example.com/landscape.jpg",
   

Kling AI Install & Auth

Overview

Kling AI uses JWT (JSON Web Token) authentication. You generate a token from your Access Key (AK) and Secret Key (SK), then pass it as a Bearer token in every request. Tokens expire after 30 minutes.

Base URL: https://api.klingai.com/v1

Prerequisites

• Kling AI account at klingai.com
• API access enabled (self-service, no waitlist)
• Python 3.8+ with PyJWT or Node.js 18+

Step 1 — Get Credentials

1. Sign in at app.klingai.com/global/dev
2. Navigate to API Keys in the developer console
3. Click Create API Key to generate an Access Key + Secret Key pair
4. Store both values securely — the Secret Key is shown only once

# .env file
KLING_ACCESS_KEY="ak_your_access_key_here"
KLING_SECRET_KEY="sk_your_secret_key_here"

Step 2 — Generate JWT Token

Python

import jwt
import time
import os

def generate_kling_token():
    """Generate a JWT token for Kling AI API authentication."""
    ak = os.environ["KLING_ACCESS_KEY"]
    sk = os.environ["KLING_SECRET_KEY"]

    headers = {"alg": "HS256", "typ": "JWT"}
    payload = {
        "iss": ak,
        "exp": int(time.time()) + 1800,  # 30 min expiry
        "nbf": int(time.time()) - 5,     # valid 5s ago (clock skew)
    }

    return jwt.encode(payload, sk, algorithm="HS256", headers=headers)

token = generate_kling_token()
# Use: Authorization: Bearer <token>

Node.js

import jwt from "jsonwebtoken";

function generateKlingToken() {
  const ak = process.env.KLING_ACCESS_KEY;
  const sk = process.env.KLING_SECRET_KEY;

  const payload = {
    iss: ak,
    exp: Math.floor(Date.now() / 1000) + 1800,
    nbf: Math.floor(Date.now() / 1000) - 5,
  };

  return jwt.sign(payload, sk, { algorithm: "HS256", header: { typ: "JWT" } });
}

Step 3 — Verify Authentication

import requests

BASE_URL = "https://api.klingai.com/v1"
token = generate_kling_token()

response = requests.get(
    f"{BASE_URL}/videos/text2video",  # any endpoint to test auth
    headers={"Authorization": f"Bearer {token}"},
)

if response.status_code == 401:
    print("Auth failed — check AK/SK values")
elif response.status_code in (200, 400):
    print("Auth working — credentials valid")

Token Management Pattern

import time

class KlingAuth:
    "&quo

Kling AI Job Monitoring

Overview

Every Kling AI generation returns a task_id. This skill covers polling strategies, batch tracking, timeout handling, and callback-based monitoring for the /v1/videos/text2video, /v1/videos/image2video, and /v1/videos/video-extend endpoints.

Task Lifecycle

Polling a Single Task

import jwt, time, os, requests

BASE = "https://api.klingai.com/v1"

def get_headers():
    ak, sk = os.environ["KLING_ACCESS_KEY"], os.environ["KLING_SECRET_KEY"]
    token = jwt.encode(
        {"iss": ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5},
        sk, algorithm="HS256", headers={"alg": "HS256", "typ": "JWT"}
    )
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

def poll_task(endpoint: str, task_id: str, interval: int = 10, timeout: int = 600):
    """Poll with adaptive interval and timeout."""
    start = time.monotonic()
    attempts = 0
    while time.monotonic() - start < timeout:
        time.sleep(interval)
        attempts += 1
        r = requests.get(f"{BASE}{endpoint}/{task_id}", headers=get_headers(), timeout=30)
        data = r.json()["data"]
        status = data["task_status"]
        elapsed = int(time.monotonic() - start)
        print(f"[{elapsed}s] Poll #{attempts}: {status}")

        if status == "succeed":
            return data["task_result"]
        elif status == "failed":
            raise RuntimeError(f"Task failed: {data.get('task_status_msg', 'unknown')}")

        if attempts > 5:
            interval = min(interval * 1.2, 30)
    raise TimeoutError(f"Task {task_id} timed out after {timeout}s")

Batch Job Tracker

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class TrackedTask:
    task_id: str
    endpoint: str
    prompt: str
    status: str = "submitted"
    created_at: float = field(default_factory=time.time)
    result_url: Optional[str] = None
    error_msg: Optional[str] = None

class BatchTracker:

Kling AI Known Pitfalls

Overview

Documented mistakes, gotchas, and anti-patterns from real Kling AI integrations. Each pitfall includes the symptom, root cause, and tested fix.

Pitfall 1: Duration as Integer

Symptom: 400 Bad Request on valid-looking requests.

# WRONG -- duration as integer
{"duration": 5}

# CORRECT -- duration as string
{"duration": "5"}

The API requires duration as a string "5" or "10", not an integer.

Pitfall 2: JWT Without Explicit Headers

Symptom: 401 Unauthorized even with correct AK/SK.

# WRONG -- missing headers parameter
token = jwt.encode(payload, sk, algorithm="HS256")

# CORRECT -- explicit JWT headers
token = jwt.encode(payload, sk, algorithm="HS256",
                   headers={"alg": "HS256", "typ": "JWT"})

Some JWT libraries don't include typ: "JWT" by default. Kling requires it.

Pitfall 3: Token Generated Once at Import Time

Symptom: Works for 30 minutes, then all requests fail with 401.

# WRONG -- token generated once
TOKEN = generate_token()  # at module import
headers = {"Authorization": f"Bearer {TOKEN}"}

# CORRECT -- generate fresh token per request (or auto-refresh)
def get_headers():
    return {"Authorization": f"Bearer {generate_token()}"}

JWT tokens expire after 30 minutes. Always implement auto-refresh.

Pitfall 4: Polling Without Timeout

Symptom: Script hangs forever on a failed task.

# WRONG -- infinite loop
while True:
    result = check_status(task_id)
    if result["status"] == "succeed":
        break
    time.sleep(10)

# CORRECT -- with timeout and failure check
start = time.monotonic()
while time.monotonic() - start < 600:  # 10 min max
    result = check_status(task_id)
    if result["status"] == "succeed":
        break
    elif result["status"] == "failed":
        raise RuntimeError(result["error"])
    time.sleep(10)
else:
    raise TimeoutError("Generation timed out")

Pitfall 5: Not Downloading Videos Promptly

Symptom: Video URLs return 404 or 403 after a day.

Kling CDN URLs are temporary (24-72 hours). Always download and store on your own infrastructure immediately after generation completes.

# WRONG -- storing only the Kling URL
db.save(video_url=kling_cdn_url)  # will expire

# CORRECT -- download and rehost
local_path = d

Kling AI Model Catalog

Overview

Kling AI offers multiple model versions across video generation, image generation, lip sync, virtual try-on, and effects. Each version trades off quality, speed, and cost. This skill is the reference for choosing the right model.

Video Generation Models

T2V = text-to-video, I2V = image-to-video

Kling v2.5 Turbo (Recommended for Speed)

• 40% faster than v2.0
• Up to 1080p at 30 FPS
• Best cost/quality ratio for production pipelines

Kling v2.6 (Recommended for Quality)

• Native audio generation (voice, SFX, ambient in one pass)
• 1080p at 30-48 FPS
• Set motionhasaudio: true for synchronized audio

Image Generation Models (Kolors)

Specialty Models

import time, requests, json

def benchmark_model(prompt: str, model: str, mode: str = "standard",
                    runs: int = 3) -> dict:
    """Benchmark generation time for a model/mode combination."""
    times = []

    for i in range(runs):
        start = time.monotonic()

        # Submit
        r = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
            "model_name": model, "prompt": prompt, "duration": "5", "mode": mode,
        }).json()
        task_id = r["data"]["task_id"]

        # Poll
        while True:
            time.sleep(10)
            result = requests.get(
                f"{BASE}/videos/text2video/{task_id}", headers=get_headers()
            ).json()
            if result["data"]["task_status"] in ("succeed", "failed"):
                break

        elapsed = time.monotonic() - start
        times.append(elapsed)
        print(f"  Run {i+1}/{runs}: {elapsed:.1f}s ({result['data']['task_status']})")

    return {
        "model": model,
        "mode": mode,
        "avg_sec": round(sum(times) / len(times), 1),
        "min_sec": round(min(times), 1),
        "max_sec": round(max(times), 1),
        "runs": runs,
    }

# Compare models
prompt = "A waterfall in a tropical forest, cinematic"
for model in ["kling-v2-5-turbo", "kling-v2-master", "kling-v2-6"]:
    result = benchmark_model(prompt, model, runs=2)
    print(f"{model}: avg={result['avg_sec']}s, min={result['min_sec']}s")

import requests

# Without pooling: new TCP connection per request (slow)
# With poolin

def estimate_cost(videos: int, duration: int = 5, mode: str = "standard",
                  audio: bool = False) -> dict:
    """Estimate credits needed for a batch of videos."""
    base_credits = {
        (5, "standard"): 10,
        (5, "professional"): 35,
        (10, "standard"): 20,
        (10, "professional"): 70,
    }
    per_video

# Pre-batch credit check
credits_needed = len(prompts) * 10  # 10 credits per 5s standard
if credits_needed > DAILY_BUDGET:
    raise RuntimeError(f"Batch needs {credits_needed}, budget is {DAILY_BUDGET}")

import time, random, requests

def exponential_backoff(attempt: int, base: float = 1.0, max_wait: float = 60.0) -> float:
    """Calculate wait time with jitter to avoid thundering herd."""
    wait = min(base * (2 ** attempt), max_wait)
    jitter = random.uniform(0, wait * 0.5)
    return wait + jitter

def request_with_retry(method, url, headers, json=None, max_retries=5):
    for attempt in range(max_retries + 1):
        response = method(url, headers=headers, json=json, timeout=30)

        if response.status_code == 429:
            if attempt == max_retries:
                raise RuntimeError("Rate limit: max retries exceeded")
            wait = exponential_backoff(attempt)
            print(f"429 rate limited. Waiting {wait:.1f}s (attempt {attempt + 1})")
            time.sleep(wait)
            continue

        if response.status_code >= 500:
            if attempt == max_retries:
                response.raise_for_status()
            time.sleep(exponential_backoff(attempt, base=2.0))
            continue

        response.raise_for_status()
        return response

    raise RuntimeError("Unreachable")

import asyncio

class TaskLimiter:
    """Limit concurrent Kling AI tasks to stay within API tier."""

    def __init__(self, max_concurrent: int = 3):
        self._semaphore = asyncio.Semaphore(max_concurrent)
        self._active = 0

    async def submit(self, coro):
        async with self._semaphore:
            self._active += 1
            try:
                return await coro
            finally:
                self._active -= 1

    @property
    def active_count(self) -> int:
        return self._active

# Usage
limiter = TaskLimiter(max_concurrent=3)
tasks = [limiter.submit(generate_video(p)) for p in prompts]
results = await asyncio.gather(*tasks, return_exceptions=True)

class RateLimitMonitor:
    """Track API call frequency and warn before hitting limits.""&

User Request
    |
[API Gateway / Load Balancer]
    |
[Application Server]
    |--- validate prompt & estimate cost
    |--- enqueue job to Redis/SQS
    |
[Job Queue (Redis / SQS / Pub/Sub)]
    |
[Worker Pool (N workers)]
    |--- generate JWT token
    |--- POST https://api.klingai.com/v1/videos/text2video
    |--- receive task_id
    |--- register callback_url OR poll
    |
[Webhook Receiver / Poller]
    |--- receive completion callback
    |--- download video from Kling CDN
    |--- upload to S3/GCS
    |--- update job status in DB
    |--- notify user
    |
[Object Storage (S3 / GCS)]
    |
[CDN (CloudFront / Cloud CDN)]
    |
User views video

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class VideoRequest(BaseModel):
    prompt: str
    model: str = "kling-v2-master"
    duration: int = 5
    mode: str = "standard"

@app.post("/api/videos")
async def create_video(req: VideoRequest):
    # 1. Validate
    if len(req.prompt) > 2500:
        raise HTTPException(400, "Prompt exceeds 2500 chars")

    # 2. Estimate cost
    credits = estimate_credits(req.duration, req.mode)
    if not budget_guard.check(credits):
        raise HTTPException(402, "Budget exceeded")

    # 3. Enqueue
    job_id = await queue.enqueue({
        "prompt": req.prompt,
        "model": req.model,
        "duration": str(req.duration),
        "mode": req.mode,
    })

    return {"job_id": job_id, "status": "queued", "estimated_credits": credits}

import redis
import json

class VideoWorker:
    def __init__(self, kling_client, storage_client, redis_url="redis://localhost"):
        self.kling = kling_client
        self.storage = storage_client
        self.redis = redis.Redis.from_url(redis_url)

    def process_loop(self):
        while True:
            raw = self.redis.brpop("kling:jobs:pending", timeout=5)
            if not raw:
                continue

            job = json.loads(raw[1])
            try:
                # Submit to Kling API
                result = self.kling.text_to_video(
                    job["prompt"],
                    model=job["model"],
                    duration=int(job["duration"]),
                    mode=job["mode"],
                    callback_url=os.environ.get("WEBHOOK_URL"),
                )

                # If using polling (no callback)
                

import jwt
import time
import os
import requests
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class KlingConfig:
    access_key: str = field(default_factory=lambda: os.environ["KLING_ACCESS_KEY"])
    secret_key: str = field(default_factory=lambda: os.environ["KLING_SECRET_KEY"])
    base_url: str = "https://api.klingai.com/v1"
    token_buffer_sec: int = 300
    poll_interval_sec: int = 10
    max_poll_attempts: int = 120  # 20 minutes max
    timeout_sec: int = 30

class KlingClient:
    """Production Kling AI client with auto-refreshing JWT."""

    def __init__(self, config: Optional[KlingConfig] = None):
        self.config = config or KlingConfig()
        self._token = None
        self._token_expires = 0

    @property
    def _headers(self) -> dict:
        now = int(time.time())
        if now >= (self._token_expires - self.config.token_buffer_sec):
            payload = {"iss": self.config.access_key, "exp": now + 1800, "nbf": now - 5}
            self._token = jwt.encode(payload, self.config.secret_key,
                                     algorithm="HS256",
                                     headers={"alg": "HS256", "typ": "JWT"})
            self._token_expires = now + 1800
        return {"Authorization": f"Bearer {self._token}",
                "Content-Type": "application/json"}

    def _post(self, path: str, body: dict) -> dict:
        r = requests.post(f"{self.config.base_url}{path}",
                          headers=self._headers, json=body,
                          timeout=self.config.timeout_sec)
        r.raise_for_status()
        return r.json()

    def _get(self, path: str) -> dict:
        r = requests.get(f"{self.config.base_url}{path}",
                         headers=self._headers,
                         timeout=self.config.timeout_sec)
        r.raise_for_status()
        return r.json()

    def _poll_task(self, endpoint: str, task_id: str) -> dict:
        """Poll with exponential backoff until task completes."""
        interval = self.config.poll_interval_sec
        for attempt in range(self.config.max_poll_attempts):
            time.sleep(interval)
            result = self._get(f"{endpoint}/{task_id}")
            status = result["data"]["task_status"]
            if status == "succeed":
                return result["data"]["task_result"]
            el

import requests
import os

def download_video(video_url: str, output_dir: str = "output") -> str:
    """Download generated video from Kling CDN."""
    os.makedirs(output_dir, exist_ok=True)

    # Extract filename or generate one
    filename = video_url.split("/")[-1].split("?")[0]
    if not filename.endswith(".mp4"):
        filename = f"kling_{int(time.time())}.mp4"

    filepath = os.path.join(output_dir, filename)
    response = requests.get(video_url, stream=True, timeout=120)
    response.raise_for_status()

    with open(filepath, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)

    size_mb = os.path.getsize(filepath) / (1024 * 1024)
    print(f"Downloaded: {filepath} ({size_mb:.1f} MB)")
    return filepath

import boto3

def upload_to_s3(filepath: str, bucket: str, key_prefix: str = "kling-videos/") -> str:
    """Upload video to S3 and return public URL."""
    s3 = boto3.client("s3")
    filename = os.path.basename(filepath)
    s3_key = f"{key_prefix}{filename}"

    s3.upload_file(
        filepath, bucket, s3_key,
        ExtraArgs={"ContentType": "video/mp4", "CacheControl": "max-age=86400"}
    )

    url = f"https://{bucket}.s3.amazonaws.com/{s3_key}"
    print(f"Uploaded to S3: {url}")
    return url

# Generate signed URL for private buckets
def get_signed_url(bucket: str, key: str, expiry: int = 3600) -> str:
    s3 = boto3.client("s3")
    return s3.generate_presigned_url(
        "get_object",
        Params={"Bucket": bucket, "Key": key},
        ExpiresIn=expiry,
    )

from google.cloud import storage

def upload_to_gcs(filepath: str, bucket_name: str, prefix: str = "kling-videos/") -> str:
    """Upload video to GCS and return public URL."""
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    filename = os.path.basename(filepath)
    blob = bucket.blob(f"{prefix}{filename}")

    blob.upload_from_filename(filepath, content_type="video/mp4")
    blob.make_public()  # or use signed URLs for private access

    print(f"Uploaded to GCS: {blob.public_url}")
    return blob.public_url

# Signed URL for private access
def get_gcs_signed_url(buck

import jwt, time, os, requests

BASE = "https://api.klingai.com/v1"

def get_headers():
    ak, sk = os.environ["KLING_ACCESS_KEY"], os.environ["KLING_SECRET_KEY"]
    token = jwt.encode(
        {"iss": ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5},
        sk, algorithm="HS256", headers={"alg": "HS256", "typ": "JWT"}
    )
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# Style: Studio Ghibli watercolor
response = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
    "model_name": "kling-v2-6",
    "prompt": "A cozy cottage in a meadow, hand-painted watercolor style, "
              "soft pastel colors, Studio Ghibli aesthetic, gentle breeze",
    "negative_prompt": "photorealistic, harsh lighting, dark, gritty",
    "duration": "5",
    "mode": "professional",
    "cfg_scale": 0.7,  # higher = stricter prompt adherence
})

# Apply an effect to an image
response = requests.post(f"{BASE}/videos/effects", he

# .env.development
KLING_ACCESS_KEY="ak_dev_..."
KLING_SECRET_KEY="sk_dev_..."

# .env.production
KLING_ACCESS_KEY="ak_prod_..."
KLING_SECRET_KEY="sk_prod_..."

from dataclasses import dataclass
from typing import Optional

@dataclass
class TeamMember:
    name: str
    email: str
    role: str  # admin, editor, viewer
    daily_credit_limit: int
    allowed_models: list[str]

@dataclass
class TeamConfig:
    name: str
    members: list[TeamMember]
    total_daily_limit: int = 1000
    default_model: str = "kling-v2-master"
    default_mode: str = "standard"

    def get_member(self, email: str) -> Optional[TeamMember]:
        return next((m for m in self.members if m.email == email), None)

# Example team configuration
team = TeamConfig(
    name="marketing",
    total_daily_limit=5000,
    members=[
        TeamMember("Alice", "alice@co.com", "admin", 2000,
                   ["kling-v2-6", "kling-v2-master", "kling-v2-5-turbo"]),
        TeamMember("Bob", "bob@co.com", "editor", 500,
                   ["kling-v2-master", "kling-v2-5-turbo"]),
        TeamMember("Carol", "carol@co.com", "viewer", 100,
                   ["kling-v2-5-turbo"]),
    ],
)

import time
from collections import defaultdict

class TeamQuotaManager:
    """Enforce per-member and team-wide credit limits."""

    def __init__(self, config: TeamConfig):
        self.config = config
        self._usage = defaultdict(int)  # email -> credits used today
        self._reset_time = time.time()

    def _check_reset(self):
        if time.time() - self._reset_time > 86400:
            self._usage.clear()
            self._reset_time = time.time()

    def authorize(self, email: str, credits_needed: int, model: str) -> bool:
        self._check_rese

import jwt, time, os, requests

BASE = "https://api.klingai.com/v1"

def get_headers():
    ak, sk = os.environ["KLING_ACCESS_KEY"], os.environ["KLING_SECRET_KEY"]
    token = jwt.encode(
        {"iss": ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5},
        sk, algorithm="HS256", headers={"alg": "HS256", "typ": "JWT"}
    )
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# Create text-to-video task
response = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
    "model_name": "kling-v2-6",
    "prompt": "Aerial drone shot of a coral reef at golden hour, "
              "tropical fish swimming through crystal clear water, "
              "sun rays penetrating the surface, cinematic 4K",
    "negative_prompt": "blurry, low quality, distorted, watermark",
    "duration": "5",
    "aspect_ratio": "16:9",
    "mode": "professional",
    "cfg_scale": 0.5,
})

task = response.json()
task_id = t

# v1.x request
body = {
    "model_name": "kling-v1-6",
    "prompt": "A sunset over mountains",
    "duration": "5",
    "mode": "standard",
}

# v2.x -- only model_name changes
body["model_name"] = "kling-v2-master"

body["model_name"] = "kling-v2-6"
body["motion_has_audio"] = True  # NEW: synchronized audio

# Cost impact: audio multiplies credits 5x
# 5s standard: 10 -> 50 credits

import json
import time
from datetime import datetime
from pathlib import Path

class KlingEventLogger:
    """Append-only JSONL event log for Kling AI operations."""

    def __init__(self, log_dir: str = "logs"):
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(exist_ok=True)

    def _write(self, event: dict):
        date = datetime.utcnow().strftime("%Y-%m-%d")
        filepath = self.log_dir / f"kling-{date}.jsonl"
        event["timestamp"] = datetime.utcnow().isoformat()
        with open(filepath, "a") as f:
            f.write(json.dumps(event) + "\n")

    def log_submission(self, task_id, prompt, model, duration, mode):
        self._write({
            "event": "task_submitted",
            "task_id": task_id,
            "model": model,
            "duration": int(duration),
            "mode": mode,
            "prompt_len": len(prompt),
        })

    def log_completion(self, task_id, status, elapsed_sec, credits_used):
        self._write({
            "event": "task_completed",
            "task_id": task_id,
            "status": status,
            "elapsed_sec": elapsed_sec,
            "credits_used": credits_used,
        })

    def log_error(self, task_id, error_type, message):
        self._write({
            "event": "task_error",
            "task_id": task_id,
            "error_type": error_type,
            "message": message[:200],
        })

from collections import defaultdict

class UsageAnalytics:
    """Aggregate metrics from JSONL event logs."""

    def __init__(self, log_dir: str = "logs"):
        self.log_dir = Path(log_dir)

    def _read_events(self, date: str = None):
        pattern = f"kling-{date}.jsonl" if date else "kling-*.jsonl"
        events = []
        for filepath in sorted(self.log_dir.glob(pattern)):
            with open(filepath) as f:
                for line in f:
                    events.append(json.loads(line))
        return events

    def daily_summary(self, date: str = None) -> dict:
        date = date or datetime.utcnow().strftime("%Y-%m-%d")
        events = self._read_events(date)

        submitted = [e for e in events if e["event"] == "task_submitted"]
        completed = [e for e in events if e["event"] == "task_completed"

import jwt, time, os, requests

BASE = "https://api.klingai.com/v1"

def get_headers():
    ak, sk = os.environ["KLING_ACCESS_KEY"], os.environ["KLING_SECRET_KEY"]
    token = jwt.encode(
        {"iss": ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5},
        sk, algorithm="HS256", headers={"alg": "HS256", "typ": "JWT"}
    )
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# Step 1: Generate the initial 5s video
initial = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
    "model_name": "kling-v2-master",
    "prompt": "A rocket launching from a desert landscape, cinematic",
    "duration": "5",
    "mode": "standard",
}).json()
initial_task_id = initial["data"]["task_id"]

# Wait for completion...
# (poll until task_status == "succeed")

# Step 2: Extend by 5 more seconds
extension = requests.post(f"{BASE}/videos/video-extend", headers=get_headers(), json={
    "task_id": initial_task_id,
    "prompt": "The rocket ascends through clouds into the stratosphere",
    "duration": "5",
    "mode": "standard",
}).json()
ext_task_id = extension["data"]["task_id"]

# Step 3: Poll extension task
while True:
    time.sleep(15)
    result = requests.get(
        f"{BASE}/videos/video-extend/{ext_task_id}", headers=get_headers()
    ).json()
    if result["data"][

import jwt, time, os, requests

BASE = "https://api.klingai.com/v1"

def get_headers():
    ak, sk = os.environ["KLING_ACCESS_KEY"], os.environ["KLING_SECRET_KEY"]
    token = jwt.encode(
        {"iss": ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5},
        sk, algorithm="HS256", headers={"alg": "HS256", "typ": "JWT"}
    )
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# Create task with callback
response = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
    "model_name": "kling-v2-master",
    "prompt": "A futuristic city skyline at night with neon lights",
    "duration": "5",
    "mode": "standard",
    "callback_url": "https://your-app.com/webhooks/kling",  # your endpoint
})

task_id = response.json()["data"]["task_id"]
print(f"Task {task_id} submitted with callback -- no polling needed")

from flask import Flask, request, jsonify
import hmac
import hashlib
import json

app = Flask(__name__)

@app.route("/webhooks/kling", methods=["POST"])
def kling_webhook():
    payload = request.get_json()

    task_id = payload["data"]["task_id"]
    status = payload["data"]["task_status"]

    if status == "succeed":
        video_url = payload["data"]["task_result"]["videos"][0]["url"]
        print(f"Task {task_id} complete: {video_url}")
        # Download video, store to S3, notify user, etc.
        process_completed_video(task_id, video_url)
    elif status == "failed":
        error = payload["data"].get("task_status_msg", "Unknown error")
        print(f"Task {task_id} failed: {

export KLING_ACCESS_KEY="ak_your_key"
export KLING_SECRET_KEY="sk_your_key"