Kling v3 API 完全Pythonチュートリアル 2026

Primary keyword: how to use kling api kling v3 api python tutorial

---

はじめに：3つの数字で理解するKling v3

Kling v3（Kling 3.0）は2026年2月4日にリリースされた。開発者が知るべき数字は3つだ。

• 生成レイテンシ： 5秒動画で約60〜120秒（モデルや負荷によって変動）
• API利用コスト： kling-v1-5モデルで1本あたり約$0.14〜$0.28、v3はプレミアム価格帯（後述の料金表参照）
• 動画解像度： 最大4K（3840×2160）出力対応、従来v1.5比でモーション品質スコアが大幅向上

このチュートリアルはKling APIをPythonから呼び出し、テキストまたは画像から動画を生成する実装を扱う。ウェブUIの解説ではなく、本番環境で動くコードを提供することが目的だ。

---

前提条件

必要なもの

項目	詳細
Klingアカウント	klingai.com または Kuaishouの開発者ポータルで登録
APIキー	KlingのAPIコンソールから発行（後述）
Python	3.9以上
支払い方法	APIアクセスにはクレジット購入が必要

ライブラリのインストール

# 必須ライブラリ
pip install requests python-dotenv

# JWT認証に必要（KlingはJWT署名方式を採用）
pip install PyJWT

# オプション：非同期処理用
pip install httpx

環境変数の設定

.env ファイルをプロジェクトルートに作成する：

# .env
KLING_ACCESS_KEY=your_access_key_here
KLING_SECRET_KEY=your_secret_key_here

なぜ環境変数か： APIキーをコードに直書きするとGitHubに誤ってpushした瞬間にキーが漏洩する。.envは.gitignoreに追加すること。

---

認証：JWT署名の実装

KlingのAPIはBearerトークン方式ではなく、アクセスキー＋シークレットキーによるJWT署名を使用する。これはAWS Signature v4に近い設計だ。毎リクエスト前にJWTを生成する必要がある。

# auth.py
import jwt
import time
import os
from dotenv import load_dotenv

load_dotenv()

def generate_kling_token(expiration_seconds: int = 1800) -> str:
    """
    Kling API用のJWTトークンを生成する。
    
    KlingはHMAC-SHA256でJWTを署名する。
    トークンの有効期限はデフォルト30分。
    長時間のバッチ処理では expiration_seconds を伸ばすこと。
    """
    access_key = os.getenv("KLING_ACCESS_KEY")
    secret_key = os.getenv("KLING_SECRET_KEY")
    
    if not access_key or not secret_key:
        raise ValueError(
            "KLING_ACCESS_KEY と KLING_SECRET_KEY が環境変数に設定されていない"
        )
    
    now = int(time.time())
    
    payload = {
        "iss": access_key,           # issuer: アクセスキーをissuerとして使用
        "exp": now + expiration_seconds,  # expiration: トークン有効期限
        "nbf": now - 5,              # not before: 5秒の時刻ズレを許容
    }
    
    # Klingが要求するアルゴリズムはHS256
    token = jwt.encode(
        payload,
        secret_key,
        algorithm="HS256"
    )
    
    return token


def get_auth_headers() -> dict:
    """
    APIリクエストに使うAuthorizationヘッダーを返す。
    毎回この関数を呼ぶことでトークン期限切れを防ぐ。
    """
    token = generate_kling_token()
    return {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }


# 動作確認
if __name__ == "__main__":
    headers = get_auth_headers()
    print(f"Token generated: {headers['Authorization'][:50]}...")

---

APIキーの取得方法

1. klingai.com にログイン
2. 右上のアカウントアイコン → 「API」 または 「Developer」 セクションへ
3. 「Create API Key」 をクリック
4. Access KeyとSecret Keyをコピーして .env に貼り付ける

注意： KlingのAPIコンソールはKuaishou（快手）の開発者ポータルと統合されている場合がある。地域によってUIが異なるため、platform.klingai.com も確認すること。

---

基本実装：テキストから動画を生成する

Step 1: タスクの作成（非同期ジョブ）

Kling APIは非同期方式で動作する。リクエストを送るとジョブIDが返り、別途ポーリングして結果を取得する設計だ。

# basic_text_to_video.py
import requests
import time
from auth import get_auth_headers

# Kling v1.5 / v1 の公式エンドポイント
# v3はkling-v1モデルの上位として同じエンドポイントを使用
BASE_URL = "https://api.klingai.com"

