HiAPI で Kandinsky 6.0 Pro のテキストto動画を使用する：curl と Python チュートリアル

kandinsky-6.0-pro/text-to-video リクエストの役割

kandinsky-6.0-pro/text-to-video はテキストプロンプトを短い動画クリップに変換します。プロンプトといくつかの生成パラメータを送信すると、モデルが保存して再生できる動画ファイルを生成します。これが製品の全体像です：プロンプトを入力し、動画クリップを出力します。

重要なアーキテクチャ上の詳細は、これが同期呼び出しではないことです。最初のリクエストのレスポンスボディで直接動画を取得することはありません。代わりにジョブを作成し、API がそのジョブの識別子を返し、モデルがその後バックグラウンドで処理を行います。コードはジョブの状態をチェックし、完了が報告されるまで待つ必要があり、その時点で結果をダウンロードできます。

このパターンには、最初から設計に組み込むべきいくつかの帰結があります：

• クライアントはポーリングする必要がある。単一の POST で完成した動画を返すことはできないため、どのような統合でも状態ポーリングループまたは webhook 的な再チェックが必要です。
• リクエストはこちら側で冪等であるべき。送信後にスクリプトがクラッシュした場合、気付かないうちに 2 つ目のジョブを作成したくはありません。ジョブ識別子を取得し、永続化して保存してください。
• レイテンシは変動する。動画生成は画像やテキストの呼び出しより遅いため、待機は正常な現象であり、欠陥ではありません。

本チュートリアルの残りの部分は、このライフサイクルに沿って展開します：準備、curl による送信、ポーリング、ダウンロード、そして Python で同じフローを再構築します。コードは動作する骨組みとして扱ってください——エンドポイントパスとフィールド名は常にそのモデルの API リファレンスと照合して確認すべきです。これらの詳細はプラットフォームの進化とともに変わるからです。

最初のリクエストを送信する前に

初回呼び出しの前に準備すべきことが 4 つあります。

1. HiAPI アカウントと API キー

HiAPI アカウントから発行された API キーで認証します。キーは可能な限りソースコード管理やシェル履歴から除外してください。一般的な方法は、環境変数としてエクスポートし、コードからそこを読み取ることです：

export HIAPI_API_KEY="your-key-here"

以下のすべてのリクエストは、キーが bearer token として Authorization ヘッダーに含まれて送信されることを前提としています。キーをまだ作成していない場合は、先にアカウントダッシュボードで作成してください。

2. HTTP クライアント

curl でリクエストが機能するか検証するには十分です。繰り返し実行する必要のある操作には、Python 3 と requests ライブラリを使用してください——依存関係は 1 つだけ、コードは読みやすく、後でスケジューラやサービスに組み込みやすいです：

python3 -m pip install requests

3. 出力の保存先

動画ファイルは JSON レスポンスよりはるかに大きくなります。ローカルディレクトリに書き込むか、オブジェクトストレージにストリーミングするかを事前に決めてください。本チュートリアルでは、ローカルの ./output ディレクトリで十分です：

mkdir -p output

4. プロンプトに慣れておく

コードを書く前にプロンプトを書いてください。具体的で記述的なプロンプト——被写体、動作、カメラの動き、視覚スタイル——は、問題が発生したときに比較できるベースラインを提供します。プロンプトがシェルプロンプトで入力する文字列ではなく、スクリプト内で名前付き定数になっていれば、一度に 1 つの変数だけを変更するのがはるかに簡単になります。

以上が前提条件の完全なリストです。SDK のインストールもビルドステップも不要で、プラットフォームは通常の HTTPS 経由でアクセスします。

動作する curl リクエスト

最初の目標は、タスク（動画ではなく）を返すコマンドを 1 つ得ることです。コードを追加する前に、まずこれを動作させてください。

リクエストは 3 つの部分で構成されます：モデルのエンドポイント、Authorization ヘッダー、そしてプロンプトと生成パラメータを含む JSON リクエストボディです。

curl -X POST "https://api.hiapi.ai/v1/models/kandinsky-6.0-pro/text-to-video/jobs" \
  -H "Authorization: Bearer $HIAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A slow drone shot over a misty pine forest at sunrise, soft golden light, cinematic",
    "duration_seconds": 5,
    "aspect_ratio": "16:9"
  }'

レスポンスを注意深く読んでください。探すべきものは：

• ジョブ識別子——ポーリングに使用するハンドルです。id、job_id などと命名されている可能性があります。
• 初期の状態フィールド。通常は queued や processing のような値です。
• リクエストボディが意図通りに解析されたことを確認するためのエコーバックされたパラメータ。

ジョブ識別子とエラーでない状態を返すレスポンスは、リクエストが受け入れられたことを意味します。そのレスポンスに動画ファイルが直接含まれている場合、ここで説明しているものとは異なることが起きています。

コントラクトに基づいて構築する前に確認する

