Kling 4 API接入状态与开发指南

# KIE provider API: use your own KIE key, not a kling4.io key.
# Submitting a generation consumes provider credits.
curl --fail-with-body https://api.kie.ai/api/v1/jobs/createTask \
  -H "Authorization: Bearer $KIE_API_KEY" \
  -H "Content-Type: application/json" \
  --data '{
  "model": "kling-3.0/video",
  "input": {
    "prompt": "One continuous shot of a ceramic cup on a wooden table. Soft morning light from the left. A slow camera push-in as a thin ribbon of steam rises. Quiet room tone, no speech or text.",
    "mode": "std",
    "duration": "5",
    "sound": true,
    "multi_shots": false,
    "aspect_ratio": "16:9"
  }
}'

# Copy data.taskId from the response; poll about every 5 seconds.
curl --fail-with-body \
  'https://api.kie.ai/api/v1/jobs/recordInfo?taskId=YOUR_TASK_ID' \
  -H "Authorization: Bearer $KIE_API_KEY"

# When data.state is success, parse data.resultJson (a JSON string).
# Download the URL in resultUrls[0] before the provider link expires.
curl --fail --location 'VIDEO_URL_FROM_RESULT' --output kling-video.mp4

// Node.js 18+. Set KIE_API_KEY in your server environment.
// This calls KIE directly and consumes KIE credits.
const apiKey = process.env.KIE_API_KEY;
if (!apiKey) throw new Error("Set KIE_API_KEY first");
const base = "https://api.kie.ai/api/v1/jobs";
const headers = {
  Authorization: "Bearer " + apiKey,
  "Content-Type": "application/json",
};
async function request(url, options = {}) {
  const response = await fetch(url, {
    ...options, headers, signal: AbortSignal.timeout(30000),
  });
  const json = await response.json();
  if (!response.ok || json.code !== 200) {
    throw new Error(json.msg || "Provider request failed");
  }
  return json.data;
}
// Submit once. An unknown POST outcome may already have created a paid job.
const created = await request(base + "/createTask", {
  method: "POST", body: JSON.stringify({
  "model": "kling-3.0/video",
  "input": {
    "prompt": "One continuous shot of a ceramic cup on a wooden table. Soft morning light from the left. A slow camera push-in as a thin ribbon of steam rises. Quiet room tone, no speech or text.",
    "mode": "std",
    "duration": "5",
    "sound": true,
    "multi_shots": false,
    "aspect_ratio": "16:9"
  }
}),
});
if (!created.taskId) throw new Error("Missing taskId");
console.log("Save this task ID:", created.taskId);
let videoUrl;
for (let attempt = 0; attempt < 180; attempt++) {
  await new Promise(resolve => setTimeout(resolve, 5000));
  const task = await request(base + "/recordInfo?taskId=" +
    encodeURIComponent(created.taskId));
  if (task.state === "fail") throw new Error(task.failMsg || "Generation failed");
  if (task.state === "success") {
    videoUrl = JSON.parse(task.resultJson || "{}").resultUrls?.[0];
    if (!videoUrl) throw new Error("Missing result URL");
    break;
  }
}
if (!videoUrl) throw new Error("Still processing. Resume polling the saved task ID.");
console.log(videoUrl);

# Python 3.9+. Uses only the standard library.
# Calls KIE directly and consumes provider credits.
import json, os, time, urllib.request

key = os.environ["KIE_API_KEY"]
base = "https://api.kie.ai/api/v1/jobs"
def request(path, body=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(base + path, data=data, headers={
        "Authorization": "Bearer " + key,
        "Content-Type": "application/json",
    })
    with urllib.request.urlopen(req, timeout=30) as response:
        result = json.load(response)
    if result.get("code") != 200:
        raise RuntimeError(result.get("msg", "Provider request failed"))
    return result["data"]

# Submit once; do not blindly retry a timed-out creation request.
payload = json.loads(r'''{
  "model": "kling-3.0/video",
  "input": {
    "prompt": "One continuous shot of a ceramic cup on a wooden table. Soft morning light from the left. A slow camera push-in as a thin ribbon of steam rises. Quiet room tone, no speech or text.",
    "mode": "std",
    "duration": "5",
    "sound": true,
    "multi_shots": false,
    "aspect_ratio": "16:9"
  }
}''')
created = request("/createTask", payload)
task_id = created["taskId"]
print("Save this task ID:", task_id)
for attempt in range(180):
    time.sleep(5)
    task = request("/recordInfo?taskId=" + task_id)
    if task["state"] == "fail":
        raise RuntimeError(task.get("failMsg") or "Generation failed")
    if task["state"] == "success":
        urls = json.loads(task.get("resultJson") or "{}").get("resultUrls", [])
        if not urls:
            raise RuntimeError("Missing result URL")
        print(urls[0])
        break
else:
    raise TimeoutError("Resume polling the saved task ID; do not resubmit")