> ## Documentation Index > Fetch the complete documentation index at: https://docs.mixroute.ai/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 中。