Kling AI Video Extension

Overview

Extend an existing video by appending additional seconds. The extension endpoint takes the task_id of a completed video and generates a seamless continuation.

Endpoint: POST https://api.klingai.com/v1/videos/video-extend

Request Parameters

Parameter	Type	Required	Description
task_id	string	Yes	Task ID of the completed source video
prompt	string	No	Motion/scene description for extension
duration	string	No	Extension length: "5" (default)
mode	string	No	"standard" or "professional"
model_name	string	No	Default: "kling-v2-master"
callback_url	string	No	Webhook for completion

Basic Extension

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"}

# Step 1: Generate the initial 5s video
initial = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
    "model_name": "kling-v2-master",
    "prompt": "A rocket launching from a desert landscape, cinematic",
    "duration": "5",
    "mode": "standard",
}).json()
initial_task_id = initial["data"]["task_id"]

# Wait for completion...
# (poll until task_status == "succeed")

# Step 2: Extend by 5 more seconds
extension = requests.post(f"{BASE}/videos/video-extend", headers=get_headers(), json={
    "task_id": initial_task_id,
    "prompt": "The rocket ascends through clouds into the stratosphere",
    "duration": "5",
    "mode": "standard",
}).json()
ext_task_id = extension["data"]["task_id"]

# Step 3: Poll extension task
while True:
    time.sleep(15)
    result = requests.get(
        f"{BASE}/videos/video-extend/{ext_task_id}", headers=get_headers()
    ).json()
    if result["data"]["task_status"] == "succeed":
        extended_url = result["data"]["task_result"]["videos"][0]["url"]
        print(f"Extended video: {extended_url}")
        break
    elif result["data"]["task_status"] == "failed":
        print(f"Failed: {result['data']['task_status_msg']}")
        break

Chain Multiple Extensions

def chain_extensions(initial_task_id: str, prompts: list[str],
                     duration: str = "5", mode: str = "standard") -> list[str]:
    """Chain multiple extensions to build a longer video."""
    current_task_id = initial_task_id
    video_urls = []

    for i, prompt in enumerate(prompts):
        print(f"Extension {i + 1}/{len(prompts)}: submitting...")

        # Submit extension
        r = requests.post(f"{BASE}/videos/video-extend", headers=get_headers(), json={
            "task_id": current_task_id,
            "prompt": prompt,
            "duration": duration,
            "mode": mode,
        }).json()
        ext_task_id = r["data"]["task_id"]

        # Poll for completion
        while True:
            time.sleep(15)
            result = requests.get(
                f"{BASE}/videos/video-extend/{ext_task_id}", headers=get_headers()
            ).json()
            status = result["data"]["task_status"]

            if status == "succeed":
                url = result["data"]["task_result"]["videos"][0]["url"]
                video_urls.append(url)
                current_task_id = ext_task_id  # next extension chains from this
                print(f"Extension {i + 1} complete: {url}")
                break
            elif status == "failed":
                raise RuntimeError(f"Extension {i + 1} failed: {result['data']['task_status_msg']}")

    return video_urls

Usage: Build a 20-Second Video

# Generate initial 5s
initial_r = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
    "model_name": "kling-v2-master",
    "prompt": "Morning sunrise over a mountain lake, mist rising",
    "duration": "5",
    "mode": "standard",
}).json()
initial_id = initial_r["data"]["task_id"]
# ... poll until complete ...

# Chain 3 more extensions = 5 + 5 + 5 + 5 = 20 seconds total
extensions = chain_extensions(initial_id, [
    "Sun rises higher, birds begin flying across the lake",
    "A deer approaches the water's edge to drink",
    "Wide shot pulling back to reveal the full mountain range",
])

Cost

Each extension costs the same as a new generation:

Extension Duration	Standard	Professional
5 seconds	10 credits	35 credits

A 20-second video (initial + 3 extensions) costs 40 credits in standard mode.

Error Handling

Error	Cause	Fix
Invalid task_id	Source task doesn't exist	Verify task_id is from a completed generation
Source not complete	Extending a task still processing	Wait for source task to reach succeed status
Extension failed	Prompt conflict with source	Align extension prompt with original scene

Prerequisites

• A completed, rights-cleared or synthetic source draft, approved extension brief, authorized workspace and credit budget, policy review, draft-only destination, and a rollback/removal owner.

Instructions

1. Confirm source ownership, completion state, and approved continuity brief before creating an extension; do not use private, unlicensed, or policy-restricted material.
2. Submit one watermarked draft canary within the approved duration and credit cap, then verify policy, rights, quality, and destination controls.
3. Halt on continuity, policy, rights, or budget drift and delete the extension task instead of publishing it.
4. Promote only after owner approval; retain a redacted receipt and remove temporary assets at the agreed retention boundary.

Output

Produce an extension receipt with source task reference, rights classification, mode/duration, credit estimate, policy review, draft destination, approver, retention/removal reference, and task ID. Exclude prompts, identities, and asset URLs.

Examples

source=synthetic-task-42; rights=cleared; duration=5s; mode=standard; policy=pass; destination=draft-only; cleanup=24h is safe for review.

Resources

• Video Extension API
• Developer Portal