From klingai-pack
Downloads expiring Kling AI video URLs and uploads to S3, GCS, or Azure Blob for persistent storage and CDN pipelines.
How this skill is triggered — by the user, by Claude, or both
Slash command
/klingai-pack:klingai-storage-integrationThis skill is limited to the following tools:
The summary Claude sees in its skill listing — used to decide when to auto-load this skill
Kling AI video URLs from `task_result.videos[].url` are temporary CDN links that expire. You must download and store videos in your own storage. This skill covers S3, GCS, and Azure Blob.
Kling AI video URLs from task_result.videos[].url are temporary CDN links that expire. You must download and store videos in your own storage. This skill covers S3, GCS, and Azure Blob.
import requests
import os
def download_video(video_url: str, output_dir: str = "output") -> str:
"""Download generated video from Kling CDN."""
os.makedirs(output_dir, exist_ok=True)
# Extract filename or generate one
filename = video_url.split("/")[-1].split("?")[0]
if not filename.endswith(".mp4"):
filename = f"kling_{int(time.time())}.mp4"
filepath = os.path.join(output_dir, filename)
response = requests.get(video_url, stream=True, timeout=120)
response.raise_for_status()
with open(filepath, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
size_mb = os.path.getsize(filepath) / (1024 * 1024)
print(f"Downloaded: {filepath} ({size_mb:.1f} MB)")
return filepath
import boto3
def upload_to_s3(filepath: str, bucket: str, key_prefix: str = "kling-videos/") -> str:
"""Upload video to S3 and return public URL."""
s3 = boto3.client("s3")
filename = os.path.basename(filepath)
s3_key = f"{key_prefix}{filename}"
s3.upload_file(
filepath, bucket, s3_key,
ExtraArgs={"ContentType": "video/mp4", "CacheControl": "max-age=86400"}
)
url = f"https://{bucket}.s3.amazonaws.com/{s3_key}"
print(f"Uploaded to S3: {url}")
return url
# Generate signed URL for private buckets
def get_signed_url(bucket: str, key: str, expiry: int = 3600) -> str:
s3 = boto3.client("s3")
return s3.generate_presigned_url(
"get_object",
Params={"Bucket": bucket, "Key": key},
ExpiresIn=expiry,
)
from google.cloud import storage
def upload_to_gcs(filepath: str, bucket_name: str, prefix: str = "kling-videos/") -> str:
"""Upload video to GCS and return public URL."""
client = storage.Client()
bucket = client.bucket(bucket_name)
filename = os.path.basename(filepath)
blob = bucket.blob(f"{prefix}{filename}")
blob.upload_from_filename(filepath, content_type="video/mp4")
blob.make_public() # or use signed URLs for private access
print(f"Uploaded to GCS: {blob.public_url}")
return blob.public_url
# Signed URL for private access
def get_gcs_signed_url(bucket_name: str, blob_name: str, expiry_min: int = 60) -> str:
from datetime import timedelta
client = storage.Client()
bucket = client.bucket(bucket_name)
blob = bucket.blob(blob_name)
return blob.generate_signed_url(expiration=timedelta(minutes=expiry_min))
from azure.storage.blob import BlobServiceClient
def upload_to_azure(filepath: str, container: str,
connection_string: str = None) -> str:
"""Upload video to Azure Blob Storage."""
conn_str = connection_string or os.environ["AZURE_STORAGE_CONNECTION_STRING"]
client = BlobServiceClient.from_connection_string(conn_str)
filename = os.path.basename(filepath)
blob_client = client.get_blob_client(container=container, blob=f"kling-videos/{filename}")
with open(filepath, "rb") as f:
blob_client.upload_blob(f, content_type="video/mp4", overwrite=True)
url = blob_client.url
print(f"Uploaded to Azure: {url}")
return url
def generate_and_store(prompt: str, bucket: str, provider: str = "s3"):
"""Generate video with Kling AI and store in cloud."""
# 1. Generate
r = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
"model_name": "kling-v2-master",
"prompt": prompt,
"duration": "5",
"mode": "standard",
}).json()
task_id = r["data"]["task_id"]
# 2. Poll
result = poll_task("/videos/text2video", task_id)
video_url = result["videos"][0]["url"]
# 3. Download
filepath = download_video(video_url)
# 4. Upload
if provider == "s3":
return upload_to_s3(filepath, bucket)
elif provider == "gcs":
return upload_to_gcs(filepath, bucket)
elif provider == "azure":
return upload_to_azure(filepath, bucket)
# 5. Cleanup temp file
os.remove(filepath)
import json
def save_with_metadata(filepath: str, task_id: str, prompt: str, model: str):
"""Save video metadata alongside the file."""
meta = {
"task_id": task_id,
"prompt": prompt,
"model": model,
"generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
"filename": os.path.basename(filepath),
}
meta_path = filepath.replace(".mp4", ".meta.json")
with open(meta_path, "w") as f:
json.dump(meta, f, indent=2)
return meta_path
npx claudepluginhub jeremylongshore/claude-code-plugins-plus-skills --plugin klingai-packProvides production reference architecture for scalable Kling AI video generation platforms, including API gateway, job queues, workers, storage, and monitoring.
Generates videos from text prompts or images, animates still images, and creates talking avatars from photos with audio using Kling AI models (VIDEO 3.0, Avatar 2.0, etc.). Handles multi-shot storyboards, character consistency, and prompt engineering.
Uploads promo videos and album content to Cloudflare R2 or AWS S3. Works after promo-director, before release-director. Manually invoked via /cloud-uploader.