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, also compatible with Codex and OpenClaw

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


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


 ():
    

 ():
    

 ():