如何使用 Kling v3 API：完整 Python 教程 2026

Kling 3.0 于 2026 年 2 月 4 日正式发布，三个关键数字：首帧生成延迟约 8–15 秒，5 秒标准视频单次生成成本约 $0.14–$0.28（取决于分辨率和套餐），社区基准测试中运动连贯性评分达 8.3/10（对比同期主流模型）。本文不讲概念，只给你能直接跑起来的代码。

---

目录

1. 前置条件
2. 身份认证与环境配置
3. 基础实现：文本生成视频
4. 图像生成视频（Image-to-Video）
5. 生产级封装：异步轮询 + 重试
6. API 参数完整参考表
7. 错误处理
8. 性能与成本对照表
9. 结语

---

1. 前置条件 {#prerequisites}

账号与 API Key

1. 前往 klingai.com 注册开发者账号
2. 进入 Console → API Keys，创建新 Key
3. 记录 access_key_id 和 access_key_secret（注意：不是单一 token，Kling 使用 JWT 签名认证）

Python 环境要求

• Python 3.9+（代码中使用了 match 语句需 3.10+，否则改用 if/elif）
• 以下库：

# 安装所有依赖，固定版本避免 breaking change
pip install requests==2.31.0 PyJWT==2.8.0 python-dotenv==1.0.1 pillow==10.3.0

环境变量配置

在项目根目录创建 .env 文件：

# .env — 不要提交到 git
KLING_ACCESS_KEY_ID=your_access_key_id_here
KLING_ACCESS_KEY_SECRET=your_access_key_secret_here

---

2. 身份认证与环境配置 {#auth-setup}

Kling API 使用 JWT（JSON Web Token） 而非简单 Bearer token。每次请求前需要用 HMAC-SHA256 签名生成有时效的 token，默认有效期 30 分钟。

# auth.py — JWT 签名生成模块
import jwt
import time
import os
from dotenv import load_dotenv

load_dotenv()

def generate_kling_jwt(expire_seconds: int = 1800) -> str:
    """
    生成 Kling API 所需的 JWT token。
    
    为什么要用 JWT 而不是直接传 API Key？
    Kling 的安全模型要求每个请求携带有时效的签名 token，
    防止 API Key 泄露后被长期滥用。
    
    Args:
        expire_seconds: token 有效期，默认 1800 秒（30 分钟）
        
    Returns:
        str: 签名后的 JWT token
    """
    access_key_id = os.getenv("KLING_ACCESS_KEY_ID")
    access_key_secret = os.getenv("KLING_ACCESS_KEY_SECRET")
    
    if not access_key_id or not access_key_secret:
        raise EnvironmentError(
            "缺少环境变量 KLING_ACCESS_KEY_ID 或 KLING_ACCESS_KEY_SECRET"
        )
    
    now = int(time.time())
    
    payload = {
        "iss": access_key_id,       # issuer: 你的 access key id
        "exp": now + expire_seconds, # 过期时间戳
        "nbf": now - 5,              # not before: 留 5 秒时钟偏差容错
    }
    
    # HS256 是 Kling 文档指定的算法
    token = jwt.encode(
        payload,
        access_key_secret,
        algorithm="HS256",
        headers={"alg": "HS256", "typ": "JWT"},
    )
    
    return token


def get_headers() -> dict:
    """返回每次请求所需的完整 HTTP headers"""
    return {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {generate_kling_jwt()}",
    }


# 快速验证：直接运行此文件测试认证是否正常
if __name__ == "__main__":
    token = generate_kling_jwt()
    print(f"Token 生成成功（前50字符）: {token[:50]}...")

---

3. 基础实现：文本生成视频（Text-to-Video） {#basic-t2v}

Kling v3 的 Text-to-Video 端点是 /v1/videos/text2video。流程分两步：提交任务 → 轮询结果（视频生成是异步的，不会立即返回文件）。

# basic_t2v.py — 最小可运行的 T2V 示例
import requests
import time
from auth import get_headers

# Kling API 基础 URL
BASE_URL = "https://api.klingai.com"


def submit_t2v_job(
    prompt: str,
    model_name: str = "kling-v1-6",   # v3 对应的模型标识符
    duration: int = 5,                  # 视频时长，单位秒
    aspect_ratio: str = "16:9",
    cfg_scale: float = 0.5,
) -> str:
    """
    提交文本生成视频任务，返回 task_id。
    
    注意：这里只是"提交"，不是立即得到视频。
    Kling 的生成是异步的，需要用 task_id 轮询状态。
    """
    url = f"{BASE_URL}/v1/videos/text2video"
    
    payload = {
        "model_name": model_name,
        "prompt": prompt,
        "cfg_scale": cfg_scale,        # 提示词遵循程度，0.0-1.0
        "duration": str(duration),     # API 接受字符串格式
        "aspect_ratio": aspect_ratio,
        # negative_prompt 可选，不填则用模型默认值
    }
    
    response = requests.post(url, json=payload, headers=get_headers())
    
    # 非 200 状态码直接抛出，让调用方处理
    response.raise_for_status()
    
    data = response.json()
    
    # 检查业务层面的错误码（HTTP 200 但业务失败的情况）
    if data.get("code") != 0:
        raise RuntimeError(
            f"API 业务错误 code={data.get('code')}: {data.get('message')}"
        )
    
    task_id = data["data"]["task_id"]
    print(f"任务已提交，task_id: {task_id}")
    return task_id


def poll_task(task_id: str, endpoint: str = "text2video", interval: int = 5) -> dict:
    """
    轮询任务状态直到完成或失败。
    
    为什么用轮询而不是 webhook？
    Kling API 目前主要支持轮询模式。
    建议 interval 不低于 5 秒，避免触发限频（429）。
    """
    url = f"{BASE_URL}/v1/videos/{endpoint}/{task_id}"
    
    while True:
        response = requests.get(url, headers=get_headers())
        response.raise_for_status()
        data = response.json()
        
        task_status = data["data"]["task_status"]
        print(f"当前状态: {task_status}")
        
        if task_status == "succeed":
            # 成功：返回完整任务数据，包含视频 URL
            return data["data"]
        elif task_status == "failed":
            error_msg = data["data"].get("task_status_msg", "未知错误")
            raise RuntimeError(f"任务失败: {error_msg}")
        
        # 状态为 "processing" 或 "submitted"，继续等待
        time.sleep(interval)


def download_video(video_url: str, output_path: str) -> None:
    """下载生成的视频到本地文件"""
    response = requests.get(video_url, stream=True)
    response.raise_for_status()
    
    with open(output_path, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    
    print(f"视频已保存到: {output_path}")


# 完整流程示例
if __name__ == "__main__":
    PROMPT = (
        "A lone astronaut walking across a red Martian desert at golden hour, "
        "cinematic wide shot, dust particles floating in backlit sunlight, "
        "photorealistic, 4K"
    )
    
    # 步骤 1：提交任务
    task_id = submit_t2v_job(prompt=PROMPT, duration=5, aspect_ratio="16:9")
    
    # 步骤 2：等待完成
    result = poll_task(task_id, endpoint="text2video")
    
    # 步骤 3：获取视频 URL 并下载
    video_url = result["task_result"]["videos"][0]["url"]
    download_video(video_url, "output_t2v.mp4")

---

4. 图像生成视频（Image-to-Video） {#i2v}

Image-to-Video 端点为 /v1/videos/image2video，需要将图像转为 base64 或提供公开 URL。

# i2v.py — 图像生成视频
import base64
import requests
from pathlib import Path
from auth import get_headers

BASE_URL = "https://api.klingai.com"


def image_to_base64(image_path: str) -> str:
    """
    将本地图像转为 base64 字符串。
    
    Kling 支持两种图像输入方式：
    1. base64 编码字符串（本地文件，隐私更好）
    2. 公开 HTTPS URL（文件必须可公开访问）
    
    文件大小限制：单张图像 ≤ 10MB，支持 jpg/png/webp
    """
    path = Path(image_path)
    if not path.exists():
        raise FileNotFoundError(f"图像文件不存在: {image_path}")
    
    # 根据扩展名确定 MIME type
    mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg",
                ".png": "image/png", ".webp": "image/webp"}
    mime_type = mime_map.get(path.suffix.lower(), "image/jpeg")
    
    with open(image_path, "rb") as f:
        encoded = base64.b64encode(f.read()).decode("utf-8")
    
    # Kling 要求 data URI 格式
    return f"data:{mime_type};base64,{encoded}"


def submit_i2v_job(
    image_path: str,
    prompt: str = "",
    tail_image_path: str = None,    # 可选：指定结尾帧，控制运动方向
    duration: int = 5,
    cfg_scale: float = 0.5,
) -> str:
    """
    提交图像生成视频任务。
    
    tail_image_path：Kling v3 支持首尾帧控制，
    如果同时提供首帧和尾帧，模型会在两张图之间插值生成运动。
    这对于需要精确控制镜头起止的场景非常有用。
    """
    url = f"{BASE_URL}/v1/videos/image2video"
    
    image_data = image_to_base64(image_path)
    
    payload = {
        "model_name": "kling-v1-6",
        "image": image_data,
        "prompt": prompt,
        "cfg_scale": cfg_scale,
        "duration": str(duration),
    }
    
    # 如果提供了尾帧，加入 payload
    if tail_image_path:
        payload["image_tail"] = image_to_base64(tail_image_path)
    
    response = requests.post(url, json=payload, headers=get_headers())
    response.raise_for_status()
    
    data = response.json()
    if data.get("code") != 0:
        raise RuntimeError(f"API 错误 code={data['code']}: {data['message']}")
    
    task_id = data["data"]["task_id"]
    print(f"I2V 任务已提交，task_id: {task_id}")
    return task_id

---

5. 生产级封装：异步轮询 + 重试机制 {#production}

上面的基础代码够用，但不够健壮。生产环境需要处理网络抖动、限频、token 过期等问题。

# kling_client.py — 生产级客户端
import asyncio
import time
import logging
from dataclasses import dataclass
from typing import Optional, Callable
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from auth import get_headers

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

BASE_URL = "https://api.klingai.com"


@dataclass
class VideoGenerationResult:
    task_id: str
    status: str                     # "succeed" | "failed"
    video_url: Optional[str]
    duration_seconds: float         # 实际生成耗时
    error_message: Optional[str]


def create_session_with_retry() -> requests.Session:
    """
    创建带自动重试的 HTTP session。
    
    重试策略：对 429（限频）和 5xx（服务端错误）自动重试，
    使用指数退避（backoff_factor=1 表示间隔 1, 2, 4, 8... 秒）。
    注意：不对 4xx 客户端错误重试，因为重试不会修复参数错误。
    """
    session = requests.Session()
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST"],
        raise_on_status=False,       # 手动处理状态码，保留错误信息
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    return session


class KlingClient:
    """
    Kling API 生产级客户端。
    
    设计决策：
    - JWT token 每次请求前检查是否接近过期，提前 60 秒刷新
    - 轮询使用指数退避，避免在服务繁忙时产生大量无效请求
    - 所有异常都包装为有意义的错误消息
    """
    
    def __init__(self, max_poll_wait: int = 600):
        """
        Args:
            max_poll_wait: 轮询超时时间（秒），默认 600 秒（10 分钟）
                           5 秒 1080p 视频通常需要 2-5 分钟
        """
        self.session = create_session_with_retry()
        self.max_poll_wait = max_poll_wait
    
    def _post(self, endpoint: str, payload: dict) -> dict:
        """统一的 POST 请求，包含错误处理"""
        url = f"{BASE_URL}{endpoint}"
        response = self.session.post(url, json=payload, headers=get_headers())
        
        if response.status_code == 401:
            raise PermissionError("认证失败：请检查 API Key 是否正确，或 JWT 签名是否正确")
        elif response.status_code == 429:
            raise RuntimeError("触发限频（429）：已超过 API 调用配额，请检查套餐限制")
        elif response.status_code == 400:
            raise ValueError(f"请求参数错误（400）: {response.text}")
        
        response.raise_for_status()
        return response.json()
    
    def _get(self, endpoint: str) -> dict:
        """统一的 GET 请求"""
        url = f"{BASE_URL}{endpoint}"
        response = self.session.get(url, headers=get_headers())
        response.raise_for_status()
        return response.json()
    
    def _poll_with_backoff(
        self,
        task_id: str,
        poll_endpoint: str,
        on_progress: Optional[Callable[[str], None]] = None,
    ) -> dict:
        """
        带指数退避的轮询。
        
        退避策略：初始间隔 5 秒，每次乘以 1.2，上限 30 秒。
        这样在任务快完成时（初期）响应及时，
        在长时间等待时（后期）减少 API 调用次数。
        """
        start_time = time.time()
        interval = 5.0
        max_interval = 30.0
        
        while True:
            elapsed = time.time() - start_time
            if elapsed > self.max_poll_wait:
                raise TimeoutError(
                    f"轮询超时：任务 {task_id} 在 {self.max_poll_wait} 秒内未完成"
                )
            
            data = self._get(f"/v1/videos/{poll_endpoint}/{task_id}")
            task_data = data["data"]
            status = task_data["task_status"]
            
            if on_progress:
                on_progress(status)
            
            logger.info(f"task_id={task_id} status={status} elapsed={elapsed:.0f}s")
            
            if status == "succeed":
                return task_data
            elif status == "failed":
                msg = task_data.get("task_status_msg", "未知错误")
                raise RuntimeError(f"任务失败: {msg}")
            
            # 指数退避
            time.sleep(interval)
            interval = min(interval * 1.2, max_interval)
    
    def generate_from_text(
        self,
        prompt: str,
        negative_prompt: str = "",
        model_name: str = "kling-v1-6",
        duration: int = 5,
        aspect_ratio: str = "16:9",
        cfg_scale: float = 0.5,
        camera_control: Optional[dict] = None,
    ) -> VideoGenerationResult:
        """
        文本生成视频，完整流程（提交 + 轮询 + 返回结果）。
        
        camera_control 示例（v3 新增功能）:
            {"type": "zoom_in"}  或  {"type": "pan_left", "config": {"speed": 5}}
        """
        start_time = time.time()
        
        payload = {
            "model_name": model_name,
            "prompt": prompt,
            "cfg_scale": cfg_scale,
            "duration": str(duration),
            "aspect_ratio": aspect_ratio,
        }
        
        if negative_prompt:
            payload["negative_prompt"] = negative_prompt
        
        if camera_control:
            payload["camera_control"] = camera_control
        
        data = self._post("/v1/videos/text2video", payload)
        
        if data.get("code") != 0:
            raise RuntimeError(f"提交失败 code={data['code']}: {data.get('message')}")
        
        task_id = data["data"]["task_id"]
        logger.info(f"任务已提交: {task_id}")
        
        try:
            result_data = self._poll_with_backoff(task_id, "text2video")
            video_url = result_data["task_result"]["videos"][0]["url"]
            
            return VideoGenerationResult(
                task_id=task_id,
                status="succeed",
                video_url=video_url,
                duration_seconds=time.time() - start_time,
                error_message=None,
            )
        except RuntimeError as e:
            return VideoGenerationResult(
                task_id=task_id,
                status="failed",
                video_url=None,
                duration_seconds=time.time() - start_time,
                error_message=str(e),
            )


# 使用示例
if __name__ == "__main__":
    client = KlingClient(max_poll_wait=600)
    
    result = client.generate_from_text(
        prompt="Macro shot of morning dew on a spider web, golden sunlight, 8K",
        negative_prompt="blurry, low quality, watermark",
        duration=5,
        aspect_ratio="16:9",
        cfg_scale=0.5,
        camera_control={"type": "zoom_in"},   # 慢慢推进的镜头效果
    )
    
    if result.status == "succeed":
        print(f"生成成功！耗时: {result.duration_seconds:.1f}s")
        print(f"视频 URL: {result.video_url}")
    else:
        print(f"生成失败: {result.error_message}")

---

6. API 参数完整参考表 {#params-table}

Text-to-Video 参数

参数名	类型	默认值	有效范围	影响什么
model_name	string	—	kling-v1, kling-v1-5, kling-v1-6	模型版本，v1-6 对应 v3 能力
prompt	string	—	最多 2500 字符	视频内容描述，越具体效果越好
negative_prompt	string	""	最多 2500 字符	要排除的视觉元素
cfg_scale	float	0.5	0.0 – 1.0	提示词遵循度；越高越字面，越低越创意
duration	string	"5"	"5" 或 "10"	视频时长（秒），10 秒费用约为 2×
aspect_ratio	string	"16:9"	"16:9", "9:16", "1:1"	画面比例，影响分辨率
camera_control.type	string	null	见下表	镜头运动方式
mode	string	"std"	"std", "pro"	std=标准质量，pro=高质量（费用 2×）

camera_control.type 有效值

值	中文含义
zoom_in	推镜头（向前缩放）
zoom_out	拉镜头（向后缩放）
pan_left	横移向左
pan_right	横移向右
tilt_up	仰拍（向上倾斜）
tilt_down	俯拍（向下倾斜）
rotate_clockwise	顺时针旋转
master_shot	大师镜头（模型自动决定）

---

7. 错误处理 {#error-handling}

常见 HTTP 状态码

HTTP 状态码	含义	解决方法
401	认证失败	检查 JWT 签名算法是否为 HS256；确认 access_key_secret 正确
400	请求参数无效	检查 duration 是否为字符串格式（"5" 而非 5）
429	超过速率限制	免费套餐限制约 2 QPM；升级套餐或实现队列
500	服务端错误	等待 30 秒后重试；如持续出现，检查 status.klingai.com
504	网关超时	任务已提交但响应超时，用 task_id 继续轮询，任务可能仍在处理

业务层错误码（HTTP 200 内）

code 值	含义	解决方法
0	成功	—
1000	系统内部错误	重试；记录 request_id 用于联系支持
1001	账户配额不足	充值或升级套餐
1002	内容审核拒绝	修改 prompt，避免违规内容
1003	模型不存在	检查 model_name 拼写，参考参数表
1004	参数超出范围	检查 cfg_scale 是否在 0.0-1.0 之间

错误处理代码示例

# error_handling_example.py — 生产环境错误处理模式
import requests
import logging
from typing import Optional

logger = logging.getLogger(__name__)


def safe_generate(client, prompt: str, retries: int = 2) -> Optional[str]:
    """
    带完整错误处理的生成函数。
    
    区分"可重试错误"和"不可重试错误"：
    - 网络超时、5xx → 重试有意义
    - 内容审核拒绝（1002）、参数错误（400）→ 重试浪费配额
    """
    for attempt in range(retries + 1):
        try:
            result = client.generate_from_text(prompt=prompt)
            
            if result.status == "succeed":
                return result.video_url
            else:
                error_msg = result.error_message or ""
                
                # 内容审核失败：不重试，直接返回 None
                if "1002" in error_msg or "content" in error_msg.lower():
                    logger.error(f"内容审核拒绝，prompt 需要修改: {prompt[:50]}...")
                    return None
                
                # 配额不足：不重试
                if "1001" in error_msg:
                    logger.error("账户配额不足，请充值")
                    return None
                
                # 其他失败：如果还有重试次数则继续
                if attempt < retries:
                    logger.warning(f"第 {attempt + 1} 次失败，准备重试: {error_msg}")
                    continue
                    
                return None
                
        except PermissionError as e:
            # 401 错误：认证问题，重试无意义
            logger.error(f"认证错误（不重试）: {e}")
            return None
            
        except requests.exceptions.ConnectionError as e:
            # 网络连接失败：可重试
            if attempt < retries:
                logger.warning(f"网络连接失败，第 {attempt + 1} 次重试...")
                continue
            logger.error(f"网络连接持续失败: {e}")
            return None
            
        except TimeoutError as e:
            logger.error(f"轮询超时（任务可能仍在处理）: {e}")
            return None
    
    return None

---

8. 性能与成本对照表 {#cost-table}

以下数字基于 Kling 官方定价页及社区测试（2026 年 Q1 数据），实际成本受套餐折扣影响。

生成时间参考（5 秒视频）

模式	分辨率	平均生成时间	最大等待时间（P95）
std 标准	720p	60–90 秒	180 秒
std 标准	1080p	90–150 秒	300 秒
pro 高质量	1080p	150–240 秒	480 秒

成本对照（每次生成）

配置	时长	模式	估算成本（美元）	适用场景
720p 16:9	5 秒	std	~$0.14	快速原型、批量测试
1080p 16:9	5 秒	std	~$0.28	常规内容生产
1080p 16:9	10 秒	std	~$0.56	长镜头内容
1080p 16:9	5 秒	pro	~$0.56	最终交付质量
1080p 9:16	5 秒	std	~$0.28	短视频/竖版内容

何时不应该用 Kling API

• 需要实时生成（延迟 < 10 秒）：Kling 不适合，最低等待 60 秒以上
• 每日生成量 > 1000 次：建议直接接触 Kling 商务谈批量折扣，按量计费成本高
• 需要精确的口型同步：v3 支持 native audio 但 API 端暂不完全开放音频输入参数
• 视频超过 10 秒：目前 API 最长支持 10 秒，长视频需拼接

---

9. 结语 {#conclusion}

本文覆盖了从 JWT 认证、T2V/I2V 提交、生产级轮询到错误处理的完整流程，所有代码块均可直接运行。Kling v3 API 的核心约束是异步生成延迟（60–240 秒）和按次计费模式，在设计系统时需围绕这两点做架构决策。如果你发现 API 行为与本文不符，请以 Kling 官方 API 文档 为准，版本迭代较快。

提示： 如果你需要在同一个项目中使用多个 AI 模型，AtlasCloud 提供统一 API 接入 300+ 模型（Kling、Flux、Seedance、Claude、GPT 等），一个 key 全部搞定。新用户首次充值享 25% 赠送（最高 $100）。