klingai-debug-bundle

name	klingai-debug-bundle
description	Set up logging and debugging for Kling AI API integrations. Use when troubleshooting video generation or building observability. Trigger with phrases like 'klingai debug', 'kling ai logging', 'klingai troubleshoot', 'debug kling video generation'.
allowed-tools	Read, Write, Edit, Bash(npm:*), Grep
version	1.18.0
license	MIT
author	Jeremy Longshore <[e-mail]>
tags	["saas","kling-ai","debugging","observability"]
compatibility	Designed for Claude Code

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 = 
        start = time.monotonic()
        trace = {
            : datetime.utcnow().isoformat(),
            : method,
            : path,
            : body,
        }

        :
             method == :
                r = requests.post(url, headers=._get_headers(), json=body, timeout=)
            :
                r = requests.get(url, headers=._get_headers(), timeout=)

            trace[] = r.status_code
            trace[] = r.json()  r.content  
            trace[] = ((time.monotonic() - start) * )

            logger.debug()

             r.status_code >= :
                logger.error()

            r.raise_for_status()
             r.json()

         Exception  e:
            trace[] = (e)
            trace[] = ((time.monotonic() - start) * )
            logger.exception()
            
        :
            ._request_log.append(trace)

     ():
        body = {
            : kwargs.get(, ),
            : prompt,
            : (kwargs.get(, )),
            : kwargs.get(, ),
        }
        result = ._traced_request(, , body)
        task_id = result[][]
        logger.info()
         ._poll_with_logging(, task_id)

     ():
        start = time.monotonic()
         attempt  (max_attempts):
            time.sleep()
            result = ._traced_request(, )
            status = result[][]
            elapsed = (time.monotonic() - start)
            logger.info()

             status == :
                logger.info()
                 result[][]
             status == :
                msg = result[].get(, )
                logger.error()
                 RuntimeError(msg)

         TimeoutError()

     ():
         (filepath, )  f:
            json.dump(._request_log, f, indent=, default=)
        logger.info()

Usage

client = KlingDebugClient()
try:
    result = client.text_to_video("A cat surfing ocean waves at sunset")
    print(f"Video: {result['videos'][0]['url']}")
except Exception:
    pass
finally:
    client.dump_log()  # always save debug log

Structured Log Entry Format

{
  "timestamp": "2026-03-22T10:30:00.000Z",
  "method": "POST",
  "path": "/videos/text2video",
  "request_body": {"model_name": "kling-v2-master", "prompt": "..."},
  "status_code": 200,
  "response_body": {"code": 0, "data": {"task_id": "abc123"}},
  "duration_ms": 342
}

Quick Diagnostic Script

#!/bin/bash
# kling-diag.sh
echo "=== Kling AI Diagnostics ==="
echo "KLING_ACCESS_KEY: ${KLING_ACCESS_KEY:+set (${#KLING_ACCESS_KEY} chars)}"
echo "KLING_SECRET_KEY: ${KLING_SECRET_KEY:+set (${#KLING_SECRET_KEY} chars)}"

python3 -c "
import jwt, time, os, requests
ak = os.environ.get('KLING_ACCESS_KEY', '')
sk = os.environ.get('KLING_SECRET_KEY', '')
if not ak or not sk: print('ERROR: Missing credentials'); exit(1)
token = jwt.encode({'iss': ak, 'exp': int(time.time())+1800, 'nbf': int(time.time())-5},
                   sk, algorithm='HS256', headers={'alg':'HS256','typ':'JWT'})
r = requests.get('https://api.klingai.com/v1/videos/text2video',
                  headers={'Authorization': f'Bearer {token}'}, timeout=10)
print(f'Auth test: HTTP {r.status_code}')
if r.status_code == 401: print('Fix: Check AK/SK values')
elif r.status_code in (200, 400): print('Auth OK')
"

Task Inspector

def inspect_task(client, endpoint, task_id):
    """Print detailed task information."""
    result = client._traced_request("GET", f"{endpoint}/{task_id}")
    data = result["data"]
    print(f"Task ID:     {data['task_id']}")
    print(f"Status:      {data['task_status']}")
    print(f"Created:     {data.get('created_at', 'N/A')}")
    if data["task_status"] == "succeed":
        for i, video in enumerate(data["task_result"]["videos"]):
            print(f"Video [{i}]:   {video['url']}")
    elif data["task_status"] == "failed":
        print(f"Error:       {data.get('task_status_msg', 'No message')}")

Resources

• API Reference
• Developer Console