上記のフィールド名——エンドポイントパス、duration_seconds、aspect_ratio、識別子のキー——は、モデルや時期によって最も変わりやすい詳細です。信頼できる真実の源は、そのモデルの API リファレンスページです。一度確認し、正確な名前を定数としてスクリプトにコピーし、名前の変更がコードベース全体の検索ではなく 1 行の編集で済むようにコードを構成してください。

4xx を受け取った場合、エラーレスポンスボディはステータスコードよりも多くの情報を提供します。401 はキーが誤っているか欠落していることを意味し、422 系の検証エラーは通常、問題のあるフィールドを指摘します。

ジョブを送信し、完了までポーリングする

送信呼び出しが機能したら、ライフサイクルは：送信、識別子の取得、ポーリング、終端状態で停止、となります。

送信

前節の POST がジョブを作成します。コードが次に何をするにせよ、返された識別子を永続的な場所——スクリプト内の変数、タスクテーブルの行、キューのメッセージ——に保存すべきです。送信レスポンスを受け取った後、識別子を永続化する前にプロセスが死ぬと、そのジョブをポーリングする能力を失います。

ポーリング

ポーリングは、取得した識別子を使用してジョブリソースに対して GET リクエストを送信することです：

curl "https://api.hiapi.ai/v1/models/kandinsky-6.0-pro/text-to-video/jobs/$JOB_ID" \
  -H "Authorization: Bearer $HIAPI_API_KEY"

レスポンスはジョブの現在の状態を報告します。実際に目にする状態は少数です：キュー待ちまたは保留中、処理中、完了、そして何らかの形の失敗またはキャンセルです。ループは正確な文字列ではなくこれらのカテゴリを中心に設計してください——API が返すものをコード内の running、done、failed にマッピングすれば、名前変更でリトライロジックが壊れることはありません。

どれだけ待つかを決める

重要な数値は 2 つです：ポーリング間隔と全体のタイムアウトです。

• 間隔。数秒ごとにポーリングし、タイトなループにはしないでください。タイトなループは追加の利点をもたらさず、レート制限を引き起こす可能性があります。動画ジョブでは、リクエスト間に数秒の間隔を置くのが妥当です。
• タイムアウト。上限を設定してください。動画生成には時間がかかりますが、「永遠に待つ」は本番スクリプトとして許容できる答えではありません。実際的な最大時間予算を選択し、超過は記録して意識的にリトライする失敗として扱ってください。

実際の運用に耐えるパターン

本番環境では、ポーリングステップは通常、スリープするプロセス内ではなく、送信ステップと分離されます。ワーカープロセスが送信し、ジョブを記録して終了します。定期的な再チェックが未完了のジョブを拾ってポーリングします。これは再起動に対してより堅牢で、監視も容易です。本チュートリアルでは、単一プロセス内のシンプルなループで十分です——ただし、インラインコードではなく関数として構成しておけば、後で定期スケジュールのワーカープロセスに移すのは簡単です。

生成された動画クリップをダウンロードする

完了したジョブはまだディスク上のファイルではありません。完了したジョブの状態レスポンスは出力を指し示します——通常は URL、時には URL を含むネストされたオブジェクトです。そのフィールドを読み取り、ダウンロードステップの入力として使用します。

curl -L "$OUTPUT_URL" -o output/clip.mp4

ダウンロードが実際に成功したことを確認するために、以下の点をチェックしてください：

• リダイレクトを追跡する。出力 URL はリダイレクト経由で提供されることがよくあります。curl では -L、Python では allow_redirects=True で処理します。
• コンテンツタイプを検証する。完了したジョブは動画ペイロードを生成するはずです。clip.mp4 として保存されたものが JSON や HTML のエラーページであれば、ダウンロードに問題があり、生成プロセスではありません。
• サイズを確認する。ゼロバイトまたは疑わしく小さいファイルはダウンロード失敗です。レスポンスヘッダーが存在する場合、バイト数をヘッダーで宣言されたサイズと比較してください。
• 有効期限を考慮する。出力 URL には通常時間制限があります。完了後すぐにダウンロードし、永続性が必要な場合は URL を永続的な参照として保存するのではなく、ファイルを独自のストレージにコピーしてください。

チュートリアル以外の用途では、ファイル全体をメモリに読み込むのではなく、ダウンロードを宛先にストリーミングし、一時ファイル名に書き込んでからリネームしてください——そうすれば未完了のダウンロードが下流のコードに完全なファイルのように見えることは決してありません。

同じリクエストを Python で実装する

以下は curl のフローを呼び出し可能な関数として再構築したものです：送信、ポーリング、ダウンロード。調整が必要な名前は先頭に定数としてまとめています。

import os
import time
import requests

API_BASE = "https://api.hiapi.ai/v1"
MODEL_PATH = "/models/kandinsky-6.0-pro/text-to-video"
API_KEY = os.environ["HIAPI_API_KEY"]

HEADERS = {"Authorization": f"Bearer {API_KEY}"}

