--- name: klingai-storage-integration description: 'Download and store Kling AI generated videos in cloud storage (S3, GCS, Azure). Use when persisting videos or building CDN pipelines. Trigger with phrases like ''klingai storage'', ''save klingai video'', ''kling ai s3 upload'', ''klingai cloud storage''. ' allowed-tools: Read, Write, Edit, Bash(npm:*), Grep version: 1.18.0 license: MIT author: Jeremy Longshore tags: - saas - kling-ai - storage - s3 - gcs compatibility: Designed for Claude Code --- # Kling AI Storage Integration ## Overview 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. ## Download from Kling CDN ```python 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 ``` ## Upload to AWS S3 ```python 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, ) ``` ## Upload to Google Cloud Storage ```python 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)) ``` ## Upload to Azure Blob Storage ```python 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 ``` ## End-to-End Pipeline ```python 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) ``` ## Metadata Preservation ```python 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 ``` ## Prerequisites - An approved storage destination, encryption and retention policy, rights-cleared or synthetic draft asset, least-privilege service identity, metadata-redaction rules, and a tested deletion path. ## Instructions 1. Upload only watermarked draft canaries to an allowlisted sandbox bucket; reject public ACLs, unapproved regions, or assets without rights and policy clearance. 2. Encrypt at rest and in transit, store only redacted metadata, and verify destination, access scope, retention, and removal controls before approval. 3. Halt uploads on permission, policy, rights, region, or retention drift; delete staged assets and restore the prior storage configuration. 4. Promote only after owner approval and preserve a redacted receipt rather than prompt text, asset URLs, or identifying metadata. ## Output Produce a storage receipt with asset classification, destination/region classification, encryption and access checks, policy/rights outcome, draft-only status, retention/deletion proof, approver, and rollback reference. Exclude prompts, URLs, identities, and credentials. ## Error Handling | Condition | Response | |---|---| | Destination, ACL, or region is unapproved | Stop the transfer, delete staged copies, and restore the approved destination policy. | | Rights, policy, or retention control fails | Quarantine and remove the draft; require owner review before resubmission. | ## Examples `asset=synthetic-draft; destination=approved-sandbox; encryption=pass; acl=private; policy=pass; retention=24h; deletion=tested` supports a controlled upload. ## Resources - [API Reference](https://app.klingai.com/global/dev/document-api/apiReference/model/textToVideo) - [AWS S3 SDK](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html) - [Google Cloud Storage](https://cloud.google.com/storage/docs)