klingai-async-workflows

# Kling AI Async Workflows ## Overview Kling AI video generation is inherently async: you submit a task, then poll or receive a callback when done. This skill covers production patterns for integrating this into larger systems using queues, state machines, and event-driven architectures. ## Core Pattern: Submit + Callback ```python 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 submit_async(prompt, callback_url=None, **kwargs): """Submit task and return immediately.""" body = { "model_name": kwargs.get("model", "kling-v2-master"), "prompt": prompt, "duration": str(kwargs.get("duration", 5)), "mode": kwargs.get("mode", "standard"), } if callback_url: body["callback_url"] = callback_url r = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json=body) return r.json()["data"]["task_id"] ``` ## Redis Queue Workflow ```python import redis import json r = redis.Redis() # Producer: enqueue video generation requests def enqueue_video_job(prompt, metadata=None): job = { "id": f"job_{int(time.time() * 1000)...