Klingタスクの照会

curl --request GET \
  --url https://api.mixroute.ai/kling/tasks

<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://api.mixroute.ai/kling/tasks",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "GET",
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}

package main

import (
	"fmt"
	"net/http"
	"io"
)

func main() {

	url := "https://api.mixroute.ai/kling/tasks"

	req, _ := http.NewRequest("GET", url, nil)

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(string(body))

}

require 'uri'
require 'net/http'

url = URI("https://api.mixroute.ai/kling/tasks")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Get.new(url)

response = http.request(request)
puts response.read_body

Klingタスクの照会

curl --request GET \
  --url https://api.mixroute.ai/kling/tasks

<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://api.mixroute.ai/kling/tasks",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "GET",
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}

package main

import (
	"fmt"
	"net/http"
	"io"
)

func main() {

	url := "https://api.mixroute.ai/kling/tasks"

	req, _ := http.NewRequest("GET", url, nil)

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(string(body))

}

require 'uri'
require 'net/http'

url = URI("https://api.mixroute.ai/kling/tasks")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Get.new(url)

response = http.request(request)
puts response.read_body

Klingの4つの生成エンドポイントは、すべてこの照会エンドポイントを共有します。送信時のレスポンスに含まれるdata.idを使用してください。

クエリパラメータ

フィールド	型	必須	説明
`task_ids`	string	はい	task_exampleなどのMixRouteゲートウェイタスクIDを1つ指定します。名前は複数形ですが、指定できるIDは1つのみです。カンマ区切りのバッチには対応していません。

external_task_idsには対応していません。上流のID、コールバックID、カスタムのexternal_task_id値を、ゲートウェイのタスクIDの代わりに使用することはできません。

curl --get "https://api.mixroute.ai/kling/tasks" \
  --header "Authorization: Bearer $MIXROUTE_API_KEY" \
  --data-urlencode "task_ids=$KLING_TASK_ID"

レスポンス

{
  "code": 0,
  "message": "",
  "data": [
    {
      "id": "task_example",
      "status": "succeeded",
      "outputs": [
        {
          "type": "video",
          "url": "https://example.com/generated.mp4",
          "duration": "5.041"
        }
      ]
    }
  ]
}

フィールド	型	必須	説明
`code`	integer	はい	0は照会リクエストの成功を意味し、生成の完了を意味するものではありません。上流側の業務エラーでは、HTTP 200と0以外のcodeが返される場合があります。
`request_id`	string	いいえ	このHTTPリクエストのトラブルシューティングに使用する識別子で、タスクIDとは異なります。含まれている場合は保持してください。
`data`	object[]	はい	照会結果の配列。自分のidと一致するタスクを選択してください。
`data[].id`	string	はい	ゲートウェイのタスクID。
`data[].status`	string	はい	ゲートウェイが判定するsubmitted、processing、succeeded、failedのいずれか。
`data[].message`	string	いいえ	タスクの失敗理由。外側のリクエストメッセージとは異なります。
`data[].outputs`	object[]	いいえ	成功時の動画出力。type=videoのエントリからurlを読み取ってください。id、watermark_url、durationが含まれる場合もあります。
`data[].outputs[].duration`	string / number	いいえ	実際に生成された秒数。小数を含む場合があります。整数に切り捨てないでください。
`data[].billing`	object[]	いいえ	任意で返される上流の課金情報。必ず返されるとは限らず、最終的なMixRouteの請求額を示す証拠ではありません。
`data[].create_time` / `update_time`	integer	いいえ	含まれる場合、上流側のUnixタイムスタンプ（ミリ秒単位）。
`data[].external_id`	string	いいえ	存在する場合に上流から転送される、カスタムの業務用識別子。

状態の処理

状態	操作
`submitted`	受け付け済み。ポーリングを続けてください。
`processing`	処理中です。ポーリングを続けてください。
`succeeded`	ポーリングを停止し、動画URLを読み取ります。
`failed`	ポーリングを停止し、messageを確認してください。

約15秒間隔でポーリングし、レート制限や一時的なネットワークエラーが発生した場合は待機時間を延ばします。レスポンスにはゲートウェイが取得した最新の状態が含まれ、数秒の遅延が生じる場合があります。クライアントの待機タイムアウトはサーバー側のタスク失敗ではありません。代わりのタスクを作成せず、IDを保持して問い合わせを続けてください。

エラー処理

{
  "code": 400,
  "message": "[gateway] unsupported model on this endpoint",
  "request_id": "REQUEST_ID"
}

