--- name: klingai-sdk-patterns description: 'Production SDK patterns for Kling AI: client wrapper, retry logic, async polling, and error handling. Use when building robust integrations. Trigger with phrases like ''klingai sdk'', ''kling ai client'', ''klingai patterns'', ''kling ai wrapper''. ' allowed-tools: Read, Write, Edit, Bash(npm:*), Grep version: 1.18.0 license: MIT author: Jeremy Longshore tags: - saas - kling-ai - sdk - patterns compatibility: Designed for Claude Code --- # 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 ```python 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"] elif status == "failed": raise KlingGenerationError(result["data"].get("task_status_msg", "Unknown")) # Increase interval up to 30s max interval = min(interval * 1.2, 30) raise KlingTimeoutError(f"Task {task_id} did not complete in time") # --- Public API --- def text_to_video(self, prompt: str, **kwargs) -> dict: body = {"model_name": kwargs.get("model", "kling-v2-master"), "prompt": prompt, "duration": str(kwargs.get("duration", 5)), "aspect_ratio": kwargs.get("aspect_ratio", "16:9"), "mode": kwargs.get("mode", "standard")} if kwargs.get("negative_prompt"): body["negative_prompt"] = kwargs["negative_prompt"] if kwargs.get("cfg_scale") is not None: body["cfg_scale"] = kwargs["cfg_scale"] if kwargs.get("callback_url"): body["callback_url"] = kwargs["callback_url"] task = self._post("/videos/text2video", body) task_id = task["data"]["task_id"] if kwargs.get("wait", True): return self._poll_task("/videos/text2video", task_id) return {"task_id": task_id} def image_to_video(self, image_url: str, **kwargs) -> dict: body = {"model_name": kwargs.get("model", "kling-v2-1"), "image": image_url, "duration": str(kwargs.get("duration", 5)), "mode": kwargs.get("mode", "standard")} if kwargs.get("prompt"): body["prompt"] = kwargs["prompt"] task = self._post("/videos/image2video", body) task_id = task["data"]["task_id"] if kwargs.get("wait", True): return self._poll_task("/videos/image2video", task_id) return {"task_id": task_id} def extend_video(self, task_id: str, **kwargs) -> dict: body = {"task_id": task_id, "prompt": kwargs.get("prompt", ""), "duration": str(kwargs.get("duration", 5)), "mode": kwargs.get("mode", "standard")} result = self._post("/videos/video-extend", body) new_task_id = result["data"]["task_id"] if kwargs.get("wait", True): return self._poll_task("/videos/video-extend", new_task_id) return {"task_id": new_task_id} class KlingError(Exception): pass class KlingGenerationError(KlingError): pass class KlingTimeoutError(KlingError): pass ``` ## Usage ```python client = KlingClient() # Synchronous (waits for result) result = client.text_to_video( "A cat playing piano in a jazz club", model="kling-v2-6", mode="professional", duration=5, ) print(result["videos"][0]["url"]) # Fire-and-forget (returns task_id) task = client.text_to_video("Ocean waves at sunset", wait=False) print(f"Submitted: {task['task_id']}") ``` ## Node.js Client ```javascript import jwt from "jsonwebtoken"; class KlingClient { #token = null; #tokenExp = 0; constructor(ak = process.env.KLING_ACCESS_KEY, sk = process.env.KLING_SECRET_KEY) { this.ak = ak; this.sk = sk; this.base = "https://api.klingai.com/v1"; } #getHeaders() { const now = Math.floor(Date.now() / 1000); if (now >= this.#tokenExp - 300) { this.#token = jwt.sign( { iss: this.ak, exp: now + 1800, nbf: now - 5 }, this.sk, { algorithm: "HS256", header: { typ: "JWT" } } ); this.#tokenExp = now + 1800; } return { Authorization: `Bearer ${this.#token}`, "Content-Type": "application/json" }; } async textToVideo(prompt, opts = {}) { const res = await fetch(`${this.base}/videos/text2video`, { method: "POST", headers: this.#getHeaders(), body: JSON.stringify({ model_name: opts.model ?? "kling-v2-master", prompt, duration: String(opts.duration ?? 5), aspect_ratio: opts.aspectRatio ?? "16:9", mode: opts.mode ?? "standard", }), }); const { data } = await res.json(); return opts.wait === false ? data : this.#poll("/videos/text2video", data.task_id); } async #poll(endpoint, taskId, interval = 10000) { for (let i = 0; i < 120; i++) { await new Promise((r) => setTimeout(r, interval)); const res = await fetch(`${this.base}${endpoint}/${taskId}`, { headers: this.#getHeaders(), }); const { data } = await res.json(); if (data.task_status === "succeed") return data.task_result; if (data.task_status === "failed") throw new Error(data.task_status_msg); interval = Math.min(interval * 1.2, 30000); } throw new Error(`Timeout: task ${taskId}`); } } ``` ## Retry Decorator ```python import functools def retry_on_transient(max_retries=3, backoff_base=2): """Retry on 429 (rate limit) and 5xx (server) errors.""" def decorator(fn): @functools.wraps(fn) def wrapper(*args, **kwargs): for attempt in range(max_retries + 1): try: return fn(*args, **kwargs) except requests.HTTPError as e: if e.response.status_code in (429, 500, 502, 503) and attempt < max_retries: wait = backoff_base ** attempt time.sleep(wait) continue raise return wrapper return decorator # Apply to client methods KlingClient._post = retry_on_transient()(KlingClient._post) ``` ## Prerequisites - A pinned SDK version, sandbox workspace, secret references, synthetic or rights-cleared brief, approved rate/credit limits, content-policy check, and rollback owner. ## Instructions 1. Exercise SDK patterns with one sandbox, draft-only canary; do not hard-code credentials or submit private/unlicensed materials. 2. Apply idempotency, bounded retries, redacted telemetry, and explicit task cancellation; verify policy, rights, credit, destination, and retention controls. 3. Stop on authentication, policy, rights, scope, budget, or retention drift and restore the previous client configuration. 4. Promote dependency or retry changes only after owner approval; remove temporary assets and retain a redacted receipt. ## Output Produce an SDK receipt with version, environment, test classification, retry/idempotency result, aggregate task outcome, policy/rights/budget checks, draft-only assertion, approver, and cleanup/rollback reference. Exclude prompts, assets, identities, and secrets. ## Error Handling | Condition | Response | |---|---| | Retry creates duplicate work | Cancel duplicate tasks, preserve only aggregate redacted evidence, and enforce idempotency before retrying. | | Policy, rights, budget, or retention drift | Stop the canary, remove drafts, and restore the approved SDK configuration. | ## Examples `sdk=pinned; env=ci-sandbox; retry=bounded; idempotency=pass; policy=pass; destination=draft-only; cleanup=verified` is an acceptable pattern test. ## Resources - [Kling AI Developer Portal](https://app.klingai.com/global/dev) - [API Overview](https://app.klingai.com/global/dev/document-api/quickStart/productIntroduction/overview)