def create_text_to_video_task(
    prompt: str,
    model: str = "kling-v1",
    duration: str = "5",
    aspect_ratio: str = "16:9",
    mode: str = "std",
) -> str:
    """
    テキストから動画生成タスクを作成し、タスクIDを返す。
    
    戻り値のtask_idを使って後でステータスをポーリングする。
    """
    endpoint = f"{BASE_URL}/v1/videos/text2video"
    
    payload = {
        "model_name": model,       # 使用するモデル名（後述のモデル一覧参照）
        "prompt": prompt,          # 動画生成の指示文
        "duration": duration,      # 動画の長さ: "5" or "10"（秒）
        "aspect_ratio": aspect_ratio,  # アスペクト比
        "mode": mode,              # "std"（標準）or "pro"（高品質、遅い）
    }
    
    response = requests.post(
        endpoint,
        headers=get_auth_headers(),
        json=payload,
        timeout=30,  # 接続タイムアウト30秒
    )
    
    # HTTPステータスコードで一次チェック
    response.raise_for_status()
    
    data = response.json()
    
    # APIレベルのエラーチェック
    if data.get("code") != 0:
        raise RuntimeError(
            f"Kling API error: code={data.get('code')}, "
            f"message={data.get('message')}"
        )
    
    task_id = data["data"]["task_id"]
    print(f"Task created: {task_id}")
    return task_id


def poll_task_status(task_id: str, max_wait_seconds: int = 300) -> dict:
    """
    タスクが完了するまでポーリングし、結果を返す。
    
    Klingの動画生成は5秒動画で60〜120秒かかる。
    ポーリング間隔は10秒に設定（短すぎるとレート制限に引っかかる）。
    """
    endpoint = f"{BASE_URL}/v1/videos/text2video/{task_id}"
    poll_interval = 10  # 10秒ごとにチェック
    elapsed = 0
    
    while elapsed < max_wait_seconds:
        response = requests.get(
            endpoint,
            headers=get_auth_headers(),  # トークンは毎回再生成
            timeout=30,
        )
        response.raise_for_status()
        data = response.json()
        
        if data.get("code") != 0:
            raise RuntimeError(f"Polling error: {data.get('message')}")
        
        task_status = data["data"]["task_status"]
        print(f"Status: {task_status} (elapsed: {elapsed}s)")
        
        if task_status == "succeed":
            # 生成完了：動画URLを含む結果を返す
            return data["data"]
        elif task_status == "failed":
            raise RuntimeError(
                f"Task failed: {data['data'].get('task_status_msg', 'Unknown error')}"
            )
        
        # "processing" or "submitted" の場合は待機
        time.sleep(poll_interval)
        elapsed += poll_interval
    
    raise TimeoutError(f"Task {task_id} did not complete within {max_wait_seconds}s")


# 実行例
if __name__ == "__main__":
    task_id = create_text_to_video_task(
        prompt="A red panda typing on a keyboard in a cozy library, cinematic lighting",
        model="kling-v1",
        duration="5",
        aspect_ratio="16:9",
        mode="std",
    )
    
    result = poll_task_status(task_id)
    
    # 生成された動画URLを取得
    videos = result.get("task_result", {}).get("videos", [])
    for video in videos:
        print(f"Video URL: {video['url']}")
        print(f"Duration: {video.get('duration')}s")

---

本番実装：画像から動画を生成する（Image-to-Video）

テキスト→動画より実用的なユースケースとして、既存の画像を起点に動画を生成するImage-to-Videoを実装する。

# image_to_video.py
import requests
import base64
import time
import os
from pathlib import Path
from auth import get_auth_headers

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


def encode_image_to_base64(image_path: str) -> str:
    """
    ローカル画像をbase64エンコードする。
    KlingはURLとbase64の両方に対応しているが、
    ローカルファイルはbase64で送る必要がある。
    """
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")


def create_image_to_video_task(
    image_source: str,          # ローカルパスまたはhttps://URL
    prompt: str,
    model: str = "kling-v1-5",  # v1.5はImage-to-Videoで高品質
    duration: str = "5",
    aspect_ratio: str = "16:9",
    mode: str = "std",
    negative_prompt: str = "",
    cfg_scale: float = 0.5,     # プロンプトへの忠実度（0.0〜1.0）
) -> str:
    """
    画像から動画生成タスクを作成する。
    
    image_sourceがhttps://で始まる場合はURLとして送信。
    それ以外はローカルファイルとしてbase64エンコードして送信。
    """
    endpoint = f"{BASE_URL}/v1/videos/image2video"
    
    # 画像のソース形式を判定
    if image_source.startswith("https://") or image_source.startswith("http://"):
        image_payload = {"image_url": image_source}
    else:
        # ローカルファイルをbase64に変換
        b64_data = encode_image_to_base64(image_source)
        # MIMEタイプを拡張子から判定（jpg/png対応）
        ext = Path(image_source).suffix.lower()
        mime = "image/jpeg" if ext in [".jpg", ".jpeg"] else "image/png"
        image_payload = {"image": f"data:{mime};base64,{b64_data}"}
    
    payload = {
        "model_name": model,
        "prompt": prompt,
        "negative_prompt": negative_prompt,
        "duration": duration,
        "aspect_ratio": aspect_ratio,
        "mode": mode,
        "cfg_scale": cfg_scale,
        **image_payload,  # image_url or image をマージ
    }
    
    response = requests.post(
        endpoint,
        headers=get_auth_headers(),
        json=payload,
        timeout=60,  # base64エンコード画像は大きいので60秒
    )
    response.raise_for_status()
    
    data = response.json()
    if data.get("code") != 0:
        raise RuntimeError(
            f"API Error {data.get('code')}: {data.get('message')}"
        )
    
    return data["data"]["task_id"]


