| name | klingai-job-monitoring |
| description | Track and monitor Kling AI video generation task status. Use when building dashboards,
tracking batch jobs, or debugging stuck tasks. Trigger with phrases like 'klingai job status',
'kling ai monitor', 'track klingai task', 'klingai progress'.
|
| allowed-tools | Read, Write, Edit, Bash(npm:*), Grep |
| version | 1.18.0 |
| license | MIT |
| author | Jeremy Longshore <jeremy@intentsolutions.io> |
| tags | ["saas","kling-ai","monitoring","jobs"] |
| compatibility | Designed for Claude Code, also compatible with Codex and OpenClaw |
Kling AI Job Monitoring
Overview
Every Kling AI generation returns a task_id. This skill covers polling strategies, batch tracking, timeout handling, and callback-based monitoring for the /v1/videos/text2video, /v1/videos/image2video, and /v1/videos/video-extend endpoints.
Task Lifecycle
| Status | Meaning | Typical Duration |
|---|
submitted | Queued for processing | 0-30s |
processing | Generation in progress | 30-120s (standard), 60-300s (professional) |
succeed | Complete, video URL available | Terminal |
failed | Generation failed | Terminal |
Polling a Single Task
import jwt, time, os, requests
BASE = "https://api.klingai.com/v1"
def get_headers():
ak, sk = os.environ["KLING_ACCESS_KEY"], os.environ["KLING_SECRET_KEY"]
token = jwt.encode(
{"iss": ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5},
sk, algorithm="HS256", headers={"alg": "HS256", "typ": "JWT"}
)
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
def poll_task(endpoint: str, task_id: , interval: = , timeout: = ):
start = time.monotonic()
attempts =
time.monotonic() - start < timeout:
time.sleep(interval)
attempts +=
r = requests.get(, headers=get_headers(), timeout=)
data = r.json()[]
status = data[]
elapsed = (time.monotonic() - start)
()
status == :
data[]
status == :
RuntimeError()
attempts > :
interval = (interval * , )
TimeoutError()
