> ## Documentation Index
> Fetch the complete documentation index at: https://mixroute.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# 查詢 Kling 任務
> 查詢 Kling 閘道器任務狀態、影片輸出與失敗原因。
四類 Kling 生成介面共用本查詢端點，使用提交響應中的 `data.id`。
## 查詢引數
| 欄位 | 型別 | 必填 | 說明 |
| - | - | - | - |
| `task_ids` | string | 是 | 一個 MixRoute 閘道器任務 ID，例如 task\_example。雖然引數名為複數，每次只能傳一個值，不支援逗號分隔批次查詢。 |
不支援 `external_task_ids`。上游 ID、回撥 ID 或自定義 external\_task\_id 不能替代閘道器任務 ID。
```bash theme={null}
curl --get "https://api.mixroute.ai/kling/tasks" \
--header "Authorization: Bearer $MIXROUTE_API_KEY" \
--data-urlencode "task_ids=$KLING_TASK_ID"
```
## 響應
```json theme={null}
{
"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 和非零 code。 |
| `request_id` | string | 否 | 本次 HTTP 請求的排查標識，與任務 ID 不同；如有返回應保留。 |
| `data` | object\[] | 是 | 查詢結果陣列；按 id 找到所查詢的任務。 |
| `data[].id` | string | 是 | 閘道器任務 ID。 |
| `data[].status` | string | 是 | submitted、processing、succeeded 或 failed，以閘道器狀態為準。 |
| `data[].message` | string | 否 | 任務失敗時的原因；與外層請求 message 區分。 |
| `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 繼續查詢，不要自動建立替代任務。
## 錯誤處理
```json theme={null}
{
"code": 400,
"message": "[gateway] unsupported model on this endpoint",
"request_id": "REQUEST_ID"
}
```
同時檢查 HTTP 狀態與 JSON `code`。`message` 以 `[gateway]` 開頭表示閘道器拒絕，例如模型與端點不匹配、空提示詞或批次查詢；上游拒絕會保留上游錯誤資訊和請求 ID。保留 `request_id` 與任務 ID 便於排查。
## Python 完整流程
安裝 `requests`，設定 `MIXROUTE_API_KEY`。首次執行提交一項任務；設定 `KLING_TASK_ID` 後只查詢已有任務。指令碼不自動重試 POST，超時後先在控制台確認任務。
```python theme={null}
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 是第三方媒體地址，可能帶有防盜鏈簽名或有效期；成功後及時儲存影片，不要把結果連結當作永久儲存。不要用帶有 Authorization 的 session 下載它，也不要將 API Key 拼到 URL 中。
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.