HTTPのステータスとJSONのコードの両方を確認してください。メッセージの接頭辞[gateway]は、モデルとエンドポイントの不一致、空のプロンプト、一括照会など、ゲートウェイによる拒否を示します。上流による拒否では、上流のエラー詳細とリクエストIDが保持されます。トラブルシューティング用にrequest_idとタスクIDを保存してください。

Pythonの完全なワークフロー

requestsをインストールし、MIXROUTE_API_KEYを設定してください。初回実行ではタスクを1件送信します。KLING_TASK_IDを設定すると、代わりに既存のタスクを照会します。POSTは自動的に再試行されません。送信がタイムアウトした場合は、まずコンソールを確認してください。

import os
import time
from urllib.parse import urlparse
import requests

BASE_URL = "https://api.mixroute.ai/kling"

def parse_response(response):
    try:
        result = response.json()
    except ValueError as exc:
        raise RuntimeError(f"HTTP {response.status_code}: invalid JSON response") from exc
    if not isinstance(result, dict):
        raise RuntimeError(f"HTTP {response.status_code}: expected a JSON object")
    code = result.get("code")
    if not 200 <= response.status_code < 300 or type(code) is not int or code != 0:
        raise RuntimeError(
            f"HTTP {response.status_code}; code={code}; "
            f"request_id={result.get('request_id', '')}; "
            f"message={result.get('message', '')}"
        )
    return result

def create_video(session):
    response = session.post(
        BASE_URL + "/text-to-video/kling-3.0",
        json={
            "prompt": "A ceramic mug slowly rotates on a white tabletop.",
            "settings": {"resolution": "720p", "duration": 5,
                         "aspect_ratio": "16:9", "audio": "off"},
        },
        timeout=120,
        allow_redirects=False,
    )
    task_id = parse_response(response).get("data", {}).get("id")
    if not task_id:
        raise RuntimeError("Submission returned no task ID")
    return task_id

def wait_for_video(session, task_id, timeout_seconds=1800):
    deadline = time.monotonic() + timeout_seconds
    delay = 15
    while time.monotonic() < deadline:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            break
        try:
            response = session.get(
                BASE_URL + "/tasks", params={"task_ids": task_id},
                timeout=min(30, remaining), allow_redirects=False,
            )
        except (requests.Timeout, requests.ConnectionError):
            delay = min(delay * 2, 60)
        else:
            if response.status_code in (429, 500, 502, 503, 504):
                delay = min(delay * 2, 60)
            else:
                tasks = parse_response(response).get("data")
                if not isinstance(tasks, list):
                    raise RuntimeError("Invalid task response")
                task = next((item for item in tasks if item.get("id") == task_id), None)
                if task is None:
                    raise RuntimeError("Task not found in response")
                status = task.get("status")
                if status == "succeeded":
                    videos = [item for item in task.get("outputs", [])
                              if item.get("type") == "video" and item.get("url")]
                    if not videos:
                        raise RuntimeError("Successful task has no video output")
                    for item in videos:
                        parsed = urlparse(item["url"])
                        if parsed.scheme not in ("https", "http") or not parsed.netloc:
                            raise RuntimeError("Invalid output URL")
                    return videos
                if status == "failed":
                    raise RuntimeError(task.get("message") or task)
                if status not in ("submitted", "processing"):
                    raise RuntimeError("Unknown task status: " + str(status))
                delay = 15
        remaining = deadline - time.monotonic()
        if remaining > 0:
            time.sleep(min(delay, remaining))
    raise TimeoutError("Keep task ID and query later: " + task_id)

if __name__ == "__main__":
    with requests.Session() as session:
        session.headers["Authorization"] = "Bearer " + os.environ["MIXROUTE_API_KEY"]
        task_id = os.environ.get("KLING_TASK_ID") or create_video(session)
        print("task_id:", task_id, flush=True)
        for video in wait_for_video(session, task_id):
            print(video["url"], video.get("duration"))

結果のURLは第三者のメディアを指し、有効期限付き署名や保存期間の制限が適用される場合があります。URLを永続ストレージとみなさず、成功した出力は速やかに保存してください。認証済みセッションを使ってダウンロードしたり、URLにAPIキーを追加したりしないでください。

人気モデル

OpenAI SDKの使い方

システムAPI

テキストシリーズ

意思決定モデル

埋め込みシリーズ

画像シリーズ

動画シリーズ

その他

Klingタスクの照会

クエリパラメータ

レスポンス

状態の処理

エラー処理

Pythonの完全なワークフロー

​クエリパラメータ

​レスポンス

​状態の処理

​エラー処理

​Pythonの完全なワークフロー

クエリパラメータ

レスポンス

状態の処理

エラー処理

Pythonの完全なワークフロー