kie-ai-generation

name kie-ai-generation description 在开发需要调用 Kie.ai（kie.ai）生成视频/图片的 API 服务时使用。本 Skill 提供统一的“创建任务→查询结果/回调”工作流、模型参数对照、后端封装最佳实践（鉴权、重试、轮询、回调验签/防重、解析 resultJson）、以及可直接复制的 curl/TypeScript/Python 示例代码。适用于 Next.js/Node/Flask 等服务端集成。 Kie.ai 生成任务 Skill（Text-to-Video / Image/Media） 目标：把 Kie.ai 的“异步任务式生成”封装成你项目里稳定、可观测、可扩展的 API： CreateTask → Poll/Callback → Parse Result → Store/Return URLs 你应该在什么时候触发这个 Skill 当你在做以下任一事情时就用它： 写后端接口： POST /api/generate/video 、 GET /api/tasks/:id 之类 在 Next.js / Express / Flask 中对接 Kie.ai 需要支持多个模型：Kling v2.1、Wan 2.2 A14B Turbo、Sora 2 Pro（文档给的 model 标识不同） 需要处理：轮询、回调、失败码、限流、余额不足、参数校验失败等 需要把 resultJson 解析成可用的 resultUrls[] 核心事实（Kie.ai API 形态） Kie.ai 的生成是 两步异步任务 ： 创建任务 POST https://api.kie.ai/api/v1/jobs/createTask 返回 data.taskId 2. 查询任务状态/结果 GET https://api.kie.ai/api/v1/jobs/recordInfo?taskId=... 返回 data.state ： waiting | success | fail 成功时 data.resultJson 是一个 JSON 字符串 ，里面通常是： { "resultUrls": ["...mp4"] } 鉴权统一为： Authorization: Bearer YOUR_API_KEY Content-Type: application/json 建议你在自己项目里定义的统一抽象（强烈建议） 1) 统一的 Task 记录（数据库/缓存都行） 建议最少字段： id （你自己的 task id，或直接用 kie taskId） provider ： "kie" model ：字符串（如 kling/v2-1-master-text-to-video ） state ： waiting | success | fail input ：你请求的 input（保存一份，便于重放/排查） result_urls ：字符串数组 fail_code 、 fail_msg created_at 、 updated_at 不要只把 Kie 的 taskId 直接吐给前端不管了——你最终会需要可观测与可控（重试/限流/回调防重/对账）。 2) 统一的业务流程 create ：创建任务成功后立刻把 taskId 存库，状态为 waiting resolve ：轮询或回调到来时更新状态 deliver ：成功时返回 resultUrls ，失败时返回可读错误 + failCode 模型与参数对照（按你提供的文档） 统一接口都是同一个 createTask ，差异仅在 model 与 input 字段结构。 A) Kling V2.1 Master Text To Video model : kling/v2-1-master-text-to-video input ： prompt (string, required, <=5000) duration ("5"|"10", optional, default "5") aspect_ratio ("16:9"|"9:16"|"1:1", optional) negative_prompt (string <=500, optional) cfg_scale (number 0-1 step 0.1, optional) B) Wan 2.2 A14b Text To Video Turbo model : wan/2-2-a14b-text-to-video-turbo input ： prompt (string, required, <=5000) resolution ("480p"|"580p"|"720p", optional, default "720p") aspect_ratio ("16:9"|"9:16"|"1:1", optional) enable_prompt_expansion (boolean, optional) seed (number 0..2147483647, optional) acceleration ("none"|"regular", optional, default "none") C) Sora 2 Pro Text To Video model : sora-2-pro-text-to-video input ： prompt (string, required, <=10000) aspect_ratio ("portrait"|"landscape", optional) n_frames ("10"|"15", optional) // 文档描述像“10s/15s” size ("standard"|"high", optional) remove_watermark (boolean, optional, default true) 你必须处理好的坑（非常常见） 1) param 与 resultJson 是“JSON 字符串” 查询接口返回里： data.param ： "{\"model\":\"...\",\"input\":{...}}" data.resultJson ： "{\"resultUrls\":[\"https://...\"]}" 你需要对它们再 JSON.parse 一次才能当对象用。 2) 轮询策略不要太猛（避免 429） 推荐策略（可按你业务调整）： 前 30 秒：每 2 秒一次 30~120 秒：每 5 秒一次 2 分钟后：每 10~15 秒一次 超时：例如 10 分钟仍 waiting ，就标记为 timeout （你自己的状态），允许用户稍后刷新/重试 3) 回调（callBackUrl）一定要“防重 + 验证来源” Kie 会在任务成功/失败时 POST 你的回调地址，内容结构与 Query API 一致。 你必须做： 幂等 ：同一个 taskId 的回调多次到来，不重复写入/不重复触发下游 校验 ：至少校验 taskId 是否存在于你库中、是否属于当前用户/订单 不要信任回调的 resultUrls ：你可以接受回调后再调用一次 Query 做最终确认（高安全场景） 文档未给出签名字段时，你只能做“业务校验 + 限制回调入口 + 记录原始请求体 + 可选二次查询确认”。 4) API Key 安全 只能在服务端使用（Node/Next API route/Edge Function/Flask），不要暴露给前端 存在环境变量： KIE_API_KEY 日志里打码： Bearer sk-**** 5) 错误码要映射成“可行动”的业务错误 常见： 401 ：API Key 无效 402 ：余额不足（应提示充值/降级到便宜模型） 422 ：参数校验失败（把你发过去的 model/input 记录下来便于排查） 429 ：限流（指数退避重试） 500 ：服务内部错误（重试 + 告警） curl 示例（可直接复制） 注意：以下示例里把 YOUR_API_KEY 替换为你自己的 key。 1) 创建 Kling v2.1 任务 curl -X POST "https://api.kie.ai/api/v1/jobs/createTask" \ -H "Authorization: Bearer YOUR_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "kling/v2-1-master-text-to-video", "input": { "prompt": "A cinematic close-up of a ramen chef plating noodles, steam swirling, warm neon reflections, shallow depth of field, handheld camera micro-shake.", "duration": "5", "aspect_ratio": "16:9", "negative_prompt": "blur, distort, low quality, watermark", "cfg_scale": 0.5 } }' 2) 创建 Wan 2.2 Turbo 任务 curl -X POST "https://api.kie.ai/api/v1/jobs/createTask" \ -H "Authorization: Bearer YOUR_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "wan/2-2-a14b-text-to-video-turbo", "input": { "prompt": "Drone shot through a rainy alley in Tokyo at night, reflections on asphalt, cinematic lighting, smooth forward motion.", "resolution": "720p", "aspect_ratio": "16:9", "enable_prompt_expansion": false, "seed": 42, "acceleration": "none" } }' 3) 创建 Sora 2 Pro 任务 curl -X POST "https://api.kie.ai/api/v1/jobs/createTask" \ -H "Authorization: Bearer YOUR_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "sora-2-pro-text-to-video", "input": { "prompt": "a happy dog running in the garden, sunlight, slow motion, natural camera tracking", "aspect_ratio": "landscape", "n_frames": "10", "size": "high", "remove_watermark": true } }' 4) 查询任务 curl -X GET "https://api.kie.ai/api/v1/jobs/recordInfo?taskId=YOUR_TASK_ID" \ -H "Authorization: Bearer YOUR_API_KEY" TypeScript（Node/Next.js）推荐封装 目标：你只需要调用 kieCreateTask() 与 kieQueryTask() ，其余都封装好。 type KieModel = | "kling/v2-1-master-text-to-video" | "wan/2-2-a14b-text-to-video-turbo" | "sora-2-pro-text-to-video" ; type CreateTaskResponse = { code : number ; msg : string ; data ?: { taskId : string } }; type RecordInfoResponse = { code : number ; msg : string ; data ?: { taskId : string ; model : string ; state : "waiting" | "success" | "fail" ; param : string ; // JSON string resultJson : string ; // JSON string failCode : string | null ; failMsg : string | null ; costTime : number | null ; completeTime : number | null ; createTime : number ; }; }; const KIE_BASE_URL = "https://api.kie.ai" ; const KIE_API_KEY = process. env . KIE_API_KEY ; function requireApiKey ( ): string { if (! KIE_API_KEY ) throw new Error ( "Missing env: KIE_API_KEY" ); return KIE_API_KEY ; } export async function kieCreateTask ( params : { model: KieModel; input: Record< string , unknown >; callBackUrl?: string ; } ): Promise <{ taskId : string }> { const apiKey = requireApiKey (); const res = await fetch ( ` ${KIE_BASE_URL} /api/v1/jobs/createTask` , { method : "POST" , headers : { Authorization : `Bearer ${apiKey} ` , "Content-Type" : "application/json" , }, body : JSON . stringify (params), }); if (!res. ok ) { const text = await res. text (). catch ( () => "" ); throw new Error ( `Kie createTask failed: HTTP ${res.status} ${text} ` ); } const json = ( await res. json ()) as CreateTaskResponse ; if (json. code !== 200 || !json. data ?. taskId ) { throw new Error ( `Kie createTask error: code= ${json.code} msg= ${json.msg} ` ); } return { taskId : json. data . taskId }; } export async function kieQueryTask ( taskId : string ): Promise <{ state : "waiting" | "success" | "fail" ; resultUrls ?: string []; failCode ?: string | null ; failMsg ?: string | null ; raw : RecordInfoResponse ; }> { const apiKey = requireApiKey (); const url = new URL ( ` ${KIE_BASE_URL} /api/v1/jobs/recordInfo` ); url. searchParams . set ( "taskId" , taskId); const res = await fetch (url, { method : "GET" , headers : { Authorization : `Bearer ${apiKey} ` }, }); if (!res. ok ) { const text = await res. text (). catch ( () => "" ); throw new Error ( `Kie recordInfo failed: HTTP ${res.status} ${text} ` ); } const json = ( await res. json ()) as RecordInfoResponse ; if (json. code !== 200 || !json. data ) { throw new Error ( `Kie recordInfo error: code= ${json.code} msg= ${json.msg} ` ); } const { state, resultJson, failCode, failMsg } = json. data ; if (state === "success" ) { // resultJson is JSON string, parse once more const parsed = safeJsonParse (resultJson) as { resultUrls ?: string [] } | null ; const resultUrls = parsed?. resultUrls ?? []; return { state, resultUrls, raw : json }; } return { state, failCode, failMsg, raw : json }; } function safeJsonParse ( value : string ): unknown | null { try { return JSON . parse (value); } catch { return null ; } } 业务层建议（Next.js API Route 伪代码） POST /api/video ：调用 kieCreateTask → 写库 waiting → 返回你的 taskId GET /api/tasks/:id ：读库；如 still waiting 可选择触发一次 kieQueryTask 刷新 Python（Flask/FastAPI）最小封装 import os import json import requests from typing import Any , Dict , Optional , List KIE_BASE_URL = "https://api.kie.ai" KIE_API_KEY = os.environ.get( "KIE_API_KEY" ) def _require_key () -> str : if not KIE_API_KEY: raise RuntimeError( "Missing env: KIE_API_KEY" ) return KIE_API_KEY def kie_create_task ( model: str , input_data: Dict [ str , Any ], callback_url: Optional [ str ] = None ) -> str : api_key = _require_key() payload: Dict [ str , Any ] = { "model" : model, "input" : input_data} if callback_url: payload[ "callBackUrl" ] = callback_url