Kling 3.0 API: Submit, Poll, and Download Your First Video

A Kling 3.0 API integration has three stages: submit a task, poll its saved task ID, and download the completed video's URL. For this version, the returned URL is data.output.video. Checking the response field before building your download worker prevents a completed generation from being mistaken for an empty result.

This walkthrough uses PiAPI's documented contract. The examples have been checked locally with synthetic responses; no generated clip or measured model latency is presented.

1. Save a text-to-video request

You need Python 3.10 or later and a PiAPI key available to your process as PIAPI_API_KEY. Obtain the key in your Workspace and keep it out of source files and command arguments. This example uses only the Python standard library.

Save the following as kling-request.json:

{
  "model": "kling",
  "task_type": "video_generation",
  "input": {
    "prompt": "A paper boat crosses a quiet puddle after rain, slow camera push-in, natural light.",
    "version": "3.0",
    "mode": "std",
    "duration": 5,
    "aspect_ratio": "16:9",
    "enable_audio": false,
    "prefer_multi_shots": false
  }
}

Send it to POST https://api.piapi.ai/api/v1/task. model is kling; task_type is video_generation; input.version selects 3.0. The request asks for a five-second, 16:9 clip without generated audio or multi-shot control.

The create response supplies data.task_id. That ID belongs to the submitted generation; it is also the recovery handle if your client loses a later status response.

2. Submit, then poll the same ID

Save this script as kling_first_call.py. It writes a receipt before submitting, refuses to reuse that receipt for another create, and uses separate submit and download commands.

"""Small tutorial client. API credentials are read only when a request is made."""
import argparse
import hashlib
import shutil
import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

API = "https://api.piapi.ai/api/v1"


