kling-video

Kling AI で動画を生成する

モデルの提供状況、API フィールド、価格、ポリシー、制限は変わる可能性があります。以下の情報は、2026-07-10に Kling／Kuaishou の一次情報源で確認したものです。利用や公開の前に、リンク先のモデルページ、料金表、更新ログ、利用規約を再確認してください。

利用面とジョブの内容を揃える

納品先（直接 API または Kling のウェブ／アプリ）、地域、タスク、入力メディア、長さ、アスペクト比、解像度、音声／セリフ、ショット構成、対象視聴者、権利／同意、プライバシー要件、バリエーション数、承認済みの予算を確認してください。

利用面	用途	混同しない項目
国際向け直接 API	https://api-singapore.klingai.com を使ったサーバーワークフロー。API キー認証、秒単位の API 課金	旧 api.klingai.com、中国向けホストのルーティング、コンシューマー向けクレジット、再販業者のスキーマ
Kling ウェブ／アプリ	インタラクティブな作成、Elements ライブラリ、メンバーシップ、無料／有料の UI 機能	UI のコントロール、クレジット、ウォーターマークの特典、デフォルト設定が API にも存在するという前提
kling-v3／VIDEO 3.0	プロンプト主導のテキスト／画像／開始・終了フレームからの生成、ネイティブ音声、柔軟な長さ、マルチショット	Omni の動画編集／参照入力
kling-v3-omni／VIDEO 3.0 Omni	参照入力を使った生成、動画／参照入力、Elements、編集、ネイティブ音声、マルチショット	Kling Image 3.0 Omni、または Kolors
kling-3.0-turbo	現行の高速な、スケールを重視した 3.0 オプション。ネイティブ音声に対応	すべての 3.0 または Omni のフィールドがそのまま受け付けられるという前提
Motion Control	参照キャラクターに、動作の元となる演技を転写	アバター、リップシンク、顔認識、単独の音声 API

事実： 現行の API ドキュメントでは、すべてのモデルに新しい形式の API キーが必要です。従来の AccessKey／SecretKey の JWT 認証情報は 3.0 以前の利用に限って引き続き有効で、新しいモデルには使用できません。サーバー側で KLING_API_KEY を使用し、ブラウザーやモバイルクライアントには決して公開しないでください。現行の国際向けホストはシンガポールにありますが、中国向けデプロイには北京の別のホストがあります。ホストの選択をデータレジデンシーの保証と見なさないでください。認証

事実： VIDEO 3.0 と 3.0 Omni は、テキストから動画、画像から動画、ネイティブ音声、マルチショットのストーリーテリング、柔軟な3〜15 秒の出力に対応します。VIDEO 3.0 はプロンプト主導の方式です。Omni では、画像／動画／Element の参照や編集のワークフローがさらに充実しています。整合性、物理挙動、テキスト、リップシンク、品質について、一次情報源が述べる「完璧」「究極」「産業グレード」といった表現は、プロバイダーの主張であり、納品の保証ではありません。VIDEO 3.0 ガイド、VIDEO 3.0 Omni ガイド、Kuaishou の発表

事実： Kling-Omni は、Kling チームの技術レポートで、生成、編集、マルチモーダルな指示への追従を統合したマルチモーダルフレームワークとして説明されています。プロバイダーが作成したレポート内のベンチマークや選好に関する結果は、あくまでプロバイダーによる根拠です。独立したランキングとして扱わないでください。Kling-Omni レポート

現在のモデルファミリーを意図的に使い分ける

• シーンをプロンプト、開始画像、または開始・終了フレームで指定できる場合は、kling-v3を選択してください。カバレッジよりも連続性が重要な場合はシングルショットモードを使用し、カットを意図している場合に限りマルチショットを使用してください。
• リクエストが参照動画、複数の画像／Element参照、被写体と音声の紐付け、または動画内編集に依存する場合は、kling-v3-omniを選択してください。参照の組み合わせはすぐに変わるため、該当するOmniエンドポイントのページを再度確認してください。
• kling-3.0-turboは、現在のタスクページでリクエストされたタスクとフィールドが確認できた場合に限り選択してください。公式プラットフォームは、より高速で大規模利用でも信頼性が高いと説明しています。日付入りのワークロードテストで検証されるまでは、これらはプロバイダーの主張として扱ってください。
• 既存の互換性またはコスト要件がある場合に限り、kling-video-o1、kling-v2-6、kling-v2-5-turbo、および以前のIDを使ってください。3.0のフィールドが拒否されたときに、黙ってダウングレードしないでください。
• パフォーマンスの転写にはMotion Controlを使用してください。入力と類似性に関するリスクが十分に異なるため、画像生成の承認だけではなく、駆動動画と出演者の同意を明示的に得る必要があります。

VIDEO 3.0のネイティブ音声について、現行の製品ガイドには、中国語、英語、日本語、韓国語、スペイン語と、一部の方言・アクセントが記載されています。該当するアカウントでの言語、モデル、APIのサポートを確認してください。音声の個人性、発音、タイミング、リップシンクの正確さを保証しないでください。発音を記録し、実際の音声を確認してください。

