klingai-sdk-patterns

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 <[e-mail]>
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

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 = (time.time())
         now >= (._token_expires - .config.token_buffer_sec):
            payload = {: .config.access_key, : now + , : now - }
            ._token = jwt.encode(payload, .config.secret_key,
                                     algorithm=,
                                     headers={: , : })
            ._token_expires = now + 
         {: ,
                : }

     () -> :
        r = requests.post(,
                          headers=._headers, json=body,
                          timeout=.config.timeout_sec)
        r.raise_for_status()
         r.json()

     () -> :
        r = requests.get(,
                         headers=._headers,
                         timeout=.config.timeout_sec)
        r.raise_for_status()
         r.json()

     () -> :
        
        interval = .config.poll_interval_sec
         attempt  (.config.max_poll_attempts):
            time.sleep(interval)
            result = ._get()
            status = result[][]
             status == :
                 result[][]
             status == :
                 KlingGenerationError(result[].get(, ))
            
            interval = (interval * , )
         KlingTimeoutError()

    

     () -> :
        body = {: kwargs.get(, ),
                : prompt,
                : (kwargs.get(, )),
                : kwargs.get(, ),
                : kwargs.get(, )}
         kwargs.get():
            body[] = kwargs[]
         kwargs.get()   :
            body[] = kwargs[]
         kwargs.get():
            body[] = kwargs[]

        task = ._post(, body)
        task_id = task[][]
         kwargs.get(, ):
             ._poll_task(, task_id)
         {: task_id}

     () -> :
        body = {: kwargs.get(, ),
                : image_url,
                : (kwargs.get(, )),
                : kwargs.get(, )}
         kwargs.get():
            body[] = kwargs[]

        task = ._post(, body)
        task_id = task[][]
         kwargs.get(, ):
             ._poll_task(, task_id)
         {: task_id}

     () -> :
        body = {: task_id,
                : kwargs.get(, ),
                : (kwargs.get(, )),
                : kwargs.get(, )}
        result = ._post(, body)
        new_task_id = result[][]
         kwargs.get(, ):
             ._poll_task(, new_task_id)
         {: new_task_id}


 ():
    

 ():
    

 ():
    

Usage

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

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":  };
  }

   () {
     res =  (, {
      : ,
      : .#(),
      : .({
        : opts. ?? ,
        prompt,
        : (opts. ?? ),
        : opts. ?? ,
        : opts. ?? ,
      }),
    });
     { data } =  res.();
     opts. ===  ? data : .#(, data.);
  }

   #() {
     ( i = ; i < ; i++) {
        ( (r, interval));
       res =  (, {
        : .#(),
      });
       { data } =  res.();
       (data. === )  data.;
       (data. === )   (data.);
      interval = .(interval * , );
    }
      ();
  }
}

Retry Decorator

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)

Resources

• Kling AI Developer Portal
• API Overview