チュートリアル
HiAPI で flux-1-kontext-max を呼び出す方法：curl と Python のチュートリアル
実行可能なリクエストの全フロー：入力、非同期ポーリング、出力のダウンロード
flux-1-kontext-max のリクエストの形
コードを書く前に、やり取りの形を理解しておくと役立ちます。HiAPI で flux-1-kontext-max への画像編集呼び出しは、同一の HTTP 呼び出し内で画像を返す単純なリクエスト/レスポンスではありません。二段階のタスクです：
- 送信 —— 入力画像と目的の編集を記述した JSON ペイロードを POST します。API はタスクを受け付け、タスク識別子を返します。
- ポーリング —— その識別子でタスクのステータスを GET し、完了が報告されるまで待ってから、レスポンスから出力 URL を読み取ります。
このパターンが採用されているのは、画像編集が計算負荷の高い処理だからです。レンダリング中ずっと単一の HTTP 接続を開いたままにするのは脆い方法です。タスクハンドルがあれば、切断し、リトライし、後で戻ってくることができます。
ペイロード自体は概念的には三つの部分に分かれます：
- モデルセレクタ —— 実行するモデルとして
flux-1-kontext-maxを指定するフィールド。 - 入力 —— 通常は一枚の画像（ドキュメント化されたスキーマに応じて URL または base64 ペイロード）と、編集を記述するテキスト指示。
- オプションの制御項目 —— モデルのドキュメントに記載があれば、アスペクト比やシードなど生成動作を調整するパラメータ。
それ以外のすべて——認証ヘッダー、送信エンドポイント、ステータスエンドポイント——は、HiAPI のすべてのタスクで使うのと同じパイプラインです。この三拍子（送信 → ポーリング → ダウンロード）を身につければ、curl 版と Python 版は交換可能で、フローを学び直すことなく切り替えられます。
前提条件と認証
最初のリクエストを成功させる前に、いくつか準備が必要です。
必要なもの
- HiAPI アカウント、およびダッシュボードで作成した API key。
- プラットフォームのドキュメントに記載された HiAPI API の ベース URL。
- API がアクセスできる入力画像——最も一般的には公開 HTTPS URL。画像がプライベートな場合は、API が取得できる場所にアップロードするか、モデルスキーマが base64 入力に対応していればインラインで渡します。
- シェルでの実習用の
curl、コード実習用の Python 3 とrequestsライブラリ。
認証
HiAPI は認可ヘッダーで API key を送信してリクエストを認証します。慣例は bearer token です：
Authorization: Bearer $HIAPI_API_KEY
また、すべての JSON リクエストはコンテンツタイプを宣言する必要があります：
Content-Type: application/json
キーをソースコードから遠ざける
キーはスクリプトに貼り付けるのではなく、環境変数に保存します：
export HIAPI_API_KEY="your_key_here"
Python では os.environ["HIAPI_API_KEY"] で読み取ります。curl では $HIAPI_API_KEY として参照します。この一つの習慣が、認証情報漏洩の最もよくある原因——キーをリポジトリにコミットしてしまい、圧力の下でローテーションを強いられる事態——を防ぎます。
リクエストが認証エラーを返した場合は、三つのことを順に確認します：キーが存在し有効であること、ヘッダーの綴りがドキュメントと完全に一致すること、環境変数に余分な空白や改行が混入していないこと。
curl でリクエストを送信する
以下は送信呼び出しです。自分の画像 URL と編集指示に置き換えてください。
curl -X POST "$HIAPI_BASE_URL/jobs" \
-H "Authorization: Bearer $HIAPI_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "flux-1-kontext-max",
"input": {
"image": "https://example.com/input.jpg",
"prompt": "Replace the sky with a dramatic sunset, keep the buildings unchanged"
}
}'
各部分の役割
-X POST—— 送信は作成操作なので POST を使います。- エンドポイント —— jobs エンドポイントは新しいタスクを受け付けます。ベース URL は環境から取得するため、コマンドは異なる環境間で移植可能です。
Authorization—— bearer token をインラインではなく環境変数から読み取ります。Content-Type: application/json—— 必須。これがないとサーバーがリクエストボディを拒否したり誤解析したりする可能性があります。-d—— JSON ペイロード。modelがflux-1-kontext-maxを選択し、input.imageが入力画像を指し、input.promptが編集を記述します。
レスポンスの読み取り
成功した送信は、タスク識別子（通常は id や job_id のようなフィールド）と、タスクがキューに入ったまたは処理中であることを示すステータスを含む JSON ボディを返します。その識別子を保存してください。ポーリングはそれに依存します。
curl -s -X POST "$HIAPI_BASE_URL/jobs" \
-H "Authorization: Bearer $HIAPI_API_KEY" \
-H "Content-Type: application/json" \
-d @payload.json | python -m json.tool
python -m json.tool にパイプするとレスポンスが整形表示され、タスク ID が見つけやすくなります。リクエストボディを payload.json に入れておけば、プロンプトが長くなってもコマンドは読みやすいままです。
この時点でタスクは送信済みですが、まだ画像は受け取っていません。画像はポーリング後に返されます。
Python での同じリクエスト
Python 版は同じロジックを担い、長く残すつもりのものにはより快適な操作性を提供します。
import os
import requests
BASE_URL = os.environ["HIAPI_BASE_URL"]
API_KEY = os.environ["HIAPI_API_KEY"]
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
}
payload = {
"model": "flux-1-kontext-max",
"input": {
"image": "https://example.com/input.jpg",
"prompt": "Replace the sky with a dramatic sunset, keep the buildings unchanged",
},
}
response = requests.post(
f"{BASE_URL}/jobs",
headers=headers,
json=payload,
timeout=30,
)
response.raise_for_status()
job = response.json()
print("job id:", job["id"])
curl 呼び出しと一致する理由
json=payload パラメータは辞書をシリアライズし、コンテンツタイプを自動設定します——送信されるバイトは curl と同じです。timeout=30 は送信リクエストを制限し、ハングした接続がプロセスを引きずるのではなく大きな音で失敗するようにします。raise_for_status() は 4xx や 5xx を、静かにエラードキュメントを返すのではなく、捕捉可能な例外に変えます。
重要な小さな違い
- 文字列の代わりに辞書 —— ペイロードを Python オブジェクトとして構築するので、プロンプトの計算や変数の注入が容易になります。
- 例外処理 —— 呼び出しを
try/except requests.HTTPErrorで包み、レスポンスボディをログに記録します。エラーペイロードは通常、入力のどこが問題かを説明してくれます。 - シェルエスケープ不要 —— 引用符付きの長いプロンプトをエスケープする必要がなくなり、エラーの一カテゴリを丸ごと排除できます。
ここで返されるタスク ID は、そのまま以下のポーリングステップに進みます。タスクは curl ではなく Python から送信されたからといって何も変わりません。
非同期タスクを処理し結果をポーリングする
送信はすぐに返りますが、レンダリングはそうではありません。タスク ID のステータスエンドポイントをポーリングして待ちます。
ポーリングループ
import time
def wait_for_job(job_id, headers, interval=3, max_attempts=100):
url = f"{BASE_URL}/jobs/{job_id}"
for attempt in range(max_attempts):
r = requests.get(url, headers=headers, timeout=30)
r.raise_for_status()
data = r.json()
status = data.get("status")
if status == "succeeded" or status == "completed":
return data
if status == "failed" or status == "error":
raise RuntimeError(f"Job failed: {data}")
time.sleep(interval)
raise TimeoutError(f"Job {job_id} did not finish in time")
タスクステータスの解釈
小さな語彙を想定してください：作業開始前の queued/pending 状態、モデルのレンダリング中の processing/running 状態、そして 終端 状態——成功または失敗です。API によっては綴りが微妙に異なることがあります。推測せず、実際に受け取ったレスポンスを確認してこれらの文字列をマッピングしてください。重要なルール：タスクが完了したとみなすのは、ステータスが終端の成功状態であり、かつ出力 URL が存在する場合のみです。 ステータスエンドポイントが 200 を返すことは、画像が準備できたことを意味しません。
ポーリングのマナー
- リクエストの間隔を空ける。 画像タスクではポーリングごとに数秒空ければ十分です。100 ミリ秒ごとのポーリングはレート制限を浪費するだけで、何の利益もありません。
- 待機時間を制限する。
max_attemptsがループを制限し、スタックしたタスクがコードを永遠にハングさせるのではなく失敗させるようにします。 - 429 にはバックオフ。 レート制限に引っかかった場合は、激しく叩くのではなく、より長くスリープしてからリトライします。
- webhook が提供されているなら優先する。 プラットフォームのドキュメントにコールバック機構が記載されていれば、ポーリングを完全に排除でき、規模に対しても優しいです。
シェルでのテストでは、簡単なループが同じように機能します：
JOB_ID="your_job_id"
while true; do
BODY=$(curl -s "$HIAPI_BASE_URL/jobs/$JOB_ID" -H "Authorization: Bearer $HIAPI_API_KEY")
echo "$BODY"
echo "$BODY" | grep -q '"succeeded"' && break
sleep 3
done
ループを抜けたら、終端レスポンスが出力参照を保持しています。それが次のステップです。
出力をダウンロードして保存する
完了したタスクのレスポンスはレンダリング後の画像を指し、通常は URL（時にはリスト）です。二つ目のリクエストで取得し、バイトをディスクに書き込みます。
import pathlib
def download_output(job_data, path="edit.png"):
out = job_data.get("output")
if isinstance(out, list):
out = out[0]
url = out["url"] if isinstance(out, dict) else out
r = requests.get(url, timeout=60)
r.raise_for_status()
pathlib.Path(path).write_bytes(r.content)
return path
注意すべきポイント
- 形をハードコードしない。 出力フィールドはモデルによって異なります——文字列 URL のことも、
urlキーを持つオブジェクトのことも、リストのこともあります。実際のレスポンスを確認し、ダウンロード前に正規化してください。 - 出力 URL は期限切れになる可能性がある。 URL は短命なものとして扱いましょう：すぐにダウンロードするか、すぐに自分のストレージにコピーします。URL を画像の永久記録として保存しないでください。
- バイトを検証する。
Content-Typeがimage/で始まり、ファイルサイズが妥当であることを確認します。.pngとして保存されたエラーページは静かな失敗です。 - 一時的なダウンロードエラーをリトライする。 GET を小さなバックオフ付きのリトライで包みます——署名付き URL はクロックスキューで 403 を返すことがあり、ネットワークも不安定になります。
素早い手動チェックなら curl で十分です：
curl -L -o edit.png "$OUTPUT_URL"
-L フラグはリダイレクトを追跡します。CDN ホストの出力はリダイレクトをよく使います。ファイルがディスクに落ちたら、開いて編集がプロンプトに確かに一致することを確認します——これがエンドツーエンドの検証であり、失敗時に画像を再生成しなければならないパイプラインでは、視覚的またはハッシュによるチェックを自動化する価値があります。
よくある失敗ポイントと修正
初回統合の失敗のほとんどは、いくつかのパターンのいずれかで発生します。
認証とヘッダー
- 401/403 —— キーが誤っている、期限切れ、または余分な空白がある。環境変数を再エクスポートし、ヘッダーの形式がドキュメントと一致することを確認します。
- Content-Type の欠落 —— このヘッダーなしで JSON を送ると、紛らわしい 4xx エラーが発生することがあります。含めるか、Python で
json=を使います。
入力の問題
- 画像にアクセスできない —— API はサーバー側で画像を取得します。
localhostURL や認証が必要なリンクは失敗します。公開 HTTPS URL を使うか、対応していれば base64 をインラインで渡します。 - プロンプトが無視されるまたは拒否される —— 曖昧な指示は曖昧な結果を生みます。夕日の例のように、何を変え何を保つかを記述します。非常に長いプロンプトは切り詰められる可能性があります。
- サポートされないパラメータ —— モデルが受け付けないフィールドを渡すと検証エラーが発生することがあります。最小のペイロードから始め、ドキュメント化された制御項目を一つずつ追加します。
非同期処理のエラー
- 送信レスポンスから画像を読み取る —— 送信呼び出しはタスクハンドルを返し、ピクセルは返しません。まずポーリングします。
- あらゆる 200 を完了とみなす —— ステータスエンドポイントはまだ実行中のタスクにも 200 を返します。ステータスフィールドと出力 URL の存在を確認します。
- ポーリングが速すぎる、または永遠に続く —— 間隔と上限を追加しなければ、レート制限を使い果たし worker をハングさせます。
ダウンロードの問題
- HTML を画像として保存する —— 常にコンテンツタイプとサイズを確認します。
- 出力 URL の期限切れ —— 完了後すぐにダウンロードし、URL を後回しにキューに入れないでください。
構造的なエラー
- タスク ID の不一致 —— 複数のタスクを送信する場合は、各 ID に自分のリクエスト ID をタグ付けし、ポーリング結果が混線しないようにします。
- タイムアウトなし —— すべてのリクエストにタイムアウトを設定します。ないと、ハングした呼び出しがバッチ全体を引きずる可能性があります。
実習を再利用可能なヘルパー関数にする
三つのステップ——送信、ポーリング、ダウンロード——は一つの関数に属し、呼び出し側がオーケストレーションを繰り返さずに済むようにします。
def edit_image(image_url, prompt, model="flux-1-kontext-max",
headers=None, poll_interval=3, max_attempts=100):
payload = {"model": model,
"input": {"image": image_url, "prompt": prompt}}
r = requests.post(f"{BASE_URL}/jobs", headers=headers,
json=payload, timeout=30)
r.raise_for_status()
job_id = r.json()["id"]
data = wait_for_job(job_id, headers,
interval=poll_interval,
max_attempts=max_attempts)
return download_output(data, path=f"{job_id}.png")
本番対応にする要素
- 単一のエントリポイント。 呼び出し側は画像とプロンプトを渡し、ファイルパスまたは例外を受け取ります。ワークフローの詳細は隠されたままです。
- 明示的な失敗。 各リクエストは HTTP エラー時に例外を投げ、ポーラーはタスク失敗またはタイムアウト時に例外を投げるので、キューや呼び出し側がリトライできます。
- 設定可能なタイミング。
poll_intervalとmax_attemptsがパラメータなので、ロジックに触れずにより重い編集向けに緩めることができます。 - 予測可能な出力命名。 タスク ID をファイル名に使うことで、ファイルとログの関連付けが容易になります。
次に進むべき方向
アプリケーション層の関心事で呼び出しを包みます：タスク ID とレイテンシを記録し、失敗した送信をジッター付きでリトライし、ダウンロードしたバイトをローカルディスクではなくオブジェクトストレージに直接プッシュします。大量の編集を処理する場合は、まずタスクを一括送信し、まとめてポーリングします——送信は安価で、並行性はアップロード中ではなく待機中に存在します。
HiAPI ドキュメントの flux-1-kontext-max ページは、受け付けるフィールドと現在のパラメータのリファレンスです。ヘルパーを簡潔に保ち、スキーマをドキュメントに駆動させれば、同じ関数が探索的スクリプトと実際のパイプラインの両方をカバーできます。
関連記事
すべての記事を見る →-
チュートリアル
HiAPI API で hailuo-2.3-fast/image-to-video を呼び出す方法：curl と Python の実践
高速・低コストな画像から動画へのティアにおける最初の実行可能なリクエスト — モデル ID から MP4 のダウンロードまで
-
チュートリアル
HiAPI APIでhailuo-2.3/text-to-videoを呼び出す方法：curl、Python、実行可能なリクエスト例
-
チュートリアル
HiAPI で Kandinsky 6.0 Pro のテキストto動画を使用する：curl と Python チュートリアル
リクエスト優先の実践デモ：プロンプトから動画クリップへの呼び出しを構築し、リクエストを送信し、完了状態をポーリングし、結果をダウンロードする