正確な構成で料金を見積もる

現在、直接APIの課金ページでは生成秒数ごとの料金が設定されています。1080pの場合、kling-v3は無音で$0.112/s、ネイティブ音声付きで$0.168/sです。720pは$0.084/s、$0.126/sです。対象となるkling-v3／kling-v3-omniの構成では、ネイティブ4Kは**$0.42/s**と記載されています。ネイティブ音声付きのkling-3.0-turboは、720pで$0.112/s、1080pで$0.14/sです。Omniの料金は、動画入力とネイティブ音声を使用するかどうかによって異なります。現在のAPI課金

有料の生成を行う前に、毎回次の手順を実施してください。

1. 課金表と該当するモデルのタスクページを再度開き、モデル、タスク、モード／解像度、音声、長さ、参照、バリアント数を見積もりに反映してください。
2. 日付入りの秒単価に長さと回数を掛けてください。1080pで5秒の無音のkling-v3生成は、現在$0.112 × 5 = $0.56です。ネイティブ音声付きの場合は$0.168 × 5 = $0.84です。
3. 再生成、編集、延長、または各バリアントは、新たな料金が発生する可能性があるものとして扱ってください。失敗または短縮されたタスクが返金されると想定しないでください。Motion Controlでは、有効なモーション区間をより短く抽出した場合でも返金されないことが明示的に警告されています。
4. リクエストの正確なダイジェストと、十分な上限額について明示的な承認を得てください。UIの「Generate」確認や一般的な「use Kling」という指示は、支出の承認ではありません。
5. 残高、パッケージの有効性、モデルの利用権限、同時実行数、QPS、IP許可リストを確認してください。コード1101/1102は残高／パッケージに関するもの、1302/1303はレート／同時実行に関するものです。権限なしに購入、チャージ、またはアカウント設定の変更を行わないでください。

時間とカバレッジを意識したプロンプトを作成する

単一ショットでは、被写体と変化しない特徴、物理的な動作、設定、フレーミング／レンズ、カメラの動き、照明／色／質感、時間の経過に伴う展開、音を指定します。3–15秒で無理のないアクション数に抑えます。

Single continuous 5-second shot, no cuts. A hand-thrown ceramic cup sits on a dark walnut table.
[0–2s] Macro close-up; a ribbon of steam curls upward in quiet morning light.
[2–5s] Slow 10 cm dolly left reveals a rain-streaked window; keep the cup shape and glaze unchanged.
Natural shallow depth of field, restrained blue-gray palette, believable steam and reflections.
No people, no lettering, no logo. Audio off.

画像から動画を生成する場合は、画像をアイデンティティと構図の基準として扱い、主に変化する要素をプロンプトで指定します。何が動くのか、何を固定したままにするのか、カメラの挙動、終了時の状態を明確にします。開始フレームと終了フレームを指定する場合は、無関係な出来事ではなく、もっともらしいつながりを求めます。

マルチショットでは、すべてのカットに意図を持たせます。VIDEO 3.0 では、自動マルチショットと、ショットごとの長さ／フレーミング／視点／アクション／カメラを指定するカスタム制御を利用できます。カスタムプランには最大6つのショットを含められます。各ショットの長さの合計は、全体の長さと一致させる必要があります。15秒に、互いに無関係なアイデアを6つ詰め込むのは避けます。

{
  "model_name": "kling-v3",
  "multi_shot": true,
  "shot_type": "customize",
  "multi_prompt": [
    {"index": 1, "duration": "3", "prompt": "Wide dawn exterior: a cyclist stops outside a small bakery; gentle locked camera."},
    {"index": 2, "duration": "2", "prompt": "Close-up: the same gloved hand lifts the warm paper bag; steam in backlight."}
  ],
  "duration": "5",
  "mode": "pro",
  "aspect_ratio": "16:9",
  "sound": "off"
}

これは例であり、すべてに使えるテンプレートではありません。使用前に、最新の3.0タスクスキーマを確認してください。multi_shot=trueの場合、カスタムショットプロンプトが単一のプロンプトに置き換わります。インデックスを連続させ、長さを正確に指定してください。

ネイティブ音声を使う場合は、引用符で囲んだ各セリフの前に話者を指定し、必要な場合に限り言語／アクセントを指定します。また、声の意図、環境音、効果音、除外する音も指定します。セリフは自然に話せる長さに抑えます。正確さが必須の場合は、事実に関する文言、法的な主張、指定どおりの音楽、ブランドテキストをポストプロダクションに分けてください。

リファレンスとモーションを本番用アセットとして管理する

3.0 Omni ガイドには、現在、次の内容が記載されています。

• 画像は最大7枚で、それぞれ縦横300 px以上、最大10 MB、JPG/JPEG/PNGです。
• リファレンス動画は1本で、3〜10秒、最大200 MB、解像度は2K以下です。
• 動画を使用する場合、画像とElementの合計は4点以下です。動画を使用しない場合は最大7点です。
• 複数画像のElementでは、最大4つのビューを使用できます。キャラクターの音声バインディングには、雑音のない一人の発話を5〜30秒使用できます。動画キャラクターElementには、3〜8秒の一人のキャラクターのクリップを使用できます。

