klingai-batch-processing

name	klingai-batch-processing
description	Process multiple video generation requests efficiently with Kling AI. Use when generating batches of videos or building content pipelines. Trigger with phrases like 'klingai batch', 'kling ai bulk', 'multiple videos klingai', 'klingai parallel generation'.
allowed-tools	Read, Write, Edit, Bash(npm:*), Grep
version	1.18.0
license	MIT
author	Jeremy Longshore <[e-mail]>
tags	["saas","kling-ai","batch","pipelines"]
compatibility	Designed for Claude Code, also compatible with Codex and OpenClaw

Kling AI Batch Processing

Overview

Generate multiple videos efficiently using controlled parallelism, rate-limit-aware submission, progress tracking, and result collection. All requests go through https://api.klingai.com/v1.

Batch Submission with Rate Limiting

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_batch(prompts, model="kling-v2-master", duration="5",
                 mode="standard", max_concurrent=3, delay=2.0):
    """Submit batch with controlled concurrency and pacing."""
    tasks = []
    active = []

    for i, prompt in enumerate(prompts):
        # Wait if at concurrency limit
        while len(active) >= max_concurrent:
            active = [t for t in active if  check_complete(t[])]
             (active) >= max_concurrent:
                time.sleep()

        response = requests.post(, headers=get_headers(), json={
            : model,
            : prompt,
            : duration,
            : mode,
        })
        data = response.json()[]
        task = {: data[], : prompt, : i}
        tasks.append(task)
        active.append(task)
        ()
        time.sleep(delay)  

     tasks

 ():
    r = requests.get(, headers=get_headers()).json()
     r[][]  (, )