def download_video(url: str, output_path: str) -> None:
    """
    生成された動画をローカルに保存する。
    動画URLには有効期限があるため、取得後すぐにダウンロードすること。
    """
    response = requests.get(url, stream=True, timeout=120)
    response.raise_for_status()
    
    with open(output_path, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    
    file_size_mb = os.path.getsize(output_path) / (1024 * 1024)
    print(f"Saved: {output_path} ({file_size_mb:.1f} MB)")


# --- メイン実行 ---
if __name__ == "__main__":
    # ポーリング関数はbasic_text_to_video.pyから再利用
    from basic_text_to_video import poll_task_status
    
    task_id = create_image_to_video_task(
        image_source="./sample_image.jpg",  # ローカル画像パス
        prompt="The character slowly turns their head and smiles",
        model="kling-v1-5",
        duration="5",
        cfg_scale=0.5,
    )
    
    result = poll_task_status(task_id, max_wait_seconds=300)
    
    videos = result.get("task_result", {}).get("videos", [])
    if videos:
        download_video(videos[0]["url"], "output_video.mp4")

---

APIパラメータリファレンス

パラメータ名	型	デフォルト値	有効範囲	影響する要素
model_name	string	kling-v1	kling-v1, kling-v1-5, kling-v1-6	生成品質・速度・コスト
prompt	string	必須	最大2500文字	動画の内容・スタイル全般
negative_prompt	string	""	最大2500文字	除外したい要素
duration	string	"5"	"5", "10"	動画の長さ（秒）
aspect_ratio	string	"16:9"	"16:9", "9:16", "1:1"	出力解像度の比率
mode	string	"std"	"std", "pro"	品質vs速度のトレードオフ
cfg_scale	float	0.5	0.0〜1.0	プロンプト忠実度（高いほど指示に従う）
camera_control	object	null	各種カメラ移動パラメータ	カメラワーク（v1.5+）
callback_url	string	null	有効なhttps://URL	完了時Webhookの送信先

---

エラーハンドリング

よくあるエラーコードと対処法

エラーコード	HTTPステータス	原因	対処法
1101	401	認証失敗（JWTが無効または期限切れ）	トークンを再生成、時刻同期を確認
1102	403	APIキーの権限不足	コンソールでAPIアクセス権限を確認
1200	429	レート制限（Too Many Requests）	exponential backoffで再試行
1301	400	プロンプトがコンテンツポリシー違反	プロンプトを修正
1302	400	画像フォーマットが無効	JPEG/PNG、最大10MB以内に変換
2000	500	サーバー内部エラー	30秒後にリトライ
task: failed	200	動画生成処理の失敗	task_status_msgを確認してプロンプト調整

本番対応のエラーハンドリング実装

# production_client.py
import requests
import time
import logging
from auth import get_auth_headers

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

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

# リトライ対象のエラーコード（一時的な障害）
RETRYABLE_ERROR_CODES = {1200, 2000}
MAX_RETRIES = 3


def api_request_with_retry(
    method: str,
    endpoint: str,
    payload: dict = None,
    max_retries: int = MAX_RETRIES,
) -> dict:
    """
    指数バックオフ付きAPIリクエスト。
    
    レート制限（1200）やサーバーエラー（2000）は自動リトライする。
    認証エラー（1101/1102）はリトライせず即座に例外を上げる。
    """
    url = f"{BASE_URL}{endpoint}"
    
    for attempt in range(max_retries + 1):
        try:
            if method.upper() == "GET":
                response = requests.get(
                    url,
                    headers=get_auth_headers(),
                    timeout=30,
                )
            else:
                response = requests.post(
                    url,
                    headers=get_auth_headers(),  # 毎回JWTを再生成
                    json=payload,
                    timeout=60,
                )
            
            # HTTPレベルのエラー（4xx, 5xx）
            if response.status_code == 429:
                # レート制限：Retry-Afterヘッダーを尊重する
                retry_after = int(response.headers.get("Retry-After", 30))
                logger.warning(f"Rate limited. Waiting {retry_after}s...")
                time.sleep(retry_after)
                continue
            
            response.raise_for_status()
            data = response.json()
            
            api_code = data.get("code", 0)
            
            if api_code == 0:
                return data  # 成功
            
            if api_code in RETRYABLE_ERROR_CODES and attempt < max_retries:
                # 指数バックオフ: 1s, 2s, 4s...
                wait_time = (2 ** attempt)
                logger.warning(
                    f"Retryable error {api_code}. "
                    f"Attempt {attempt + 1}/{max_retries}. "
                    f"Waiting {wait_time}s..."
                )
                time.sleep(wait_time)
                continue
            
            # リトライ不可、または上限に達した場合
            raise RuntimeError(
                f"Kling API error: code={api_code}, "
                f"message={data.get('message', 'unknown')}"
            )
        
        except requests.exceptions.ConnectionError as e:
            # ネットワーク接続エラー
            if attempt < max_retries:
                wait_time = 2 ** attempt
                logger.error(f"Connection error. Retrying in {wait_time}s: {e}")
                time.sleep(wait_time)
            else:
                raise
        
        except requests.exceptions.Timeout:
            logger.error(f"Request timeout on attempt {attempt + 1}")
            if attempt >= max_retries:
                raise
    
    raise RuntimeError("Max retries exceeded")

---

Webhookを使った非同期完了通知

本番環境でポーリングを避ける場合、callback_urlを使う。

# webhook_example.py
# タスク作成時にcallback_urlを指定するだけ
payload = {
    "model_name": "kling-v1-5",
    "prompt": "A timelapse of cherry blossoms falling",
    "duration": "5",
    "aspect_ratio": "16:9",
    "mode": "std",
    "callback_url": "https://your-server.com/webhooks/kling",
    # KlingはPOSTでJSONボディをこのURLに送信する
    # ペイロード例: {"task_id": "...", "task_status": "succeed", "task_result": {...}}
}

# Webhookサーバー側（FastAPIの例）
# from fastapi import FastAPI, Request
# app = FastAPI()
# @app.post("/webhooks/kling")
# async def handle_kling_webhook(request: Request):
#     data = await request.json()
#     task_id = data["task_id"]
#     status = data["task_status"]
#     if status == "succeed":
#         video_url = data["task_result"]["videos"][0]["url"]
#         # 動画URLをDBに保存するなどの処理
#     return {"ok": True}

---

パフォーマンスとコストの比較表

実測値または公開情報に基づく参考値。価格はクレジット単価によって変動する。

モデル	動画長	モード	生成時間（目安）	クレジット消費	最大解像度	推奨用途
kling-v1	5秒	std	60〜90秒	低（参考: $0.14相当）	1080p	プロトタイプ・テスト
kling-v1	10秒	std	90〜150秒	中	1080p	コスト重視の量産
kling-v1-5	5秒	std	90〜120秒	中	1080p	バランス型本番環境
kling-v1-5	5秒	pro	120〜180秒	高（参考: $0.28相当）	1080p	高品質コンテンツ
kling-v1-6	5秒	std	90〜150秒	中〜高	4K	最高品質優先

コスト注意： Kling v3（kling-v1-6相当）はv1より約2〜3倍のクレジットを消費する。量産用途では事前にバッチテストでコストを計測すること。

---

Kling APIを使うべきでないケース

• リアルタイム性が必要な場合： 最低でも60秒のレイテンシがある。ライブストリーミングには使えない
• 1秒以下の短いクリップが必要な場合： 最短5秒が最小単位
• 月間1000本以上の量産： コストが急増する。RunwayやPikaと比較コスト試算を先にすること
• 著作権管理が厳格な商用コンテンツ： Klingの利用規約と出力の商用利用条件を法務確認すること

---

まとめ

Kling v3 APIはJWT認証→タスク作成→ポーリング（またはWebhook）の3ステップで統合できる。本番環境ではexponential backoff付きのリトライ実装とcallback_urlによる非同期処理が必須だ。コスト管理のためにモデルとmodeの選択を用途別に分けることで、品質とコストのバランスを取れる。

---

情報の正確性については公式ドキュメント platform.klingai.com および Kuaishou APIポータルで最新情報を確認すること。APIエンドポイントや料金は2026年中に変更される可能性がある。

メモ： 複数の AI モデルを一つのパイプラインで使う場合、AtlasCloud は Kling、Flux、Seedance、Claude、GPT など 300+ モデルへの統一 API アクセスを提供します。API キー一つで全モデル対応。新規ユーザーは初回チャージで 25% ボーナス（最大 $100）。