Implement cost controls and budget management for Kling AI. Use when managing spending limits, tracking costs, or preventing budget overruns. Trigger with phrases like 'klingai cost', 'kling ai budget', 'klingai spending', 'video generation costs'.
npxskills add jeremylongshore/claude-code-plugins-plus-skills--skill klingai-cost-controlsLoading…
Prevent unexpected spending with per-request cost estimation, daily budget enforcement, threshold alerts, and usage dashboards. Credits are consumed per task based on duration, mode, and audio.
| Config | Credits |
|---|---|
| 5s standard | 10 |
| 5s professional | 35 |
| 10s standard | 20 |
| 10s professional | 70 |
| 5s standard + audio (v2.6) | 50 |
| 10s professional + audio (v2.6) | 200 |
| Image generation (Kolors) | 1 |
| Virtual try-on | 5 |
import time
from dataclasses import dataclass, field
@dataclass
class BudgetGuard:
"""Enforce daily credit budget with alerting."""
daily_limit: int = 1000
alert_threshold: float = 0.8 # alert at 80%
_used: int = 0
_reset_time: float = field(default_factory=time.time)
_alerts_sent: set = field(default_factory=set)
def _check_reset(self):
if time.time() - self._reset_time > 86400:
self._used = 0
self._reset_time = time.time()
self._alerts_sent.clear()
def estimate_credits(self, duration: int = 5, mode: str = "standard",
audio: bool = False) -> int:
base = {(5, "standard"): 10, (5, "professional"): 35,
(10, "standard"): 20, (10, "professional"): 70}
credits = base.get((duration, mode), 10)
if audio:
credits *= 5
return credits
def check(self, credits_needed: int) -> bool:
self._check_reset()
# Check alert threshold
usage_pct = (self._used + credits_needed) / self.daily_limit
if usage_pct >= self.alert_threshold and "80pct" not in self._alerts_sent:
self._alerts_sent.add("80pct")
self._on_alert(f"Budget at {usage_pct:.0%} ({self._used + credits_needed}/{self.daily_limit})")
if self._used + credits_needed > self.daily_limit:
raise RuntimeError(
f"Daily budget exceeded: {self._used} + {credits_needed} > {self.daily_limit} credits"
)
return True
def record(self, credits: int):
self._used += credits
def _on_alert(self, message: str):
"""Override for custom alerting (Slack, email, PagerDuty)."""
print(f"ALERT: {message}")
@property
def remaining(self) -> int:
self._check_reset()
return max(0, self.daily_limit - self._used)
@property
def usage_report(self) -> dict:
self._check_reset()
return {
"used": self._used,
"limit": self.daily_limit,
"remaining": self.remaining,
"usage_pct": f"{(self._used / self.daily_limit) * 100:.1f}%",
}
def pre_batch_check(prompts: list, budget: BudgetGuard,
duration: int = 5, mode: str = "standard"):
"""Estimate and validate batch cost before submission."""
per_video = budget.estimate_credits(duration, mode)
total = len(prompts) * per_video
print(f"Batch estimate: {len(prompts)} videos x {per_video} credits = {total} credits")
print(f"Budget remaining: {budget.remaining}")
if total > budget.remaining:
raise RuntimeError(
f"Batch needs {total} credits but only {budget.remaining} remaining. "
f"Reduce to {budget.remaining // per_video} videos or lower mode."
)
return total
class CostAwareKlingClient:
"""Kling client that enforces budget on every request."""
def __init__(self, base_client, budget: BudgetGuard):
self.client = base_client
self.budget = budget
def text_to_video(self, prompt: str, **kwargs):
credits = self.budget.estimate_credits(
kwargs.get("duration", 5),
kwargs.get("mode", "standard"),
kwargs.get("audio", False),
)
self.budget.check(credits)
result = self.client.text_to_video(prompt, **kwargs)
self.budget.record(credits)
return result
| Strategy | Savings | Implementation |
|---|---|---|
| Standard for drafts | 3.5x cheaper | mode: "standard" for iterations |
| 5s clips, extend later | 50% per clip | Generate 5s, use video-extend selectively |
| v2.5 Turbo over v2.6 | Faster (less queue cost) | model: "kling-v2-5-turbo" |
| Skip audio, add in post | 5x cheaper | motion_has_audio: false |
| Batch off-peak | Faster processing | Schedule overnight |
| Cache prompts | Avoid duplicates | Hash prompt + params, check before submitting |
import json
from datetime import datetime
class UsageTracker:
"""Log every generation for cost analysis."""
def __init__(self, log_file: str = "kling_usage.jsonl"):
self.log_file = log_file
def log(self, task_id: str, credits: int, model: str,
duration: int, mode: str, prompt: str):
entry = {
"timestamp": datetime.utcnow().isoformat(),
"task_id": task_id,
"credits": credits,
"model": model,
"duration": duration,
"mode": mode,
"prompt_preview": prompt[:100],
}
with open(self.log_file, "a") as f:
f.write(json.dumps(entry) + "\n")
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Set up and use 1Password CLI (op). Use when installing the CLI, enabling desktop app integration, signing in (single or multi-account), or reading/injecting/running secrets via op.
CLI to manage emails via IMAP/SMTP. Use `himalaya` to list, read, write, reply, forward, search, and organize emails from the terminal. Supports multiple accounts and message composition with MML (MIME Meta Language).
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Set up and use 1Password CLI (op). Use when installing the CLI, enabling desktop app integration, signing in (single or multi-account), or reading/injecting/running secrets via op.
CLI to manage emails via IMAP/SMTP. Use `himalaya` to list, read, write, reply, forward, search, and organize emails from the terminal. Supports multiple accounts and message composition with MML (MIME Meta Language).