def api_json(path, payload=None, timeout=30):
    key = os.environ.get("PIAPI_API_KEY")
    if not key:
        raise RuntimeError("Set PIAPI_API_KEY in your environment")
    request = urllib.request.Request(
        API + path,
        data=None if payload is None else json.dumps(payload).encode(),
        headers={"X-API-Key": key, "Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            body = json.load(response)
    except urllib.error.HTTPError as error:
        raise RuntimeError(f"API HTTP {error.code}; do not blindly resubmit") from None
    if not isinstance(body, dict):
        raise RuntimeError("Expected a JSON object")
    if "code" in body and body["code"] != 200:
        raise RuntimeError(f"API code {body['code']}; inspect the task separately")
    return body


def task_object(body):
    # Unified video responses wrap data; the documented image async shell does not.
    task = body.get("data", body)
    if not isinstance(task, dict):
        raise RuntimeError("Expected a task object")
    return task


def submit_once(path, payload, receipt, request=api_json):
    # Reserve the receipt before POST. An interrupted create must be reconciled.
    with Path(receipt).open("x") as file:
        json.dump({"state": "submission_started", "endpoint": path}, file)
    task = task_object(request(path, payload))
    task_id = task.get("task_id")
    if not isinstance(task_id, str) or not task_id:
        raise RuntimeError("Create response has no task_id; reconcile before resubmitting")
    print(f"task_id={task_id}", flush=True)
    Path(receipt).write_text(json.dumps({"task_id": task_id, "endpoint": path}) + "\n")
    return task_id


def wait_for_task(task_id, request=api_json, interval=5, timeout=900):
    if interval < 0 or timeout <= 0:
        raise ValueError("Invalid polling interval or timeout")
    deadline = time.monotonic() + timeout
    while (remaining := deadline - time.monotonic()) > 0:
        task = task_object(request(
            "/task/" + urllib.parse.quote(task_id, safe=""),
            timeout=min(30, remaining),
        ))
        state = task.get("status")
        if state == "completed":
            return task
        if state == "failed":
            raise RuntimeError(f"Task {task_id} failed; inspect its error response")
        if state not in ("pending", "processing", "staged"):
            raise RuntimeError(f"Unrecognized task status {state!r}; inspect before continuing")
        time.sleep(min(interval, max(0, deadline - time.monotonic())))
    raise TimeoutError(f"Client timeout; resume polling saved task {task_id}")


def saved_task_id(receipt):
    task_id = json.loads(Path(receipt).read_text()).get("task_id")
    if not isinstance(task_id, str) or not task_id:
        raise RuntimeError("No saved task_id; reconcile the interrupted create before retrying")
    return task_id



def video_url(task):
    url = (task.get("output") or {}).get("video")
    if not isinstance(url, str) or urllib.parse.urlsplit(url).scheme != "https":
        raise RuntimeError("Completed Kling 3.0 task has no HTTPS output.video")
    return url


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("action", choices=["submit", "download"])
    parser.add_argument("receipt")
    parser.add_argument("file", help="request JSON for submit; new MP4 path for download")
    args = parser.parse_args()
    if args.action == "submit":
        payload = json.loads(Path(args.file).read_text())
        if (payload.get("model"), payload.get("task_type"), payload.get("input", {}).get("version")) != ("kling", "video_generation", "3.0"):
            raise ValueError("This example expects Kling video_generation version 3.0")
        submit_once("/task", payload, args.receipt)
    else:
        # Never forward the API key to the output asset host.
        task = wait_for_task(saved_task_id(args.receipt))
        url = video_url(task)
        with urllib.request.urlopen(url, timeout=60) as response, Path(args.file).open("xb") as file:
            shutil.copyfileobj(response, file)
        digest = hashlib.sha256(Path(args.file).read_bytes()).hexdigest()
        print(f"Saved {args.file}; sha256={digest}")


if __name__ == "__main__":
    main()

Run the two stages:

python3 kling_first_call.py submit kling-task.json kling-request.json
python3 kling_first_call.py download kling-task.json kling-result.mp4

The second command polls GET /api/v1/task/{task_id}. pending, staged, and processing keep the wait active; completed permits download; failed stops it. An unfamiliar status also stops the example so it cannot silently treat a changed response contract as success. The five-second polling interval and fifteen-minute client deadline are choices in this example, not a generation-time guarantee.

A read or network error stops the command. Resume with the download command and the existing receipt. If a create was interrupted before its ID was saved, the receipt remains submission_started: reconcile that submission in task history before deciding whether to create a replacement. Do not delete that receipt and rerun blindly.

3. Download the video field for version 3.0

The script reads output.video from the task object, downloads the asset without forwarding the API key, and prints the file's SHA-256. The video URL is not a permanent archive: retain the file and the request/receipt together.

The hash identifies the downloaded bytes. It does not prove the clip is decodable, visually correct, or suitable for your use. Inspect the MP4 before publishing it. If a download is interrupted, treat its partial file as incomplete and rerun into a new filename.

4. Add reference frames after the first call works

Keep the same endpoint, model, task type, and version. To guide image-to-video, add input.image_url containing a permitted public first-frame URL. To specify an ending frame as well, also add input.image_tail_url. Retain the motion prompt.

The current Kling 3.0 API documentation describes durations from 3 through 15 seconds. The five-second example above is within that range. The site selector's preset choices are not the API's entire duration range. Expand to audio or multi-shot requests using the Kling 3.0 parameter documentation, since those settings have version-specific constraints.

FAQ

Does a successful POST mean the video is ready?

No. Persist data.task_id, poll until completed, then read data.output.video.

Should I look for video_url?

The current Kling 3.0 documentation specifies the compact output.video field. A parser copied from a different model or earlier response example may read the wrong field.

What should I do after a client timeout?

Continue reading the saved task ID. A timeout does not establish whether a create was accepted or whether a generation failed. Only submit another generation after reconciling the original.

Continue with the Kling 3.0 model page. For an image-focused walkthrough, see the existing Kling image-to-video Python tutorial.

Sources checked September 17, 2026: Kling 3.0 API documentation and the model's request guide.