How to Use the Kling Image-to-Video API with Python

To use the Kling image-to-video API from Python, send a source image URL and a motion prompt to PiAPI's POST /api/v1/task endpoint. Save data.task_id, then query GET /api/v1/task/{task_id} until processing finishes. For the Kling 3.0 family used here, a completed response places the video URL in data.output.video.

This guide uses requests and the documented 3.0-turbo version for a single image-guided clip. The create request and the polling request are separate operations: an accepted task is not yet a finished video.

Verification, September 7, 2026: the request was checked against PiAPI's published Kling 3.0 schema, and response handling was checked with offline fixtures. Not live-tested: no generation API key was available for this article, so no video-generation request was submitted.

Prepare your image and API key

Use Python 3.10 or newer and install requests:

python3 -m venv .venv
.venv/bin/python -m pip install requests

Create an API key in the PiAPI dashboard and provide it to your process as PIAPI_API_KEY. Keep it in your environment or secret store. The code reads the key at runtime and sends it in the X-API-Key header.

Set PIAPI_IMAGE_URL to an HTTPS URL that returns the image itself. PiAPI needs to fetch it independently, without your browser session or local filesystem. A private drive sharing page and a path such as ./photo.png are not equivalent to an accessible image URL.

For a public sample input, this PiAPI-hosted image returned HTTP 200 with an image/png content type when checked for this article:

export PIAPI_IMAGE_URL='https://piapi.ai/models/flux/input_example.png'

This is the source image, not a generated result. If you use your own signed URL, give the service enough time to retrieve it before the URL expires. Checking it from your machine confirms your access; it does not prove that a generation worker can fetch it.

Choose the documented version and fields

The dedicated Kling 3.0 API documentation is the reference for this example. Its current request-schema enum explicitly includes 3.0-turbo; its prose and some examples also describe 3.0. This article uses the enum-listed value, without treating those inconsistent examples as a verified request for another version.

The schema documents durations from 3 to 15 seconds and modes std and pro. These are request settings, not evidence of a generated result. It says aspect_ratio is ignored when a starting image is supplied, so this image-to-video example omits it. The Turbo documentation excludes audio generation and custom multi-shot input; neither appears in this request.

Create a task and retain its ID

The following three Python blocks form one file, example.py. The first defines the payload and submits it once. It checks the HTTP response, any supplied API status code, the data object, and the returned task ID before polling.

import argparse
import math
import os
import sys
import time
from urllib.parse import quote, urlsplit

import requests


API_BASE = "https://api.piapi.ai/api/v1"
PROMPT = "The flowers move gently in a light breeze. Keep the lettering and camera steady."


def request_payload(image_url):
    if urlsplit(image_url).scheme != "https" or not urlsplit(image_url).netloc:
        raise ValueError("PIAPI_IMAGE_URL must be an accessible HTTPS image URL")
    return {
        "model": "kling",
        "task_type": "video_generation",
        "input": {
            "image_url": image_url,
            "prompt": PROMPT,
            "version": "3.0-turbo",
            "mode": "std",
            "duration": 5,
        },
        "config": {"service_mode": "public"},
    }


def response_data(response):
    if response.status_code != 200:
        raise RuntimeError(f"PiAPI returned HTTP {response.status_code}")
    try:
        body = response.json()
    except ValueError:
        raise RuntimeError("PiAPI returned a non-JSON response") from None
    if not isinstance(body, dict):
        raise RuntimeError("PiAPI returned an unexpected response envelope")
    # The published get-task example has data + timestamp, without a code field.
    if "code" in body and body["code"] != 200:
        raise RuntimeError("PiAPI returned a non-success API code")
    if not isinstance(body.get("data"), dict):
        raise RuntimeError("PiAPI response is missing the task data object")
    return body["data"]


def create_task(client, image_url):
    payload = request_payload(image_url)
    try:
        response = client.post(
            f"{API_BASE}/task", json=payload, timeout=(10, 30), allow_redirects=False
        )
    except requests.RequestException:
        raise RuntimeError(
            "Submission was not confirmed. Check task history before creating another task."
        ) from None
    data = response_data(response)
    task_id = data.get("task_id")
    if not isinstance(task_id, str) or not task_id.strip():
        raise RuntimeError("Create response has no task_id; check task history before resubmitting")
    return task_id

An interrupted submission has an uncertain outcome: the service may have accepted it even though your client did not receive confirmation. The example does not automatically repeat POST. Inspect task history before creating another task after that error.

Poll for the completed video

