Kling AI Async Workflows

Overview

Kling AI video generation is inherently async: you submit a task, then poll or receive a callback when done. This skill covers production patterns for integrating this into larger systems using queues, state machines, and event-driven architectures.

Core Pattern: Submit + Callback

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 submit_async(prompt, callback_url=None, **kwargs):
    """Submit task and return immediately."""
    body = {
        "model_name": kwargs.get("model", "kling-v2-master"),
        "prompt": prompt,
        "duration": str(kwargs.get("duration", 5)),
        "mode": kwargs.get("mode", "standard"),
    }
    if callback_url:
        body["callback_url"] = callback_url

    r = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json=body)
    return r.json()["data"]["task_id"]

Redis Queue Workflow

import redis
import json

r = redis.Redis()

# Producer: enqueue video generation requests
def enqueue_video_job(prompt, metadata=None):
    job = {
        "id": f"job_{int(time.time() * 1000)}",
        "prompt": prompt,
        "metadata": metadata or {},
        "status": "queued",
        "created_at": time.time(),
    }
    r.lpush("kling:jobs:pending", json.dumps(job))
    return job["id"]

# Worker: process jobs from queue
def process_jobs(max_concurrent=3):
    active_tasks = {}

    while True:
        # Submit new jobs if under concurrency limit
        while len(active_tasks) < max_concurrent:
            raw = r.rpop("kling:jobs:pending")
            if not raw:
                break
            job = json.loads(raw)
            task_id = submit_async(job["prompt"])
            active_tasks[task_id] = job
            r.hset("kling:jobs:active", task_id, json.dumps(job))

        # Check active tasks
        completed = []
        for task_id, job in active_tasks.items():
            result = requests.get(
                f"{BASE}/videos/text2video/{task_id}", headers=get_headers()
            ).json()
            status = resul…

Kling AI Audit Logging

Overview

Compliance-grade audit logging for Kling AI API operations. Every task submission, status change, and credential usage is captured in tamper-evident structured logs.

Audit Event Schema

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

class AuditLogger:
    """Append-only audit log with integrity checksums."""

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

    def _compute_hash(self, entry: dict) -> str:
        raw = json.dumps(entry, sort_keys=True) + self._prev_hash
        return hashlib.sha256(raw.encode()).hexdigest()[:16]

    def log(self, event_type: str, actor: str, details: dict):
        """Write a tamper-evident audit entry."""
        entry = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "event_type": event_type,
            "actor": actor,
            "details": details,
            "prev_hash": self._prev_hash,
        }
        entry["hash"] = self._compute_hash(entry)
        self._prev_hash = entry["hash"]

        date = datetime.utcnow().strftime("%Y-%m-%d")
        filepath = self.log_dir / f"audit-{date}.jsonl"
        with open(filepath, "a") as f:
            f.write(json.dumps(entry) + "\n")

        return entry["hash"]

Audit Events for Kling AI

class KlingAuditClient:
    """Kling client with full audit trail."""

    def __init__(self, base_client, audit: AuditLogger, actor: str = "system"):
        self.client = base_client
        self.audit = audit
        self.actor = actor

    def text_to_video(self, prompt: str, **kwargs):
        # Log submission
        self.audit.log("task_submitted", self.actor, {
            "action": "text_to_video",
            "model": kwargs.get("model", "kling-v2-master"),
            "duration": kwargs.get("duration", 5),
            "mode": kwargs.get("mode", "standard"),
            "prompt_hash": hashlib.sha256(prompt.encode()).hexdigest()[:16],
            "prompt_length": len(prompt),
        })

        result = self.client.text_to_video(prompt, **kwargs)

        # Log completion
        self.audit.log("task_completed", self.actor, {
            "action": "text_to_video",
            "status": "succeed",
            "video_count": len(result.get("videos", [])),
        })

        …

Kling AI Batch Processing

Overview

Generate multiple videos efficiently using controlled parallelism, rate-limit-aware submission, progress tracking, and result collection. All requests go through https://api.klingai.com/v1.

