Complete Kling AI integration skill pack with 30 skills covering AI video generation, text-to-video, image-to-video, and creative workflows. Flagship+ tier vendor pack.
Installation
Open Claude Code and run this command:
/plugin install klingai-pack@claude-code-plugins-plus
Use --global to install for all projects, or --project for current project only.
Skills (30)
Build async video generation workflows with Kling AI using queues, state machines, and event-driven patterns.
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 = result["data&qImplement audit logging for Kling AI operations for compliance and security.
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", [])),
})
return result
Process multiple video generation requests efficiently with Kling AI.
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"]["tControl camera movements in Kling AI video generation.
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&quoIntegrate Kling AI video generation into CI/CD pipelines.
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,
"duraDiagnose and fix common Kling AI API errors.
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
| Code | Error | Cause | Solution |
|---|---|---|---|
400 |
Bad Request | Invalid parameters, malformed JSON | Validate all required fields; check model_name is valid |
401 |
Unauthorized | Invalid/expired JWT token | Regenerate JWT; verify AK/SK; check exp claim |
402 |
Payment Required | Insufficient credits | Top up API resource pack or subscription |
403 |
Forbidden | Content policy violation or API disabled | Review prompt against content policy; enable API access |
404 |
Not Found | Invalid task_id or wrong endpoint | Verify task_id; check endpoint path spelling |
429 |
Too Many Requests | Rate limit exceeded | Implement exponential backoff (see pattern below) |
500 |
Internal Server Error | Kling platform issue | Retry after 30s; if persistent, check status page |
502 |
Bad Gateway | Upstream service unavailable | Retry with backoff; typically transient |
503 |
Service Unavailable | System maintenance | Wait and retry; check announcements |
Task-Level Failures
When HTTP returns 200 but task_status is "failed":
taskstatusmsg |
Cause | Solution |
|---|---|---|
| Content policy violation | Prompt contains restricted content | Remove violent, adult, or copyrighted references |
| Image quality too low | Source image is blurry or too small | Use image >= 300x300px, clear and sharp |
| Prompt too complex | Too many scene elements | Simplify to 1-2 subjects, clear action |
| Generation timeout | Internal processing exceeded limit | Retry; reduce duration from 10s to 5s |
| Invalid image format | Unsupported file type | Use JPG, PNG, or WebP |
| Mask dimension mismatch | Mask size differs from source | Ensure mask matches source image dimensions exactly |
JWT Authentication Errors
Problem: 401 on every request
# WRONG — missing headers parameter
token = jwt.encode(payload, sk, algoritSecurity and compliance review framework for Kling AI integrations.
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
| Data | Location | Retention |
|---|---|---|
| Prompts | Sent to Kling servers (China/global) | Processing only |
| Generated videos | Kling CDN (temporary URLs) | ~24-72 hours |
| API keys | Your infrastructure | You control |
| Audit logs | Your infrastructure | You control |
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
| Question | Consideration |
|---|---|
| Do prompts contain PII? | Filter PII before sending to API |
| Do images contain faces? | Check consent requirements (GDPR Art. 6) |
| Are generated videos stored? | Define retention policy |
| Who has access to generated content? | RBAC on storage layer |
| Cross-border data transfer? | Kling API servers may be in China |
GDPR Considerations
class GDPRCompliantClient:
"""Kling client with GDPR data haImplement content policy compliance for Kling AI prompts and outputs.
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:
| Category | Examples |
|---|---|
| Violence/gore | Graphic injuries, torture, weapons used violently |
| Adult/sexual | Explicit nudity, sexual acts, suggestive content |
| Hate/discrimination | Slurs, targeted harassment, supremacist imagery |
| Illegal activity | Drug manufacturing, terrorism, fraud instructions |
| Real people | Deepfakes of identifiable individuals without consent |
| Copyrighted characters | Trademarked characters (Mickey Mouse, Spider-Man) |
| Misinformation | Fake news, fabricated events presented as real |
| Self-harm | Suicide, eating disorders, self-injury instructions |
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(&quImplement budget limits, usage alerts, and spending controls for Kling AI.
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
| Config | Credits |
|---|---|
| 5s standard | 10 |
| 5s professional | 35 |
| 10s standard | 20 |
| 10s professional | 70 |
| 5s standard + audio (v2.6) | 50 |
| 10s professional + audio (v2.6) | 200 |
| Image generation (Kolors) | 1 |
| Virtual try-on | 5 |
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 Set up logging and debugging for Kling AI API integrations.
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(Create your first Kling AI video generation with a minimal working example.
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-authsetup - Python 3.8+ with
requestsandPyJWT - 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/text2viAnimate static images into video using Kling AI.
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
| Parameter | Type | Required | Description |
|---|---|---|---|
model_name |
string | Yes | kling-v1-5, kling-v2-1, kling-v2-master, etc. |
image |
string | Yes | URL of the source image (JPG, PNG, WebP) |
prompt |
string | No | Motion description for the animation |
negative_prompt |
string | No | What to exclude |
duration |
string | Yes | "5" or "10" seconds |
aspect_ratio |
string | No | "16:9" default |
mode |
string | No | "standard" or "professional" |
cfg_scale |
float | No | Prompt adherence (0.0-1.0) |
image_tail |
string | No | End-frame image URL (mutually exclusive with masks/camera) |
camera_control |
object | No | Camera movement (mutually exclusive with masks/image_tail) |
static_mask |
string | No | Mask image URL for fixed regions |
dynamic_masks |
array | No | Motion brush trajectories |
callback_url |
string | No | Webhook for completion |
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",
Set up Kling AI API authentication with JWT tokens.
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
PyJWTor Node.js 18+
Step 1 — Get Credentials
- Sign in at app.klingai.com/global/dev
- Navigate to API Keys in the developer console
- Click Create API Key to generate an Access Key + Secret Key pair
- 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:
"&quoTrack and monitor Kling AI video generation task status.
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
| Status | Meaning | Typical Duration |
|---|---|---|
submitted |
Queued for processing | 0-30s |
processing |
Generation in progress | 30-120s (standard), 60-300s (professional) |
succeed |
Complete, video URL available | Terminal |
failed |
Generation failed | Terminal |
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:
Avoid common mistakes when using Kling AI API.
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 = dExplore Kling AI models, versions, and capabilities for video and image generation.
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
| Model ID | Supports | Max Duration | Resolution | Speed | Quality |
|---|---|---|---|---|---|
kling-v1 |
T2V, I2V | 10s | 720p | Fast | Good |
kling-v1-5 |
I2V only | 10s | 1080p | Fast | Better |
kling-v1-6 |
T2V, I2V | 10s | 1080p | Medium | Better+ |
kling-v2-master |
T2V, I2V | 10s | 1080p | Medium | High |
kling-v2-1 |
I2V only | 10s | 1080p | Medium | High |
kling-v2-1-master |
T2V, I2V | 10s | 1080p | Medium | High |
kling-v2-5-turbo |
T2V, I2V | 10s | 1080p 30fps | Fast | High |
kling-v2-6 |
T2V, I2V | 10s | 1080p 30-48fps | Medium | Highest |
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: truefor synchronized audio
Image Generation Models (Kolors)
| Model ID | Purpose | Resolution |
|---|---|---|
kolors-v1-5 |
Face/subject reference | Up to 2048x2048 |
kolors-v2-0 |
Image restyle | Up to 2048x2048 |
kolors-v2-1 |
Text-to-image | Up to 2048x2048 |
Specialty Models
| Feature | Endpoint | Model Versions | ||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| Lip Sync | /v1/videos/lip-sync |
v1.6+ | ||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
| Virtual Try-On | /v1/images/kolors-virtual-try-on |
v1.5 | ||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
| Video Extension | /v1/videos/video-extend |
All video models | ||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
| Effects | /v1/videos/effects |
v1.6+ | ||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
| <
Optimize Kling AI for speed, quality, and cost efficiency.
ReadWriteEditBash(npm:*)Grep
Kling AI Performance TuningOverviewOptimize 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
Connection PoolingUnderstand Kling AI pricing, credits, and cost optimization strategies.
ReadWriteEditBash(npm:*)Grep
Kling AI Pricing BasicsOverviewKling 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 PacksAPI 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 EstimationProduction readiness checklist for Kling AI integrations.
ReadWriteEditBash(npm:*)Grep
Kling AI Production ChecklistOverviewChecklist covering authentication, error handling, cost controls, monitoring, security, and content policy before deploying Kling AI video generation to production. Authentication
Error Handling
Cost Controls
Task Management
Content Safety
Security
Monitoring
Performance
Handle Kling AI API rate limits with backoff and queuing strategies.
ReadWriteEditBash(npm:*)Grep
Kling AI Rate LimitsOverviewKling AI enforces rate limits per API key. When exceeded, the API returns Rate Limit Tiers
Exponential Backoff with Jitter
Concurrent Task Limiter (asyncio)
Rate Limit MonitorProduction reference architecture for Kling AI video generation platforms.
ReadWriteEditBash(npm:*)Grep
Kling AI Reference ArchitectureOverviewProduction architecture for video generation platforms built on Kling AI. Covers API gateway, job queue, worker pool, storage, and monitoring layers. Architecture Diagram
Component DetailsAPI Layer
Worker ServiceProduction SDK patterns for Kling AI: client wrapper, retry logic, async polling, and error handling.
ReadWriteEditBash(npm:*)Grep
Kling AI SDK PatternsOverviewProduction-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 WrapperDownload and store Kling AI generated videos in cloud storage (S3, GCS, Azure).
ReadWriteEditBash(npm:*)Grep
Kling AI Storage IntegrationOverviewKling AI video URLs from Download from Kling CDN
Upload to AWS S3
Upload to Google Cloud StorageApply artistic styles and visual effects to Kling AI video generation.
ReadWriteEditBash(npm:*)Grep
Kling AI Style Transfer & EffectsOverviewApply 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 EngineeringThe most direct approach -- include style descriptors in your prompt:
Style Prompt Recipes
Effects APIThe Effects API applies pre-built transformations to existing images. Available on v1.6+. Endpoint: Configure Kling AI for teams with per-project API keys, usage quotas, and role-based access.
ReadWriteEditBash(npm:*)Grep
Kling AI Team SetupOverviewManage 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 KeysCreate separate API key pairs in the Kling AI developer console for each environment:
Team Configuration
Usage Quotas Per MemberGenerate videos from text prompts with Kling AI.
ReadWriteEditBash(npm:*)Grep
Kling AI Text-to-VideoOverviewGenerate videos from text prompts using the Endpoint: Request Parameters
Complete Example — PythonMigrate between Kling AI model versions safely.
ReadWriteEditBash(npm:*)Grep
Kling AI Upgrade & MigrationOverviewGuide 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
Breaking changes:
Migration: v2.x to v2.6 with Audio
Feature Availability Matrix
Basic ExtensionConfigure webhook callbacks for Kling AI task completion.
ReadWriteEditBash(npm:*)Grep
Kling AI Webhook ConfigurationOverviewInstead of polling task status, pass a Supported on: All video generation endpoints ( How It Works
Sending a Task with Callback
Webhook Receiver (Flask)Ready to use klingai-pack?Related Pluginsai-ethics-validatorAI ethics and fairness validation ai-experiment-loggerTrack and analyze AI experiments with a web dashboard and MCP tools ai-ml-engineering-packProfessional AI/ML Engineering toolkit: Prompt engineering, LLM integration, RAG systems, AI safety with 12 expert plugins ai-sdk-agentsMulti-agent orchestration with AI SDK v5 - handoffs, routing, and coordination for any AI provider (OpenAI, Anthropic, Google) anomaly-detection-systemDetect anomalies and outliers in data automl-pipeline-builderBuild AutoML pipelines
Tags
klingaiklingvideo-generationtext-to-videoai-videogenerative-aicreative
|