RUNNING_STATES = {"queued", "pending", "processing", "running"}
DONE_STATES = {"completed", "succeeded", "done"}
FAILED_STATES = {"failed", "cancelled", "error"}


def generate_clip(prompt, out_path, timeout_s=600, interval_s=5):
    # 1. submit
    resp = requests.post(
        API_BASE + MODEL_PATH + "/jobs",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"prompt": prompt, "duration_seconds": 5, "aspect_ratio": "16:9"},
        timeout=30,
    )
    resp.raise_for_status()
    job = resp.json()
    job_id = job["id"]
    print(f"submitted job {job_id}")

    # 2. poll
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        status_resp = requests.get(
            f"{API_BASE}{MODEL_PATH}/jobs/{job_id}",
            headers=HEADERS,
            timeout=30,
        )
        status_resp.raise_for_status()
        data = status_resp.json()
        state = str(data.get("status", "")).lower()

        if state in DONE_STATES:
            break
        if state in FAILED_STATES:
            raise RuntimeError(f"job {job_id} ended in state {state}: {data}")
        time.sleep(interval_s)
    else:
        raise TimeoutError(f"job {job_id} did not finish within {timeout_s}s")

    # 3. download
    url = data["output"]["url"] if isinstance(data.get("output"), dict) else data["output"]
    with requests.get(url, stream=True, allow_redirects=True, timeout=60) as r:
        r.raise_for_status()
        tmp = out_path + ".part"
        with open(tmp, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 16):
                f.write(chunk)
        os.replace(tmp, out_path)
    print(f"saved {out_path}")
    return job_id


if __name__ == "__main__":
    generate_clip(
        "A slow drone shot over a misty pine forest at sunrise, soft golden light, cinematic",
        "output/clip.mp4",
    )

なぜこのように書くのか

• 正確な一致ではなく状態集合を使用する。返された状態を running/done/failed にマッピングすることで、新しい中間状態がループを壊すことはありません。
• 単調デッドライン。time.monotonic() は時計調整の影響を受けないため、タイムアウトが正確です。
• 原子的書き込み。.part に書き込んでからリネームすることで、クラッシュしたダウンロードが有効に見える切り詰めファイルを残すことは決してありません。
• ジョブ ID を返す。呼び出し元は記録、保存し、必要に応じて再度ポーリングできます——これはポーリングをこのプロセスから外に移す最初のステップです。

エンドポイントパス、リクエストフィールド名、レスポンスキーはモデルの API リファレンスに合わせて調整してください。それ以外のすべて——ループ、タイムアウト、ダウンロード——は汎用的です。

エラー状態とデバッグチェックリスト

実行が失敗した場合、何かを変更する前にまずフェーズを切り分けてください。各フェーズには異なる失敗の特徴があります。

認証失敗

送信時の 401 または 403 は、キーが欠落、誤り、または送信されていないことを意味します。コードを実行しているシェルまたはプロセスで環境変数が実際に設定されているか確認してください——あるターミナルでエクスポートしたキーは cron ジョブには見えません。キーの最初の数文字を出力して解析されたことを確認し、完全なキーは決して出力しないでください。

リクエスト形式エラー

送信時の 400 または 422 系のエラーは、リクエストボディが検証に失敗したことを意味します。レスポンスボディは通常、問題のあるフィールドを指摘します。一般的な原因：パラメータのスペルミス、サポート範囲外の長さ、または空のプロンプト。JSON を API リファレンスとフィールドごとに比較してください。単一の名前変更されたキーが最も一般的な原因です。

ジョブが実行状態でスタックする

ポーリングが永遠に終端状態に到達しない場合、3 つのことを確認してください：ポーリングしている識別子が送信したものと一致しているか；ポーリング間隔が頻繁すぎてレート制限されていないか；タイムアウトがモデルに対して十分に長いか。各ポーリングで生の状態レスポンスを記録し、状態が進んでいるのか凍結しているのかを確認してください。

ジョブは完了したがダウンロードが失敗する

ジョブは完了したがダウンロードが失敗するのは、生成の問題ではなくファイル処理の問題です。リダイレクトを追跡しているか、URL が期限切れでないか、コンテンツタイプが JSON ではなく動画かどうかを確認してください。curl -L でその URL を単独で試し、スクリプトとは切り離してトラブルシューティングしてください。

同じプロンプトが繰り返し失敗する

あるプロンプトが継続的に失敗し、他のプロンプトが成功する場合、インフラではなくコンテンツポリシーまたはプロンプト長を先に疑ってください。1 つの変数を変更して——プロンプトを短くする、シーン記述を簡素化する——再試行してください。

一般的なルール

各ジョブについて 4 つのことを記録してください：ジョブ識別子、最終状態、実際の所要時間、ダウンロードの HTTP ステータス。この 4 つの値があれば、ほとんどの「うまくいかなかった」という報告を 1 分以内に具体的で修正可能なフェーズに変換できます。