Batch Submission with Rate Limiting

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 submit_batch(prompts, model="kling-v2-master", duration="5",
                 mode="standard", max_concurrent=3, delay=2.0):
    """Submit batch with controlled concurrency and pacing."""
    tasks = []
    active = []

    for i, prompt in enumerate(prompts):
        # Wait if at concurrency limit
        while len(active) >= max_concurrent:
            active = [t for t in active if not check_complete(t["task_id"])]
            if len(active) >= max_concurrent:
                time.sleep(5)

        response = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
            "model_name": model,
            "prompt": prompt,
            "duration": duration,
            "mode": mode,
        })
        data = response.json()["data"]
        task = {"task_id": data["task_id"], "prompt": prompt, "index": i}
        tasks.append(task)
        active.append(task)
        print(f"[{i+1}/{len(prompts)}] Submitted: {data['task_id']}")
        time.sleep(delay)  # pace requests

    return tasks

def check_complete(task_id):
    r = requests.get(f"{BASE}/videos/text2video/{task_id}", headers=get_headers()).json()
    return r["data"]["task_status"] in ("succeed", "failed")

Collect Results

def collect_results(tasks, timeout=600):
    """Wait for all tasks and collect results."""
    results = {}
    start = time.monotonic()

    while len(results) < len(tasks) and time.monotonic() - start < timeout:
        for task in tasks:
            if task["task_id"] in results:
                continue
            r = requests.get(
                f"{BASE}/videos/text2video/{task['task_id']}", headers=get_headers()
            ).json()
            status = r["data…

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":…

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,
    …

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(…

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 …

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:
            …

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()
 …

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…

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}/…

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",
    …

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 KlingAu…

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 …

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
…

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 motion_has_audio: true for synchronized audio

Image Generation Models (Kolors)

Specialty Models

Kling AI Performance Tuning

Overview

Optimize video generation for your use case by choosing the right model, mode, and parameters. Covers benchmarking, speed vs. quality trade-offs, connection pooling, and caching strategies.

Speed vs. Quality Matrix

Benchmarking Tool

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")

Connection Pooling

import requests

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

Kling AI Pricing Basics

Overview

Kling AI uses a credit-based pricing system. Credits are consumed per video/image generation based on duration, mode, and model. API pricing uses resource packs billed separately from subscription plans.

Subscription Plans (Web UI)

Warning: Paid credits expire at end of billing period. Unused credits do not roll over.

Video Generation Costs

With Native Audio (v2.6)

Image Generation Costs (Kolors)

API Resource Packs

API access is billed separately from subscriptions via prepaid packs:

1 unit = 1 credit equivalent. API pricing works out to ~$0.07-0.14 per second of generated video.

Cost Estimation

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,
    }…

Kling AI Production Checklist

Overview

Checklist covering authentication, error handling, cost controls, monitoring, security, and content policy before deploying Kling AI video generation to production.

Authentication

• [ ] AK/SK stored in secrets manager (not .env in repo)
• [ ] JWT auto-refresh with 5-min buffer before 30-min expiry
• [ ] Separate API keys per environment (dev/staging/prod)
• [ ] Key rotation schedule (quarterly minimum)
• [ ] Authorization: Bearer <token> format verified

Error Handling

• [ ] HTTP 400/401/402/403/429/5xx all handled
• [ ] Exponential backoff with jitter for 429/5xx retries
• [ ] Max retry limit set (3-5, not infinite)
• [ ] task_status: "failed" logs task_status_msg
• [ ] 30s timeout on all HTTP calls
• [ ] duration sent as string "5" not integer 5

Cost Controls

• [ ] Daily credit budget enforced in code
• [ ] Alert at 80% daily budget consumption
• [ ] standard mode used for non-final renders
• [ ] Max poll attempts cap (no infinite loops)
• [ ] Credit estimate before batch submission

# 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}")

Task Management

• [ ] All task_ids logged with timestamp
• [ ] Stuck task detection (>10 min in processing)
• [ ] callback_url used instead of polling in production
• [ ] Failed tasks queued for retry
• [ ] Video URLs downloaded promptly (Kling CDN URLs expire)

Content Safety

• [ ] Prompts validated before API submission
• [ ] User-generated prompts sanitized
• [ ] Default negative prompt includes safety terms
• [ ] Content moderation on user-facing apps
• [ ] Policy violation errors handled gracefully

Security

• [ ] API keys never logged (redacted in debug output)
• [ ] Video URLs treated as temporary (store on own CDN)
• [ ] Webhook endpoints HTTPS-only
• [ ] Rate limiting on your API layer
• [ ] No sensitive data in prompt strings