フィールドごとの制限であり、全体に共通する制限ではありません： 現在のOmniガイドによると、生成されるVIDEO 3.0/3.0 Omni出力は3〜15秒ですが、FAQではアップロードするリファレンス動画を別途3〜10秒に制限しています。15秒の生成に関する発表を、アップロード可能な動画の長さが15秒になったという意味に解釈しないでください。正確な現行APIタスクスキーマでその入力フィールドが明示的に変更されない限り、リファレンス動画の長さは10秒以下にしてください。例外を設ける場合は、根拠と日付を記録してください。同様に、リファレンス動画／ネイティブ音声／4Kの組み合わせがサポートされていることを確認してください。別々の機能に関する発表を組み合わせて判断しないでください。

アップロード前に、すべてのソースのハッシュを計算し、コンテンツを検証してください。バイトサイズ、MIME、権利、同意、目的、リモート識別子を記録してください。公開バケットではなく、非公開で有効期間の短い取得URL、または受け付けられるbase64を使用してください。署名付きURLやbase64をログに記録しないでください。リファレンスタグ／Element名は、実際のアセットに一意に対応させる必要があります。タグがあるからといって、人物の同一性、製品の色、衣装、テキスト、音声が保持されるとは限りません。

モーションコントロールは、駆動動画の身体、顔、手の動きを1枚のキャラクター画像に反映します。全身または半身の構図を合わせ、身体と頭部が画面内に収まるようにしてください。カットやカメラ移動のない、動きが中程度の3〜30秒の連続した一人のキャラクターのテイクが推奨されます。公式ガイドでは、複数人が映る入力では画面上で最も大きい人物が選択されること、複雑な動きや速い動きが途中で切り捨てられる場合があること、生成時間が短くなってもクレジットが消費されることが警告されています。Elementのバインディングでは、衣服、髪、メイク、持ち物ではなく、顔のリファレンス情報が使用されます。モーションコントロールガイド、技術レポートと影響に関する声明

参照される人物と駆動パフォーマーの両方から、モーション、顔、音声、台詞、配布、実質的な変換について、文書化された同意を得てください。親密な画像を使った虐待、欺瞞的ななりすまし、嫌がらせ、未成年者の搾取、違法な監視、政治家や公人への説得を目的とした無許可の利用は拒否してください。安全性に関する1300/1301の応答を弱めたり、回避するためにリクエストをゲートウェイに移したりしないでください。

フェイルクローズ方式の直接 API トランザクションを使用する

直接 API は非同期です。現在の動画の結果はプロバイダーがホストする配信アーティファクトであり、永続ストレージではありません。ファーストパーティの API 資料によると、生成された画像／動画は 30 日後に削除されます。また、有料 API の規約によると、プロンプトとレスポンスは 30 日間記録されます。速やかに管理下のストレージへダウンロードしてください。external_task_id は照合に役立ちますが、冪等性キーとしては文書化されていません。POST の結果が不明確で、永続化されたプロバイダーのタスク ID がない場合は、停止して照合してください。「念のため」の再試行は絶対にしないでください。

以下の完全な例では、現在の国際 API を通じて、プロンプト主導の kling-v3 テキストから動画へのタスクを 1 件送信します。デフォルトではオフラインのドライランを実行し、正確な承認と最新の料金証拠、および正確な配信ホストの許可リストを紐付け、POST を最大 1 回だけ実行し、回数を制限した GET の再試行のみを使用します。API キーを転送せずにダウンロードし、MP4 を検証し、アーティファクトを上書きしません。また、永続化されたステージ済みアーティファクトの記録によって、検証、アトミックな公開、最終的な台帳状態の間にクラッシュが発生する隙間も塞ぎます。4K、Omni リファレンス、コールバック、自動的な作成リクエストの再送は意図的に除外しています。

#!/usr/bin/env python3
import argparse, hashlib, http.client, ipaddress, json, os, random, socket, ssl
import subprocess, tempfile, threading, time, uuid
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from urllib.parse import urlsplit

API_HOST = "api-singapore.klingai.com"
MODEL = "kling-v3"
MAX_JSON = 1024 * 1024
MAX_VIDEO = 250 * 1024 * 1024
MAX_TOOL = 1024 * 1024
RATES = {("std", "off"): Decimal("0.084"), ("std", "on"): Decimal("0.126"),
         ("pro", "off"): Decimal("0.112"), ("pro", "on"): Decimal("0.168")}

class APIError(RuntimeError):
    def __init__(self, message, status=None, retry_after=None, body_sha256=None):
        super().__init__(message); self.status = status; self.retry_after = retry_after; self.body_sha256 = body_sha256

def canonical(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()

def fsync_directory(path):
    if os.name == "nt": return
    fd = os.open(path, os.O_RDONLY)
    try: os.fsync(fd)
    finally: os.close(fd)

def atomic_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(value, f, indent=2, sort_keys=True); f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path); fsync_directory(path.parent)
    finally:
        if os.path.exists(tmp): os.unlink(tmp)