Add this block next. pending and processing continue polling; failed stops with a bounded error message. completed requires an actual HTTP or HTTPS video URL. An unexpected status or response shape stops for inspection.

def wait_for_video(client, task_id, max_wait=600, poll_interval=5):
    if not all(math.isfinite(value) and value > 0 for value in (max_wait, poll_interval)):
        raise ValueError("Polling budget and interval must be finite and positive")
    deadline = time.monotonic() + max_wait
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError("Polling budget expired; the remote task may still be running")
        try:
            response = client.get(
                f"{API_BASE}/task/{quote(task_id, safe='')}",
                timeout=(min(10, remaining), min(30, remaining)),
                allow_redirects=False,
            )
        except requests.RequestException:
            raise RuntimeError("Polling request interrupted; resume with the saved task ID") from None
        data = response_data(response)
        status = data.get("status")
        if status == "completed":
            output = data.get("output")
            video = output.get("video") if isinstance(output, dict) else None
            if not isinstance(video, str) or not video.strip():
                raise RuntimeError("Completed task is missing data.output.video")
            url = urlsplit(video)
            if url.scheme not in {"https", "http"} or not url.netloc:
                raise RuntimeError("Task returned an invalid video URL")
            return video
        if status == "failed":
            error = data.get("error")
            code = error.get("code") if isinstance(error, dict) else None
            code = code if isinstance(code, int) else "unavailable"
            raise RuntimeError(f"Kling task failed; API error code: {code}")
        if status not in {"pending", "processing"}:
            raise RuntimeError("Task returned an unexpected status; inspect the saved task")
        time.sleep(min(poll_interval, max(0, deadline - time.monotonic())))

The ten-minute polling budget and five-second interval are local example settings. They do not predict generation time. The budget stops new polls after the deadline; requests connection and read timeouts are not a strict wall-clock deadline for an in-flight request.

Exhausting this budget or interrupting a polling request does not cancel the remote task. Keep the task ID so you can query that same job later. The Get Task documentation contains the Kling 3.0 response example: use its output.video field, rather than assuming older Kling output layouts apply.

Run the file or resume a task

Finish example.py with the command-line entry point:

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--task-id", help="Query an existing task without creating another")
    parser.add_argument("--max-wait", type=float, default=600)
    args = parser.parse_args()
    key = os.environ.get("PIAPI_API_KEY")
    image_url = os.environ.get("PIAPI_IMAGE_URL")
    if not key:
        parser.error("Set PIAPI_API_KEY in your environment")
    if not args.task_id and not image_url:
        parser.error("Set PIAPI_IMAGE_URL, or use --task-id to resume polling")
    if not math.isfinite(args.max_wait) or args.max_wait <= 0:
        parser.error("--max-wait must be finite and positive")
    with requests.Session() as client:
        client.headers.update({"X-API-Key": key, "Content-Type": "application/json"})
        task_id = args.task_id or create_task(client, image_url)
        print(f"Task ID: {task_id}", flush=True)
        print(wait_for_video(client, task_id, max_wait=args.max_wait))


if __name__ == "__main__":
    try:
        main()
    except (RuntimeError, ValueError, TimeoutError) as error:
        print(str(error), file=sys.stderr)
        raise SystemExit(1)

With the key and image URL set, run:

.venv/bin/python example.py

The program prints the accepted task ID first and, if processing completes with the documented output, the video URL. To query an existing task, replace YOUR_SAVED_TASK_ID below with that ID. This command does not submit another generation:

.venv/bin/python example.py --task-id YOUR_SAVED_TASK_ID --max-wait 600

Common failures

The script avoids printing raw response bodies, headers, or upstream exception text. Those can contain private inputs or service details. In an application, retain task IDs in durable storage so a process restart does not lose the link to submitted work.

Frequently asked questions

Which image field should I send?

Use input.image_url for the starting image. The dedicated schema also documents image_tail_url for an ending image, but this example uses only one starting frame.

Why does the first response have no finished video?

Generation is asynchronous. A create response can return a task ID while the task is still pending. Poll that ID and inspect both the terminal status and output.

Can I reuse this for text-to-video?

The documented endpoint supports text-to-video too. Remove the image input, adapt the prompt and payload validation, and check the applicable aspect-ratio and version settings. This image-specific function deliberately requires an image URL.

Can I switch to another Kling version?

Check its request and output documentation before changing the version string. Shared endpoint paths do not mean identical model options or identical result fields.

For current integration options, visit the PiAPI Kling API page. Keep the Kling 3.0 request reference and Get Task reference alongside your implementation.