Klingai Rate Limits

What it does

When your agent uses it

• Hitting 429 errors
• Planning high-volume workflows
• With phrases like klingai rate limit
• Klingai throttle

Example prompts

• “klingai rate limit”
• “kling ai 429”
• “klingai throttle”
• “/klingai-rate-limits”

Requirements

• Python 3
• Compatibility (from SKILL.md): Designed for Claude Code
• Pre-approved tools (allowed-tools): Read, Write, Edit, Bash(npm:*), Grep

Workflow steps

4 steps, taken from the first numbered list in SKILL.md.

1. 1Exercise limits with bounded draft-only canaries; reject unapproved sources, publishing destinations, or requests that exceed the approved…
2. 2Use idempotency keys and backoff, recording aggregate status and credit consumption rather than prompt content or asset URLs.
3. 3Stop queued work on quota, policy, rights, or retention drift; cancel tasks and restore the prior rate configuration before retrying.
4. 4Keep a redacted receipt only and delete test artifacts when the approved retention window ends.

What it can do on your machine

Read from SKILL.md and the folder at commit cfae287. It shows what the files ask for, not the result of running them.

• Tool permissions

  Pre-approves these tools, so the agent can use them without asking each time:

  • Read
  • Write
  • Edit
  • Bash(npm:*)
  • Grep

  From allowed-tools in the SKILL.md frontmatter.

• Runs code

  No scripts in the folder and no shell commands in SKILL.md (its code samples are python).

  From the folder's file list and the shell code blocks in SKILL.md.

• Network

  No URLs in SKILL.md.

  From URLs in SKILL.md, links to its own repository left out.

• Credentials

  Names no API keys, tokens, secrets or passwords.

  From names ending in _API_KEY, _TOKEN, _SECRET, _KEY or _PASSWORD in SKILL.md.

• Compatibility

  Designed for Claude Code

  From compatibility in the SKILL.md frontmatter.

Context cost

Klingai Rate Limits loads about 1.7k tokens when it runs, and up to ~3.7k if it reads all its reference files. Until then it costs about 64 tokens; SKILL.md has 312 words of instructions outside code blocks.

Always · name and description, kept in context so the agent knows when to use it

~64

When it runs · the whole SKILL.md, loaded when a task matches

~1.7k

With references · SKILL.md plus every file in references/, read only if the agent opens them

~3.7k

Estimates: characters ÷ 4, the usual rule of thumb; real counts depend on the model's tokenizer. Scripts and assets cost tokens only if the agent reads them.

Safety

Automated static check — not a guarantee. Review scripts before installing. It scans the text of SKILL.md for risky patterns (piping downloads into a shell, reading credential files, hidden Unicode, destructive commands); files beside SKILL.md are not scanned.

SKILL.md

The full file from jeremylongshore/tons-of-skills-marketplace at commit cfae287, republished under its MIT licence (© jeremylongshore). 312 words, ~1,676 tokens.

name

klingai-rate-limits

description

Handle Kling AI API rate limits with backoff and queuing strategies. Use when hitting 429 errors or planning high-volume workflows. Trigger with phrases like 'klingai rate limit', 'kling ai 429', 'klingai throttle', 'kling api limits'.

allowed-tools

Read, Write, Edit, Bash(npm:*), Grep

compatibility

Designed for Claude Code

version

1.18.0

license

MIT

author

Jeremy Longshore <jeremy@intentsolutions.io>

tags

saas, kling-ai, rate-limits, reliability

import time, random, requests

def exponential_backoff(attempt: int, base: float = 1.0, max_wait: float = 60.0) -> float:
    """Calculate wait time with jitter to avoid thundering herd."""
    wait = min(base * (2 ** attempt), max_wait)
    jitter = random.uniform(0, wait * 0.5)
    return wait + jitter

def request_with_retry(method, url, headers, json=None, max_retries=5):
    for attempt in range(max_retries + 1):
        response = method(url, headers=headers, json=json, timeout=30)

        if response.status_code == 429:
            if attempt == max_retries:
                raise RuntimeError("Rate limit: max retries exceeded")
            wait = exponential_backoff(attempt)
            print(f"429 rate limited. Waiting {wait:.1f}s (attempt {attempt + 1})")
            time.sleep(wait)
            continue

        if response.status_code >= 500:
            if attempt == max_retries:
                response.raise_for_status()
            time.sleep(exponential_backoff(attempt, base=2.0))
            continue

        response.raise_for_status()
        return response

    raise RuntimeError("Unreachable")

import asyncio

class TaskLimiter:
    """Limit concurrent Kling AI tasks to stay within API tier."""

    def __init__(self, max_concurrent: int = 3):
        self._semaphore = asyncio.Semaphore(max_concurrent)
        self._active = 0

    async def submit(self, coro):
        async with self._semaphore:
            self._active += 1
            try:
                return await coro
            finally:
                self._active -= 1

    @property
    def active_count(self) -> int:
        return self._active

# Usage
limiter = TaskLimiter(max_concurrent=3)
tasks = [limiter.submit(generate_video(p)) for p in prompts]
results = await asyncio.gather(*tasks, return_exceptions=True)

class RateLimitMonitor:
    """Track API call frequency and warn before hitting limits."""

    def __init__(self, max_per_minute: int = 30):
        self.max_per_minute = max_per_minute
        self._calls = []

    def record_call(self):
        now = time.time()
        self._calls = [t for t in self._calls if now - t < 60]
        self._calls.append(now)

    @property
    def usage_pct(self) -> float:
        now = time.time()
        recent = sum(1 for t in self._calls if now - t < 60)
        return (recent / self.max_per_minute) * 100

    def wait_if_needed(self):
        if self.usage_pct > 80 and self._calls:
            wait = 60 - (time.time() - self._calls[0])
            if wait > 0:
                print(f"Throttling: waiting {wait:.1f}s ({self.usage_pct:.0f}% of limit)")
                time.sleep(wait)

from collections import deque
import threading

class RequestQueue:
    """FIFO queue with rate-limit-aware dispatch."""

    def __init__(self, client, max_per_minute: int = 30):
        self.client = client
        self.interval = 60.0 / max_per_minute
        self._queue = deque()

    def enqueue(self, endpoint: str, body: dict, callback=None):
        self._queue.append((endpoint, body, callback))

    def process_all(self):
        while self._queue:
            endpoint, body, callback = self._queue.popleft()
            try:
                result = self.client._post(endpoint, body)
                if callback:
                    callback(result, error=None)
            except Exception as e:
                if callback:
                    callback(None, error=e)
            time.sleep(self.interval)

© jeremylongshore, MIT. Rendered from Markdown: HTML in the file is shown as text, images as links, and headings moved down two levels. Raw file

Files

SKILL.md and 6 other files (references) in skills/.curated/klingai-rate-limits of jeremylongshore/tons-of-skills-marketplace.

• SKILL.md
• references/concurrent-job-manager.md
• references/errors.md
• references/examples.md
• references/exponential-backoff.md
• references/request-queue.md
• references/token-bucket-rate-limiter.md

Open the folder on GitHubat commit cfae287