def claim_once(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        json.dump(value, f, indent=2, sort_keys=True); f.flush(); os.fsync(f.fileno())
    fsync_directory(path.parent)

def read_small_json(path):
    if path.stat().st_size > MAX_JSON: raise RuntimeError("record exceeds cap")
    with path.open("rb") as f: raw = f.read(MAX_JSON + 1)
    value = json.loads(raw)
    if not isinstance(value, dict): raise RuntimeError("record must be an object")
    return value

def run_bounded(argv, timeout):
    process = subprocess.Popen(argv, shell=False, stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, err, total, lock, overflow = bytearray(), bytearray(), [0], threading.Lock(), threading.Event()
    def drain(stream, bucket):
        while True:
            chunk = stream.read(65536)
            if not chunk: break
            with lock:
                room = max(0, MAX_TOOL - total[0]); bucket.extend(chunk[:room]); total[0] += len(chunk)
                if total[0] > MAX_TOOL: overflow.set()
            if overflow.is_set():
                try: process.kill()
                except OSError: pass
    threads = [threading.Thread(target=drain, args=(process.stdout, out), daemon=True),
               threading.Thread(target=drain, args=(process.stderr, err), daemon=True)]
    for thread in threads: thread.start()
    timed_out = False
    try: code = process.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        timed_out = True; process.kill(); code = process.wait()
    for thread in threads: thread.join()
    return code, bytes(out), bytes(err), timed_out, overflow.is_set()

def api_json(method, path, key, payload=None, timeout=60):
    body = canonical(payload) if payload is not None else None
    headers = {"Authorization": "Bearer " + key, "Accept": "application/json"}
    if body is not None: headers["Content-Type"] = "application/json"
    connection = http.client.HTTPSConnection(API_HOST, 443, timeout=timeout, context=ssl.create_default_context())
    try:
        connection.request(method, path, body=body, headers=headers); response = connection.getresponse()
        raw = response.read(MAX_JSON + 1)
        if len(raw) > MAX_JSON: raise RuntimeError("API response exceeds cap")
        if response.status < 200 or response.status >= 300:
            retry = response.getheader("Retry-After")
            try: retry = min(60.0, max(0.0, float(retry)))
            except (TypeError, ValueError): retry = None
            raise APIError(f"API HTTP {response.status}", response.status, retry, hashlib.sha256(raw).hexdigest())
        try: value = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise APIError("API returned invalid JSON", response.status, body_sha256=hashlib.sha256(raw).hexdigest()) from exc
        if not isinstance(value, dict): raise RuntimeError("API response is not an object")
        return value
    finally: connection.close()

def safe_get(call, deadline, attempts=5):
    for attempt in range(attempts):
        try: return call()
        except Exception as exc:
            if attempt + 1 == attempts or time.monotonic() >= deadline: raise
            delay = getattr(exc, "retry_after", None) or min(20, 2 ** attempt) * (1 + random.random() * 0.25)
            if time.monotonic() + delay >= deadline: raise
            time.sleep(delay)

class PinnedHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, host, address):
        super().__init__(host, 443, timeout=60, context=ssl.create_default_context()); self.address = address
    def connect(self):
        sock = socket.create_connection((self.address, 443), self.timeout)
        try: self.sock = self._context.wrap_socket(sock, server_hostname=self.host)
        except BaseException:
            sock.close(); raise

def download_video(url, stage_path, allowed_hosts):
    parsed = urlsplit(url)
    host = (parsed.hostname or "").lower()
    if (parsed.scheme != "https" or parsed.username or parsed.password or parsed.fragment
            or parsed.port not in (None, 443) or not parsed.path
            or host not in allowed_hosts):
        raise RuntimeError("delivery URL host is not an exact approved host")
    try: ipaddress.ip_address(host); raise RuntimeError("IP-literal delivery URL rejected")
    except ValueError: pass
    addresses = sorted({item[4][0] for item in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM)})
    if not addresses or any(not ipaddress.ip_address(item).is_global for item in addresses):
        raise RuntimeError("delivery host did not resolve publicly")
    connection = PinnedHTTPSConnection(host, addresses[0]); created = False
    try:
        target = parsed.path + (("?" + parsed.query) if parsed.query else "")
        connection.request("GET", target, headers={"Accept": "video/mp4"}); response = connection.getresponse()
        if 300 <= response.status < 400: raise RuntimeError("delivery redirect rejected")
        if response.status != 200: raise RuntimeError(f"delivery HTTP {response.status}")
        if response.getheader("Content-Type", "").split(";", 1)[0].lower() != "video/mp4":
            raise RuntimeError("delivery MIME rejected")
        length = response.getheader("Content-Length")
        if length and (not length.isdigit() or int(length) > MAX_VIDEO): raise RuntimeError("declared size rejected")
        fd = os.open(stage_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600); created = True
        total = 0
        with os.fdopen(fd, "wb") as f:
            while True:
                chunk = response.read(1024 * 1024)
                if not chunk: break
                total += len(chunk)
                if total > MAX_VIDEO: raise RuntimeError("download crossed byte cap")
                f.write(chunk)
            f.flush(); os.fsync(f.fileno())
        return stage_path
    except BaseException:
        if created and stage_path.exists(): stage_path.unlink()
        raise
    finally: connection.close()

