Klingai Sdk PatternsSAFE

Model-agnostic agent-skills platform with a harness-free canonical layer, verified adapters, and the ccpi package manager. Explore at tonsofskills.com.

The instruction file, verbatim from the audited commit — this is the text the model reads, and the surface the audit's instruction layer examines. Quoted here so you can judge it without cloning anything.

---
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 <[email protected]>
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

SAFEgrade B · trust 89/100 Nothing in the source contradicts what it says it does. Grade A is reserved for packages that have also passed the behavioural sandbox.

What the source does

Filesystem

none-observed

Network

none-observed

Shell

none-observed

Dependencies

pinned

Secrets in source

none-found

Findings (0)

No findings outside the package's declared scope.

Gates applied: no_behavioural_pass.