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

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

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