def inspect_mp4(path, output, mode, aspect, duration, sound):
    with path.open("rb") as f: head = f.read(12)
    if len(head) < 12 or head[4:8] != b"ftyp": raise RuntimeError("MP4 signature rejected")
    args = ["ffprobe", "-v", "error", "-show_entries",
            "format=duration,size:stream=codec_type,codec_name,width,height,r_frame_rate", "-of", "json", str(path)]
    code, out, _, timed_out, overflow = run_bounded(args, 30)
    if code or timed_out or overflow: raise RuntimeError("ffprobe failed")
    probe = json.loads(out); streams = probe.get("streams", []); videos = [s for s in streams if s.get("codec_type") == "video"]
    if len(videos) != 1: raise RuntimeError("expected one video stream")
    if sound == "on" and not any(s.get("codec_type") == "audio" for s in streams):
        raise RuntimeError("native audio was requested but absent")
    width, height = videos[0].get("width"), videos[0].get("height"); expected = 720 if mode == "std" else 1080
    if not isinstance(width, int) or not isinstance(height, int) or min(width, height) != expected:
        raise RuntimeError("resolution does not match approved mode")
    expected_ratio = {"16:9": Decimal(16) / 9, "9:16": Decimal(9) / 16, "1:1": Decimal(1)}[aspect]
    if abs(Decimal(width) / Decimal(height) - expected_ratio) > Decimal("0.02"):
        raise RuntimeError("aspect ratio does not match approval")
    try: actual = Decimal(str(probe.get("format", {}).get("duration", "")))
    except InvalidOperation as exc: raise RuntimeError("invalid media duration") from exc
    if not actual.is_finite() or abs(actual - Decimal(duration)) > Decimal("0.75"):
        raise RuntimeError("duration outside tolerance")
    code, _, _, timed_out, overflow = run_bounded(["ffmpeg", "-v", "error", "-i", str(path), "-f", "null", "-"], 300)
    if code or timed_out or overflow: raise RuntimeError("full decode failed")
    digest = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""): digest.update(chunk)
    return {"path": str(output), "bytes": path.stat().st_size, "sha256": digest.hexdigest(),
            "media": {"duration": str(actual), "width": width, "height": height,
                      "videoCodec": videos[0].get("codec_name"),
                      "audioStreams": sum(s.get("codec_type") == "audio" for s in streams)}}

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--prompt", required=True); parser.add_argument("--negative-prompt", default="")
    parser.add_argument("--duration", type=int, choices=range(3, 16), default=5)
    parser.add_argument("--aspect", choices=("16:9", "9:16", "1:1"), default="16:9")
    parser.add_argument("--mode", choices=("std", "pro"), default="pro")
    parser.add_argument("--sound", choices=("off", "on"), default="off")
    parser.add_argument("--attempt-id", required=True); parser.add_argument("--attempt-record", type=Path, default=Path("kling-attempt.json"))
    parser.add_argument("--output", type=Path, default=Path("kling-output.mp4"))
    parser.add_argument("--max-usd", required=True); parser.add_argument("--approval-sha256")
    parser.add_argument("--pricing-checked-at", required=True)
    parser.add_argument("--pricing-evidence-sha256", required=True)
    parser.add_argument("--delivery-host", action="append", required=True,
                        help="exact approved Kling delivery host; repeat for each host")
    parser.add_argument("--execute", action="store_true"); parser.add_argument("--resume", action="store_true")
    parser.add_argument("--show-full-plan", action="store_true")
    args = parser.parse_args()
    if not args.prompt.strip() or len(args.prompt) > 2500 or len(args.prompt.encode()) > 10000:
        raise SystemExit("prompt must be 1..2500 characters and at most 10000 UTF-8 bytes")
    if len(args.negative_prompt) > 2500 or len(args.negative_prompt.encode()) > 10000:
        raise SystemExit("negative prompt exceeds local cap")
    try: attempt = uuid.UUID(args.attempt_id); maximum = Decimal(args.max_usd)
    except (ValueError, InvalidOperation) as exc: raise SystemExit("attempt ID or maximum invalid") from exc
    if attempt.version != 4 or str(attempt) != args.attempt_id.lower(): raise SystemExit("attempt ID must be canonical UUIDv4")
    if (len(args.pricing_evidence_sha256) != 64
            or any(c not in "0123456789abcdef" for c in args.pricing_evidence_sha256)):
        raise SystemExit("pricing evidence must be a lowercase SHA-256")
    try: checked_at = datetime.fromisoformat(args.pricing_checked_at.replace("Z", "+00:00"))
    except ValueError as exc: raise SystemExit("pricing timestamp must be ISO-8601") from exc
    if checked_at.tzinfo is None: raise SystemExit("pricing timestamp must include a timezone")
    allowed_hosts = sorted(set(host.lower().rstrip(".") for host in args.delivery_host))
    if (not allowed_hosts or len(allowed_hosts) > 8
            or any(not host or len(host) > 253 or "*" in host or "/" in host or "@" in host for host in allowed_hosts)):
        raise SystemExit("delivery hosts must be 1..8 exact DNS names")
    for host in allowed_hosts:
        try: ipaddress.ip_address(host); raise SystemExit("delivery hosts cannot be IP literals")
        except ValueError: pass
    estimate = RATES[(args.mode, args.sound)] * args.duration
    if not maximum.is_finite() or maximum <= 0 or maximum < estimate: raise SystemExit("maximum must be finite and cover estimate")
    output, record_path = args.output.resolve(), args.attempt_record.resolve()
    stage_path = output.with_name("." + output.name + ".kling-stage")
    if output.suffix.lower() != ".mp4" or output == record_path or stage_path in {output, record_path}:
        raise SystemExit("output path rejected")
    request = {"model_name": MODEL, "prompt": args.prompt, "negative_prompt": args.negative_prompt,
               "duration": str(args.duration), "mode": args.mode, "aspect_ratio": args.aspect,
               "sound": args.sound, "external_task_id": str(attempt)}
    request_hash = hashlib.sha256(canonical(request)).hexdigest()
    envelope = {"backend": "kling-direct-international", "endpoint": "/v1/videos/text2video", "host": API_HOST,
        "request": request, "attemptId": str(attempt), "attemptRecord": str(record_path),
        "outputPolicy": {"path": str(output), "stagePath": str(stage_path), "exactDeliveryHosts": allowed_hosts,
                         "maxBytes": MAX_VIDEO, "mime": "video/mp4", "fullDecode": True, "noOverwrite": True},
        "cost": {"pricingCheckedAt": checked_at.isoformat(), "pricingEvidenceSha256": args.pricing_evidence_sha256,
                 "rateUsdPerSecond": str(RATES[(args.mode, args.sound)]),
                 "seconds": args.duration, "creates": 1, "estimatedMaxUsd": str(estimate), "approvedMaxUsd": str(maximum)}}
    approval = hashlib.sha256(canonical(envelope)).hexdigest()
    safe_request = {**request, "prompt": "sha256:" + hashlib.sha256(args.prompt.encode()).hexdigest(),
                    "negative_prompt": "sha256:" + hashlib.sha256(args.negative_prompt.encode()).hexdigest()}
    print(json.dumps({"dryRun": not args.execute, "approvalSha256": approval,
        "plan": {**envelope, "request": safe_request}}, indent=2))
    if args.show_full_plan: print(json.dumps({"protectedFullPlan": envelope}, indent=2))
    if not args.execute: return 0
    if args.approval_sha256 != approval: raise SystemExit("exact approval digest mismatch")
    age = datetime.now(timezone.utc) - checked_at.astimezone(timezone.utc)
    if age.total_seconds() < -300 or age.total_seconds() > 86400:
        raise SystemExit("pricing evidence must be rechecked within 24 hours of execution")
    key = os.getenv("KLING_API_KEY")
    if not key: raise SystemExit("set server-side KLING_API_KEY")
    if args.resume:
        if not record_path.exists(): raise SystemExit("resume record missing")
        record = read_small_json(record_path)
        if record.get("approval_sha256") != approval or record.get("request_sha256") != request_hash:
            raise SystemExit("resume does not match exact request")
        task_id = record.get("task_id")
        if not isinstance(task_id, str) or not task_id: raise SystemExit("no known task ID; do not replay create")
        expected_artifact = record.get("artifact_staged") or record.get("artifact")
        candidate = output if output.exists() else stage_path if stage_path.exists() else None
        if candidate is not None:
            if not isinstance(expected_artifact, dict):
                if record.get("status") == "downloading" and candidate == stage_path:
                    stage_path.unlink(); fsync_directory(stage_path.parent)
                else: raise SystemExit("unclaimed output/stage exists; refusing overwrite or adoption")
            else:
                artifact = inspect_mp4(candidate, output, args.mode, args.aspect, args.duration, args.sound)
                if (artifact.get("sha256") != expected_artifact.get("sha256")
                        or artifact.get("bytes") != expected_artifact.get("bytes")):
                    raise SystemExit("recovered artifact does not match durable staged evidence")
                if candidate == stage_path:
                    if output.exists(): raise SystemExit("output appeared during recovery")
                    os.replace(stage_path, output); fsync_directory(output.parent)
                record.update(status="artifact_saved", artifact=artifact, updated_unix=int(time.time()))
                record.pop("artifact_staged", None); atomic_json(record_path, record)
                print(json.dumps(record, indent=2)); return 0
    else:
        if output.exists() or stage_path.exists(): raise SystemExit("output or stage already exists")
        record = {"attempt_id": str(attempt), "approval_sha256": approval, "request_sha256": request_hash,
                  "status": "posting", "task_id": None, "created_unix": int(time.time())}
        try: claim_once(record_path, record)
        except FileExistsError as exc: raise SystemExit("attempt already claimed; use resume") from exc
        try: response = api_json("POST", "/v1/videos/text2video", key, request, 120)
        except Exception as exc:
            known_rejection = (isinstance(exc, APIError) and exc.status is not None and 400 <= exc.status < 500
                               and exc.status not in {408, 409, 425, 429})
            record.update(status="create_rejected" if known_rejection else "create_outcome_unknown",
                          error_type=type(exc).__name__, http_status=getattr(exc, "status", None),
                          error_body_sha256=getattr(exc, "body_sha256", None))
            atomic_json(record_path, record)
            if known_rejection: raise RuntimeError("provider rejected create before acceptance") from exc
            raise RuntimeError("create outcome unknown; reconcile account, never replay") from exc
        if response.get("code") != 0:
            record.update(status="create_rejected", service_code=response.get("code")); atomic_json(record_path, record)
            raise RuntimeError("provider rejected create")
        data = response.get("data") or {}; task_id = data.get("task_id")
        if (not isinstance(task_id, str) or not 1 <= len(task_id) <= 200
                or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_." for c in task_id)):
            record.update(status="create_outcome_unknown", error_type="MissingOrInvalidTaskId"); atomic_json(record_path, record)
            raise RuntimeError("create returned no safe task ID; reconcile")
        record.update(status="task_id_saved", task_id=task_id, request_id=response.get("request_id"), updated_unix=int(time.time()))
        atomic_json(record_path, record)
    deadline = time.monotonic() + 1200
    while True:
        response = safe_get(lambda: api_json("GET", "/v1/videos/text2video/" + task_id, key, timeout=45), deadline)
        if response.get("code") != 0: raise RuntimeError("task query failed")
        data = response.get("data") or {}; status = data.get("task_status")
        if status == "succeed": break
        if status == "failed":
            message = str(data.get("task_status_msg", ""))
            record.update(status="generation_failed", task_status_msg_sha256=hashlib.sha256(message.encode()).hexdigest())
            atomic_json(record_path, record)
            raise RuntimeError("generation failed")
        if status not in {"submitted", "processing"}: raise RuntimeError("unknown task status")
        if time.monotonic() >= deadline: raise TimeoutError("generation exceeded 20 minutes")
        time.sleep(8 + random.random() * 2)
    videos = ((data.get("task_result") or {}).get("videos") or [])
    if len(videos) != 1 or not isinstance(videos[0], dict) or not isinstance(videos[0].get("url"), str):
        raise RuntimeError("expected exactly one video URL")
    url = videos[0]["url"]; output.parent.mkdir(parents=True, exist_ok=True)
    record.update(status="downloading", delivery_url_sha256=hashlib.sha256(url.encode()).hexdigest(), stage_path=str(stage_path))
    atomic_json(record_path, record)
    try:
        download_video(url, stage_path, set(allowed_hosts))
        artifact = inspect_mp4(stage_path, output, args.mode, args.aspect, args.duration, args.sound)
    except BaseException as exc:
        record.update(status="artifact_validation_failed", error_type=type(exc).__name__); atomic_json(record_path, record); raise
    record.update(status="artifact_staged", artifact_staged=artifact, final_unit_deduction=data.get("final_unit_deduction"),
        remote_purge_expected_within_days=30, rights_review="required", creative_qa="pending",
        ai_disclosure="required", provenance_credentials="not_documented", updated_unix=int(time.time()))
    atomic_json(record_path, record)
    if output.exists(): raise RuntimeError("output appeared before publication")
    os.replace(stage_path, output); fsync_directory(output.parent)
    record.update(status="artifact_saved", artifact=artifact, updated_unix=int(time.time()))
    record.pop("artifact_staged", None); atomic_json(record_path, record)
    print(json.dumps(record, indent=2)); return 0

if __name__ == "__main__": raise SystemExit(main())

ドライランを実行し、表示されたダイジェストを承認します。

python kling_generate.py --attempt-id 7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4 --max-usd 0.56 --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 "$BILLING_SHA256" --delivery-host "$KLING_DELIVERY_HOST" --prompt "Single continuous five-second macro dolly shot of steam rising from a ceramic cup; rain outside; no people, text, or logo."

変更していないリクエストを 1 回実行します。

python kling_generate.py --execute --approval-sha256 "<exact digest>" --attempt-id 7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4 --max-usd 0.56 --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 "$BILLING_SHA256" --delivery-host "$KLING_DELIVERY_HOST" --prompt "Single continuous five-second macro dolly shot of steam rising from a ceramic cup; rain outside; no people, text, or logo."

BILLING_SHA256 は、実際に確認した日付入りのファーストパーティ料金証拠の SHA-256 です。KLING_DELIVERY_HOST は、現在の直接 API 配信契約で許可されている正確なホスト名であり、ワイルドカードや信頼できない URL から推定したホストではありません。追加で正確なホストを許可する場合に限り、--delivery-host を繰り返し指定してください。実行時には、24 時間を超えた料金証拠は拒否されます。

同じ引数に --resume --execute を加えて使用できるのは、記録にすでにタスク ID がある場合のみです。posting または create_outcome_unknown の状態を復旧するために --resume を使用しないでください。まず external_task_id、アカウントの使用量、リクエスト ID、ログ、またはサポート窓口を通じて照合してください。再開処理では、バイト列と SHA-256 が永続的な台帳の証拠と一致する、ステージ済みまたは公開済みのアーティファクトのみを採用できます。ドキュメントの UI は現在再構成中のため、本番環境で使用する前に、稼働中の kling-v3 テキストから動画へのページでエンドポイントとフィールドの型を確認してください。

データ、権利、リリースを管理する

事実： 有料 API の規約では、Kling はデータ処理者として機能し、有料 API のプロンプト／ファイル／応答をモデルの改善に使用せず、サービス／安全上の目的でプロンプト／応答を 30 日間ログに記録するとされています。API のプライバシーポリシーでは、データはシンガポールのサーバーに保存され、運用チーム／プロバイダーが国際的にアクセス／転送する場合があるとされています。法的、コンプライアンス、紛争、または正当な目的のために、保持期間が延長される場合があります。これはゼロ保持でも EU 域内でのデータ保管でもありません。API の有料規約、API のプライバシーポリシー

事実： 有料 API の規約では商用利用が認められていますが、オペレーターは入力に関する権利／許諾を保有する必要があり、生成された出力も権利侵害となる可能性があると警告されています。所有権／著作権の成立可否は、適用される法律によって異なります。Kling の一般規約では、書面による許可がない限り、出力を配布する際は Kling AI であることを目立つように明示する必要があります。適用されるアカウント／API 契約のうち、より厳格なものと、地域の合成メディアに関する法律を適用してください。「ウォーターマークなし」が開示不要を意味すると推測してはいけません。API の有料規約、Kling の利用規約

確認したファーストパーティの情報源には、Kling の動画を直接出力する際の C2PA や、永続的な暗号署名付き来歴証明に関する保証はありません。来歴は不明として扱ってください。リクエスト／ソースのハッシュと元ファイルを保持し、明確な AI 開示と必要なブランドマークを追加してください。ただし、それらがプロバイダーによって署名されていると主張してはいけません。

リリース前に、すべてのフレームを確認し、最初から最後まで音声を聴いてください。寸法、長さ、フレームレート、音声ストリーム、完全なデコード、ラウドネス／ピーク、台詞／リップシンク／話者の割り当て、時間的な整合性、顔／手／解剖学的表現、アイデンティティ／衣装／製品の一貫性、ロゴ／テキスト／色、安全でないコンテンツ、事実に関する主張／広告上の主張、権利、開示、プラットフォームの仕様を確認してください。技術的に有効な MP4 は、承認済みのクリエイティブマスターではありません。

フェーズごとの失敗への対処

• POST の前： スキーマ、権利、認証、残高、モデルへのアクセス、または明示的な承認を修正します。生成リクエストはまだ受理されていません。
• POST がタスク ID を返した場合： その ID を保存し、そのタスクを照会します。処理の遅延、429、500、503、504 が発生しても、2 回目の作成リクエストが許可されるわけではありません。
• POST の結果が不明な場合： 停止してください。external_task_id は照合用の識別子であり、冪等性が保証されていると文書で明示されているものではありません。
• 安全性に関する拒否： 制御を回避するために言い換えたり、再販業者を経由したりしないでください。正当性と同意を再評価してください。
• タスクの失敗またはモーションの短縮： ステータスと実際の差し引き額を記録してください。返金を約束してはいけません。
• 配信 URL またはメディアが無効な場合： 一時ファイルを隔離／削除し、サニタイズした証拠を保持して、公開しないでください。新たに有料で生成するには、新しい試行 ID と承認が必要です。
• プレビュー／更新内容に差異がある場合： 公式の認証、モデル、課金、更新、ポリシーの各ページを再確認してください。代替 ID を推測したり、ゲートウェイの契約を直接 API に混在させたりしないでください。

ラベルを正確に使う

• FACT： 現行の公式 API のフィールド、モデルの機能、価格、利用条件、または制限です。
• PROVIDER CLAIM： Kling／Kuaishou の品質、ベンチマーク、「ネイティブ」、一貫性、物理特性、効率に関する、独立して検証されていない主張です。
• OBSERVATION： サーフェス、モデル、リクエストハッシュ、入力、アカウントのリージョン、出力チェックを含む、再現可能なタスクでの日時付きの結果です。
• HEURISTIC： アクションの制限、モーション参照のフレーミングの一致、編集ごとに変更を1つにするなど、本番環境でのアドバイスです。
• UNKNOWN： 文書化されていない確定的なレイテンシー、正確な作成処理の冪等性、API からの直接的な C2PA、保証されたアイデンティティ／テキスト／音声の精度、または現在のタスクページに表示されていないフィールドです。

公式情報を再確認してください：API overview、authentication/errors、billing、3.0 task index、3.0 guide、3.0 Omni guide、Motion Control guide、API paid terms、API privacy、Kling-Omni report、Motion Control report。