Monitoring

• [ ] API latency tracked per endpoint
• [ ] Success/failure rate dashboard
• [ ] Credit consumption metrics
• [ ] Alert on >5% failure rate
• [ ] Structured JSON logs for all API calls

Performance

• [ ] Connection pooling via requests.Session()
• [ ] Concurrent tasks within API tier limit
• [ ] Video downloads async/back…

Kling AI Rate Limits

Overview

Kling AI enforces rate limits per API key. When exceeded, the API returns 429 Too Many Requests. This skill covers detection, backoff strategies, request queuing, and concurrent job management.

Rate Limit Tiers

Exponential Backoff with Jitter

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")

Concurrent Task Limiter (asyncio)

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)

Rate Limit Monitor

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

Kling AI Reference Architecture

Overview

Production architecture for video generation platforms built on Kling AI. Covers API gateway, job queue, worker pool, storage, and monitoring layers.

Architecture Diagram

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

Component Details

API Layer

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}

Worker Service

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)
  …

Kling AI SDK Patterns

Overview

Production-ready client patterns for the Kling AI API. Covers auto-refreshing JWT, typed request/response models, exponential backoff polling, async batch submission, and structured error handling.

Python Client Wrapper

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"]
…

Kling AI Storage Integration

Overview

Kling AI video URLs from task_result.videos[].url are temporary CDN links that expire. You must download and store videos in your own storage. This skill covers S3, GCS, and Azure Blob.

Download from Kling CDN

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

Upload to AWS S3

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,
    )

Upload to Google Cloud Storage

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_s…

Kling AI Style Transfer & Effects

Overview

Apply artistic styles through prompt engineering, use the Effects API for pre-built visual transformations, and leverage Kolors for image-based style references. Available on v1.6+ models.

Style via Prompt Engineering

The most direct approach -- include style descriptors in your prompt:

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
})

Style Prompt Recipes

Effects API

The Effects API applies pre-built transformations to existing images. Available on v1.6+.

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

…

Kling AI Team Setup

Overview

Manage team access to the Kling AI API using separate API keys, environment-based routing, usage quotas per team member, and centralized credential management.

Per-Environment API Keys

Create separate API key pairs in the Kling AI developer console for each environment:

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

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

Team Configuration

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"]),
    ],
)

Usage Quotas Per Member

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) -…

Kling AI Text-to-Video

Overview

Generate videos from text prompts using the /v1/videos/text2video endpoint. Supports models v1 through v2.6, standard/professional modes, camera control, negative prompts, and native audio (v2.6+).

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

Request Parameters

Complete Example — Python

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": "profes…

Kling AI Upgrade & Migration

Overview

Guide for migrating between Kling AI model versions. Covers breaking changes, parameter differences, feature availability, and parallel testing strategies.

Version History

Migration: v1.x to v2.x

# 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"

Breaking changes:

• kling-v2-1 is I2V-only (no text-to-video support)
• Camera control intensities produce different results at same values
• Generation times differ (v2.x generally slower, higher quality)

Migration: v2.x to v2.6 with Audio

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

Feature Availability Matrix

Kling AI Usage Analytics

Overview

Track video generation usage with structured logging, aggregate metrics, daily reports, and cost analysis. Built on JSONL event logs that can feed into any analytics platform.

Event Logger

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],
        })

Analytics Aggregator

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_c…

Kling AI Video Extension

Overview

Extend an existing video by appending additional seconds. The extension endpoint takes the task_id of a completed video and generates a seamless continuation.

Endpoint: POST https://api.klingai.com/v1/videos/video-extend

Request Parameters

Basic Extension

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_h…

Kling AI Webhook Configuration

Overview

Instead of polling task status, pass a callback_url when creating a task. Kling AI will POST the completed task result to your URL when generation finishes. This eliminates polling overhead and reduces API calls.

Supported on: All video generation endpoints (text2video, image2video, video-extend, lip-sync, effects)

How It Works

1. Include callback_url in your task creation request
2. Kling queues the task normally
3. When task reaches terminal state (succeed or failed), Kling POSTs the full result to your URL
4. Your webhook handler processes the result

Sending a Task with Callback

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")

Webhook Receiver (Flask)

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…