leonardo-image

Leonardo 画像の制作

Leonardo.Ai の公式 Production API を使用して静止画像を制作します。ネイティブの Leonardo モデル、Leonardo がホストするパートナーモデル、リアルタイムキャンバスのエンドポイント、レガシーエンドポイントは、ルートとスキーマがそれぞれ異なるため、明確に区別してください。

制作ブリーフから始める

次の項目を確認するか、提案してください。

• 納品物：被写体、設定、スタイル、構図、アスペクト比／寸法、出力数、受け入れ基準。
• 操作：テキストから画像、画像から画像、参照ガイダンス、マスク編集、アップスケール、背景削除、カスタムトレーニング。
• ソースメディア：ローカルパスまたは Leonardo の画像 ID、各画像がアップロード済みか生成済みか、各参照画像の用途。
• 権利：所有権／ライセンス、肖像の使用許可、プライバシー区分、公開してよいかどうか。
• 実行環境：直接 REST、公式 Python SDK、公式 TypeScript SDK。
• 費用：計算ツールによる見積もり、上限予算、同時実行数の上限、自動チャージが無効になっているかどうか。

デフォルトでは、控えめなサイズで非公開のプレビュー画像を 1 枚作成します。料金が発生する API 呼び出しの前に、正確なルート、モデル／モデル ID、設定、サンプル数、計算ツールによる見積もり、およびその理由を伝えてください。明示的な承認を待ってください。モデル、公開範囲、参照方法、品質モード、サンプル数とバッチの適用範囲を、黙って変更してはいけません。

根拠のスナップショットと適用範囲

以下の仕様は、2026-07-10 に Leonardo の一次資料、公式 SDK リポジトリ、変更履歴、ポリシー、法的情報ページを照合して確認したものです。Leonardo は API の料金体系を PAYG に積極的に移行しており、モデルごとの v2 仕様を追加し、相互に矛盾する例を公開することもあるため、本番環境で使用する前に再確認してください。

このスキルの対象：

• POST /api/rest/v1/generations を使用する、ネイティブの Leonardo 静止画像モデル。
• 静止画像の参照、アップロード済み入力、現行の controlnets、LCM インペインティング。
• 静止画像のアップスケーラー、カスタム画像 Elements／モデル。

対象外：動画エンドポイント、3D 生成、汎用オーケストレーションシステムとしての Blueprints、Leonardo のウェブエディター自体、MCP プロキシ、非公式 SDK。一部のパートナー画像モデルは Leonardo の API 内で利用できます。その場合はパートナーモデルとして明記し、Lucid／Phoenix の派生モデルとして扱わず、各モデル固有の v2 ガイドを使用してください。

ネイティブモデルを慎重に選択する

以下の ID は、確認日時点で最新のものです。

ネイティブモデル	modelId	プロバイダーが説明する強み	実用上の選択
Lucid Origin	7b592283-e8a7-4c5a-9ba6-d18c31f258b9	汎用性が高く、プロンプトへの応答性に優れ、幅広いスタイルに対応し、グラフィックやテキストの制作に適しています	一般的なアイデア出し、様式化された画像やデザイン画像、精密なブリーフ
Lucid Realism	05ce0082-2d80-4a2d-8653-4d1c85e2418e	写実性、自然光、映画のような雰囲気	ポートレート、製品、エディトリアル写真
Phoenix 1.0	de7d3faf-762f-48e0-b3b7-9d0ac3a3fcf3	プロンプトへの忠実性と、レイアウトの制御性	複雑な構図、様式化されたアセット、複数の参照画像によるガイダンス
Phoenix 0.9	6b645e3a-d64f-4341-a6d8-7a3690fbf042	以前の Phoenix の挙動	確立済みの 0.9 パイプラインを再現する場合のみ

強みの説明は Leonardo による主張と編集上の指針であり、普遍的なベンチマーク結果ではありません。モデル選択が重要な場合は、代表的な画像を 1 枚ずつ生成して比較してください。GET /api/rest/v1/platformModels を呼び出すか、Get API Code を使用して、利用可能なモデルと ID を確認してください。ドキュメントに記載があっても、すべてのアカウントで利用できるとは限りません。

ネイティブ v1 の作成契約

POST https://cloud.leonardo.ai/api/rest/v1/generations
Authorization: Bearer $LEONARDO_API_KEY
Accept: application/json
Content-Type: application/json

本文では大文字と小文字が区別されます。汎用リファレンスで必須とされるフィールドは prompt のみですが、本番リクエストではモデルと設定を指定してください。保守的なネイティブリクエストは次のとおりです。

{
  "modelId": "7b592283-e8a7-4c5a-9ba6-d18c31f258b9",
  "prompt": "Editorial product photograph of a cobalt travel mug on pale limestone, three-quarter view, soft north-window light, clear negative space on the right.",
  "negative_prompt": "warped logo, duplicate handle, illegible text, watermark",
  "width": 1024,
  "height": 1024,
  "num_images": 1,
  "contrast": 3.5,
  "alchemy": false,
  "ultra": false,
  "public": false,
  "seed": 18427
}

現在のドキュメントに記載されている制約と注意事項：

• prompt：必須です。汎用 v1 生成では、1,500 文字を超えると拒否されます。
• negative_prompt：汎用 v1 の任意フィールドです。具体的な視覚的除外項目を指定してください。モデルによって遵守度は異なります。
• num_images：一般的な範囲として 1～8、デフォルトは 4 と記載されていますが、サイズや機能によって許容される最大値が小さくなる場合があります。まず 1 から始めてください。
• contrast：モデルガイドでは、低・中・高にそれぞれ 3、3.5、4 を使用します。汎用リファレンスでは、より広い 1～4.5 の範囲が示されています。Get API Code で別の値が使えると確認できない限り、モデルガイドにある 3 つの値を使用してください。
• width、height：モデルに対応した値である必要があります。一般的な値のページでは 512～1536、かつ 8 の倍数とされています。Lucid の公式サンプルでは 1920×1080 が使われています。エラーガイドでは、Alchemy の入力値は 32～1024、かつ 8 の倍数とされています。これらの一次情報の記述は矛盾しています。現在のモデルガイド／Get API Code と計算ツールを使い、1 枚のプレビューで正確な組み合わせを検証してください。すべてに適用される単一の範囲を設定しないでください。
• alchemy：Phoenix および一部の旧式 SD モデルでサポートされています。Lucid のガイドではサポート対象外とされています。Lucid では省略するか、エクスポートしたコードと一致させる場合に限り false を設定してください。
• ultra：モデル固有の任意の品質／アップスケールモードです。コストが変わり、サイズに影響する場合があります。
• styleUUID：現在のスタイル UUID を指定する任意フィールドです。スタイル ID は頻繁に変更されるため、モデルガイドまたはエクスポートしたコードから取得してください。
• enhancePrompt：API リファレンスではブール値です。一部のモデルガイドの表では文字列とされていますが、API リファレンスの記載に従ってください。enhancePromptInstruction は拡張内容を調整します。元の意図と拡張後の意図を両方記録してください。
• public：プライベートな API の本番利用では、明示的に false を設定してください。ドキュメントに記載のないデフォルト値に頼らないでください。
• seed：任意です。公開されている v1 作成ドキュメントでは、すべてのモデルに共通する一貫した最大値は示されていません。モデルガイドの Flux の上限値には 1 ほどの差があります。大きすぎない非負の整数を使用し、seed は再現性を補助するものであって、バイト単位で同一になる保証ではないと考えてください。
• transparency、scheduler、推論ステップ数、photoReal、Elements、旧式のフィールドはモデル固有です。選択した最新のモデルガイドまたは Get API Code に記載されたものだけをコピーしてください。

作成リクエストは generationId（通常は sdGenerationJob.generationId の中）を返します。生成済みの画像は返しません。

v1 と v2 のモデル形式を混在させないでください

Leonardo の v2 ルートはモデルディスパッチャーです。

{
  "model": "model-specific-slug",
  "public": false,
  "parameters": {
    "prompt": "model-specific prompt",
    "quantity": 1,
    "width": 1024,
    "height": 1024
  }
}

現在のガイドでその形式が定義されているモデルに限り、POST /api/rest/v2/generations に送信してください。現在の画像ガイドには、FLUX、GPT Image、Ideogram、Nano Banana/Gemini、Seedream などのパートナーファミリーが含まれています。これらは、許可されるサイズ、参照画像の数、品質名、プロンプト拡張、seed がそれぞれ異なります。

2026-07-10 時点で、GPT Image-1.5 と Ideogram 3.0 は mode を受け付けなくなりました。ドキュメントに記載された代替フィールドは quality です（GPT Image-1.5 では LOW|MEDIUM|HIGH、Ideogram 3.0 では TURBO|BALANCED|QUALITY）。古い GPT ガイドには、まだ mode が記載されています。キャッシュされたスニペットではなく、非推奨化のページと最新の /me/docs/ ガイドに従ってください。

そのモデル固有のガイドに記載がない限り、v1 の modelId、num_images、styleUUID、controlnets を v2 のパートナーリクエストに含めないでください。

Leonardo に画像を安全に取り込む

参照画像のアップロードには、次の2段階のフローを使用します。

1. { "extension": "png" } を指定して POST /api/rest/v1/init-image を実行します。ドキュメントに記載されている使用可能な拡張子は png、jpg、jpeg、webp です。
2. uploadInitImage.fields を解析し、そのフィールドと生のバイナリファイルをマルチパート形式で、返された署名付き url に2分以内に POST します。

署名付きアップロードでは通常、200 または 204 が返されます。init image ID はステップ1で発行されます。ストレージからの応答で別の ID が発行されることはありません。Leonardo のベアラートークンを署名付きストレージホストに送信しないでください。リダイレクトを無効にし、HTTPS とホストを承認済みサフィックスリストに照らして検証し、ファイルのマジックバイトと MIME タイプが宣言された拡張子と一致することを確認してください。

アップロードした入力画像が不要になったら、DELETE /api/rest/v1/init-image/{id} を使用します。生成画像 ID とアップロードした init image ID は互換性がありません。

• init_image_id：Upload Init Image が返す ID。
• init_generation_image_id：完了した Leonardo 生成処理内で返される画像 ID。
• controlnets[].initImageType：UPLOADED または GENERATED。適切な画像 ID と組み合わせて使用します。

参照画像と画像ガイダンス

非推奨の controlNet、controlNetType、weighting フィールドではなく、現行の controlnets オブジェクトを優先してください。コントロールオブジェクトには、次の内容を含められます。

{
  "initImageId": "IMAGE_UUID",
  "initImageType": "UPLOADED",
  "preprocessorId": 430,
  "strengthType": "Mid"
}

現行のネイティブプリプロセッサー ID：

ネイティブモデル	スタイル参照	キャラクター参照	コンテンツ参照
Lucid Origin / Lucid Realism のドキュメント	431	ドキュメントに記載なし	430
Phoenix 1.0 / 0.9	166	397	364

Lucid ガイドでは Origin に同じ ID が記載されていますが、一般的なガイダンステーブルでは Lucid Realism に割り当てられています。選択したモデルのガイドで確認してください。Character Reference は顔の入れ替えではなく、完全な複製を保証するものでもありません。コンテンツ参照とキャラクター参照で、ドキュメントに記載されている強度は Low、Mid、High です。スタイル参照では Ultra と Max まで指定できます。複数のスタイル参照では、相対比率として influence を使用できます。値の合計を1にする必要はありません。

従来の画像から画像への変換では、init_strength とともに init_image_id または init_generation_image_id のいずれか一方だけを使用してください。強度を高くすると、一般に元画像からの差異が大きくなります。変形を抑えるため、元画像とリクエストする出力画像のアスペクト比は互換性のあるものにしてください。

調整中は、一度に1つの参照ロールを使用してください。

1. コンテンツと構図を固定します。
2. スタイルを固定します。
3. 同意を得たうえで、明るく鮮明な元画像を使用する場合に限り、キャラクターガイダンスを追加します。
4. 出力が硬くなったり、元画像の無関係な細部までコピーしたりする場合は、最も強いコントロールを弱めます。

マスク編集とアップスケーリング

Realtime-canvas のインペインティング

POST /api/rest/v1/lcm-inpainting は、現行のマスク編集仕様が最も明確なエンドポイントです。imageDataUrl と maskDataUrl に JPEG データ URI を指定し、さらに prompt が必要です。マスクは画像と同じ寸法にする必要があります。白い部分は再生成され、黒い部分は保持されます。ドキュメントに記載されているパラメーターは次のとおりです。

• width、height：それぞれ 512、640、1024。
• guidance：0.5～20。
• strength：0.1～1。値が高いほど差異が大きくなります。
• steps：4～16。
• オプションのスタイルとシード。

これは realtime-canvas の LCM 編集であり、Lucid/Phoenix のインペインティングではありません。汎用の v1 create スキーマでは、POST /api/rest/v1/canvas-init-image の後に canvasRequest、canvasRequestType（INPAINT|OUTPAINT|SKETCH2IMG|IMG2IMG）、canvasInitId、canvasMaskId も公開されますが、現行の公開ガイドにはモデル互換性マトリックスが完全には記載されていません。本番環境で使用する前に、エクスポートしたコードまたはサポートへの確認を利用してください。

アップスケールの選択

現在の Pro アップスケーラーでは、画像参照を 1 つ指定して POST /api/rest/v2/generations を呼び出します。

• model: "aurora-upscaler-precise"：upscale_factor は 2|3|4|5|6|8、upscale_mode は clean または detailed です。忠実度の維持やノイズ除去を重視する場合に選択します。
• model: "aurora-upscaler-creative"：同じ倍率を指定でき、creativity は low|mid|high です。再構成を許容できる場合に選択します。
• 参照タイプには GENERATED、UPLOADED、VARIATION を指定できます。入力画像の実際のピクセル幅と高さを送信し、public:false を設定します。

Creative ガイドのサンプルとタイトルでは aurora-upscaler-creative を使用していますが、パラメーター表には誤って aurora-upscaler-precise と記載されています。サンプルとタイトルが意図したモデルとして扱い、Get API Code で確認してください。

旧 Universal Upscaler は POST /api/rest/v1/variations/universal-upscaler です。ドキュメントに記載されたフィールドには、generatedImageId、initImageId、variationId のいずれか 1 つ、upscaleMultiplier（旧ガイドでは 1–2）、creativityStrength、任意のプロンプト、スタイルが含まれます。GET と誤って記載しているレシピではなく、API リファレンスに記載された HTTP メソッドを正としてください。ガイドでは、Universal の出力は 20 MP が上限です。旧バリエーションエンドポイントではアップロード画像が拒否される場合がありますが、Pro v2 アップスケーラーはアップロード画像を明示的にサポートしています。

カスタム Elements とモデル

再利用可能なアイデンティティ、スタイル、または製品によるメリットが、コストとデータリスクを正当化できる場合にのみトレーニングしてください。

Element（LoRA）のフロー：

1. POST /api/rest/v1/datasets。
2. 各画像について、POST /api/rest/v1/datasets/{datasetId}/upload を呼び出し、続いて 2 分間有効な署名付きマルチパートアップロードを完了します。
3. データセット、トリガー（instance_prompt）、SDXL の設定、トレーニングパラメーターを指定して POST /api/rest/v1/elements を呼び出します。
4. GET /api/rest/v1/elements/{id} をポーリングします。トレーニングには数分から数時間かかる場合があります。
5. 互換性のあるベースモデルで、userElements: [{"userLoraId": ID, "weight": 1}] を指定して生成し、トリガーフレーズを含めます。

現在の Element ガイドでは、1024×1024 の SDXL トレーニングについて説明しています。別のカスタムモデルエンドポイントは POST /api/rest/v1/models で、name、datasetId、instance_prompt が必須です。ドキュメントに記載されたオプションには、modelType、sd_version（v1_5|v2）、strength、resolution があります。Elements とフルカスタムモデルを混同しないでください。

権利がクリアされ、代表性があり、鮮明で、一貫したキャプションが付いた画像を厳選してください。ほぼ同一の画像、透かし、センシティブな属性、同意のない人物の顔は避けてください。検証用の画像は別に確保してください。データセットファイルのハッシュ、ライセンス／同意、キャプション、トレーニングリクエスト、トリガー、返却された ID、ベースモデルとの互換性、コスト、削除の判断を記録してください。トレーニングの同時実行数は、デフォルトで 5 ジョブです。

プロンプトと反復的な制作

視覚的な事実をもとにポジティブプロンプトを組み立てます。

[medium] of [subject and defining attributes] [action/pose] in [environment].
[framing, viewpoint, spatial arrangement]. [lighting]. [palette/materials].
[lens/focus/finish/mood]. Exact visible text: "...".

スタイルキーワードより先に、被写体と状況を具体的に指定してください。カメラや照明の説明は、結果に影響する場合にのみ加えてください。正確な文字表現が必要な場合は、テキストを引用符で囲み、配置と優先順位を説明したうえで、すべての文字を確認してください。どのモデルを使っても、商標やスペルの確認は不要になりません。

negative_prompt には、矛盾だらけの長文ではなく、目で確認できる少数の欠陥を指定してください。プロンプト拡張を有効にする場合は、創造的な変換として扱い、元のプロンプト、拡張指示、エクスポートしたリクエスト、最終的に確定した設定を保持してください。

次の記録を残しながら反復します。

1. 安全なサイズで、非公開の画像 1 枚のドラフトを作成します。
2. プロンプト、モデル、スタイル、参照強度、シードのいずれか 1 つを変更します。
3. 承認済みの画像 ID を次の参照／編集ステップに引き継ぎます。
4. 構図、テキスト、アイデンティティ、権利の確認に合格した後にのみ、アップスケールします。
5. モデル ID／スラッグ、完全なリクエスト、シード、参照 ID とハッシュ、生成 ID、コスト、出力 ID／ハッシュ、判断を記録します。

フル解像度で、プロンプトへの忠実度、構図、手や解剖学的な部分とオブジェクトの形状、キャラクター／製品のアイデンティティ、テキスト／ロゴの正確さ、照明／パースの一貫性、参照画像の混入、継ぎ目／ハロー、NSFW フラグ、プライバシー／肖像／知的財産権、最終的なピクセル、色、アルファ、ファイル形式を確認してください。

非同期ライフサイクル：まず Webhook、照合にはポーリング

GET /api/rest/v1/generations/{generationId} は generations_by_pk を返します。文書化されているステータスは PENDING、COMPLETE、FAILED です。ステータスが保留中の場合、generated_images: [] を含む 200 レスポンスは成功ではありません。

Leonardo は、頻繁なポーリングより Webhook の利用を推奨しています。Production API キーの作成時に、HTTPS コールバック URL とコールバック API キーを設定してください。Leonardo は次の情報を送信します。

• authorization: Bearer <callback-secret>（Production API キーとは別のシークレット）。
• image_generation.complete などのイベントタイプ。
• ステータスや画像 URL を含む生成データ。

Bearer の比較には定時間比較を使用し、TLS、バイト数の上限、厳格な JSON／コンテンツタイプ解析、イベントタイプと生成 ID などを組み合わせた冪等性キーを使用してください。処理をキューに登録してすぐに応答し、その後、独自の API 認証情報を使って ID で生成データを取得します。ログに記録する前に機密情報をマスキングしてください。公開されているサンプルペイロードには、apiKey オブジェクトとコールバックの詳細が含まれています。Leonardo はコールバック元の IP アドレスを公開していますが、これらは変動するため、シークレット検証の補助として使用し、置き換えには使用しないでください。

公開ドキュメントには、HMAC 署名、配信の再試行スケジュール、順序保証、リプレイウィンドウについての記載がありません。そのため、停止したジョブを照合するために定期的な GET による確認を続け、Webhook の処理を冪等にしてください。コールバック設定をローテーションするには、新しい URL／シークレットで新しい Production API キーを作成し、トラフィックを切り替えてから、古いキーを削除してください。

まずドライランを行う Python の完全な例

この例では、プライベートな Lucid Origin のプレビュー、任意でアップロードしたコンテンツ参照、タスク台帳、保守的なポーリング、NSFW の拒否、安全なアーティファクトの保存を扱います。デフォルトではドライランを出力します。課金対象となるアップロード呼び出しを行うのは、--execute と見積もり計算、およびその見積もりを許容する予算を指定した場合のみです。

#!/usr/bin/env python3
import argparse, hashlib, ipaddress, json, math, os, random, socket, stat, subprocess, tempfile, time
from pathlib import Path
from urllib.parse import urlparse
import requests
from PIL import Image, UnidentifiedImageError

BASE = "https://cloud.leonardo.ai/api/rest"
ORIGIN = "7b592283-e8a7-4c5a-9ba6-d18c31f258b9"
MAX_DOWNLOAD = 30 * 1024 * 1024
MAX_UPLOAD = 30 * 1024 * 1024
MAX_API_JSON = 4 * 1024 * 1024
MAX_IMAGE_PIXELS = 24_000_000
Image.MAX_IMAGE_PIXELS = MAX_IMAGE_PIXELS

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

def allowed_https(url, suffixes):
    p = urlparse(url); host = (p.hostname or "").lower()
    if (p.scheme != "https" or p.username or p.password or p.fragment or not host
            or p.port not in (None, 443)):
        raise ValueError("unsafe URL")
    normalized = tuple(s.lower().lstrip(".").rstrip(".") for s in suffixes)
    if not any(host == s or host.endswith("." + s) for s in normalized):
        raise ValueError(f"host not allowlisted: {host}")
    for info in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM):
        if not ipaddress.ip_address(info[4][0]).is_global:
            raise ValueError("URL resolves to a non-public address")
    return url

def api(session, key, method, path, body=None, timeout=(10, 60)):
    with session.request(method, BASE + path, json=body, timeout=timeout, stream=True,
            allow_redirects=False, headers={"Authorization": f"Bearer {key}",
            "Accept": "application/json", "Content-Type": "application/json"}) as response:
        response.raise_for_status()
        if response.status_code == 204 or response.headers.get("Content-Length") == "0":
            return {}
        ctype = response.headers.get("Content-Type", "").split(";", 1)[0].lower()
        if ctype != "application/json": raise ValueError(f"unexpected API media type: {ctype}")
        raw = response.raw.read(MAX_API_JSON + 1, decode_content=True)
        if len(raw) > MAX_API_JSON: raise ValueError("API JSON exceeded byte cap")
    value = json.loads(raw.decode("utf-8"))
    if not isinstance(value, dict): raise ValueError("API response was not a JSON object")
    return value

def inspect_image_handle(handle, extension):
    opened = os.fstat(handle.fileno())
    if not stat.S_ISREG(opened.st_mode): raise ValueError("reference must be a regular file")
    size = opened.st_size
    if size <= 0 or size > MAX_UPLOAD: raise ValueError("reference file size is outside the local cap")
    handle.seek(0)
    head = handle.read(12)
    if head.startswith(b"\xff\xd8\xff"):
        expected_format, ext, mime = "JPEG", {"jpg", "jpeg"}, "image/jpeg"
    elif head.startswith(b"\x89PNG\r\n\x1a\n"):
        expected_format, ext, mime = "PNG", {"png"}, "image/png"
    elif head[:4] == b"RIFF" and head[8:12] == b"WEBP":
        expected_format, ext, mime = "WEBP", {"webp"}, "image/webp"
    else:
        raise ValueError("reference magic is not JPEG, PNG, or WebP")
    if extension.lower().lstrip(".") not in ext: raise ValueError("reference extension/magic mismatch")
    try:
        handle.seek(0)
        with Image.open(handle) as probe:
            dimensions, decoded_format = probe.size, probe.format
            if dimensions[0] * dimensions[1] > MAX_IMAGE_PIXELS:
                raise ValueError(f"reference exceeds pixel cap: {dimensions}")
            probe.verify()
        handle.seek(0)
        with Image.open(handle) as probe: probe.load()
    except (UnidentifiedImageError, OSError, Image.DecompressionBombError) as exc:
        raise ValueError("reference failed bounded full decode") from exc
    if decoded_format != expected_format: raise ValueError("reference decoder/magic mismatch")
    digest = hashlib.sha256()
    handle.seek(0)
    for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk)
    handle.seek(0)
    return {"extension": extension.lower().lstrip("."), "mime": mime, "bytes": size,
            "dimensions": list(dimensions), "sha256": digest.hexdigest()}

def inspect_local_image(path):
    if path.is_symlink(): raise ValueError("reference symlinks are not accepted")
    path = path.resolve(strict=True)
    with path.open("rb") as source:
        evidence = inspect_image_handle(source, path.suffix)
    return path, evidence

def upload_init(session, key, path, upload_suffixes, expected_evidence):
    if path.is_symlink(): raise ValueError("reference symlinks are not accepted")
    source_path = path.resolve(strict=True)
    suffix = source_path.suffix.lower()
    if suffix.lstrip(".") not in {"png", "jpg", "jpeg", "webp"}:
        raise ValueError("unsupported image extension")
    with tempfile.TemporaryFile(prefix="leonardo-approved-", suffix=suffix) as snapshot:
        with source_path.open("rb") as source:
            opened = os.fstat(source.fileno())
            if not stat.S_ISREG(opened.st_mode): raise ValueError("reference must be a regular file")
            if opened.st_size <= 0 or opened.st_size > MAX_UPLOAD:
                raise ValueError("reference file size is outside the local cap")
            copied = 0
            while chunk := source.read(1024 * 1024):
                copied += len(chunk)
                if copied > MAX_UPLOAD: raise ValueError("reference changed beyond upload byte cap")
                snapshot.write(chunk)
            snapshot.flush(); os.fsync(snapshot.fileno())
        evidence = inspect_image_handle(snapshot, suffix)
        if evidence != expected_evidence:
            raise ValueError("reference changed after approval; obtain a new approval hash")
        ext = suffix.lstrip(".")
        issued = api(session, key, "POST", "/v1/init-image", {"extension": ext})["uploadInitImage"]
        fields = json.loads(issued["fields"]) if isinstance(issued["fields"], str) else issued["fields"]
        snapshot.seek(0)
        # Upload the exact validated, continuously held snapshot handle; never reopen a path.
        # Never forward the Leonardo Authorization header to storage.
        with requests.post(allowed_https(issued["url"], upload_suffixes), data=fields,
                files={"file": (source_path.name, snapshot, evidence["mime"])},
                timeout=(10, 120), allow_redirects=False) as response:
            if response.status_code not in (200, 204): response.raise_for_status()
        return issued["id"], evidence

def poll(session, key, generation_id, out, timeout_s=900):
    deadline, delay = time.monotonic() + timeout_s, 5.0
    while time.monotonic() < deadline:
        try:
            raw = api(session, key, "GET", f"/v1/generations/{generation_id}")
            generation = raw["generations_by_pk"]; delay = 5.0
        except requests.HTTPError as e:
            if e.response.status_code not in (429, 500, 502, 503, 504): raise
            time.sleep(delay + random.uniform(0, delay * 0.5)); delay = min(delay * 2, 30); continue
        except (requests.ConnectionError, requests.Timeout):
            time.sleep(delay + random.uniform(0, delay * 0.5)); delay = min(delay * 2, 30); continue
        atomic_json(out / "status.json", {"id": generation_id, "status": generation.get("status")})
        status = generation.get("status")
        if status == "COMPLETE": return generation
        if status == "FAILED": raise RuntimeError("Leonardo generation failed")
        if status != "PENDING": raise RuntimeError(f"unknown status: {status}")
        time.sleep(5 + random.uniform(0, 1))
    raise TimeoutError("local wait expired; no cancellation was performed")

def download(session, url, image_id, out, output_suffixes, scanner_command):
    fd = None
    tmp = None
    digest = hashlib.sha256()
    total = 0
    try:
        with session.get(allowed_https(url, output_suffixes), stream=True, timeout=(10, 60),
                allow_redirects=False, headers={"Accept": "image/*"}) as response:  # no API Authorization
            response.raise_for_status()
            ctype = response.headers.get("Content-Type", "").split(";", 1)[0].lower()
            ext = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"}.get(ctype)
            if not ext: raise ValueError(f"unexpected media type: {ctype}")
            declared = response.headers.get("Content-Length")
            if declared:
                try: declared_size = int(declared)
                except ValueError as exc: raise ValueError("invalid Content-Length") from exc
                if declared_size < 0 or declared_size > MAX_DOWNLOAD:
                    raise ValueError("artifact exceeds declared byte cap")
            fd, tmp = tempfile.mkstemp(dir=out, suffix=".part")
            with os.fdopen(fd, "wb") as f:
                fd = None
                for chunk in response.iter_content(1024 * 1024):
                    if not chunk: continue
                    total += len(chunk)
                    if total > MAX_DOWNLOAD: raise ValueError("artifact exceeded streaming byte cap")
                    digest.update(chunk); f.write(chunk)
                f.flush(); os.fsync(f.fileno())
        head = Path(tmp).read_bytes()[:12]
        ok = (head.startswith(b"\xff\xd8\xff") if ext == ".jpg" else
              head.startswith(b"\x89PNG\r\n\x1a\n") if ext == ".png" else
              head[:4] == b"RIFF" and head[8:12] == b"WEBP")
        if not ok: raise ValueError("image signature mismatch")
        expected_format = {".jpg": "JPEG", ".png": "PNG", ".webp": "WEBP"}[ext]
        try:
            with Image.open(tmp) as probe:
                dimensions, decoded_format = probe.size, probe.format
                if dimensions[0] * dimensions[1] > MAX_IMAGE_PIXELS:
                    raise ValueError(f"artifact exceeds pixel cap: {dimensions}")
                probe.verify()
            with Image.open(tmp) as probe: probe.load()
        except (UnidentifiedImageError, OSError, Image.DecompressionBombError) as exc:
            raise ValueError("artifact failed bounded full decode") from exc
        if decoded_format != expected_format: raise ValueError("artifact MIME/magic/decoder mismatch")
        subprocess.run([*scanner_command, tmp], check=True, timeout=120,
                       stdin=subprocess.DEVNULL, capture_output=True)
        final = out / f"{image_id}{ext}"
        os.replace(tmp, final)
        tmp = None
        return {"id": image_id, "path": str(final), "bytes": total, "mime": ctype,
                "dimensions": list(dimensions), "sha256": digest.hexdigest(),
                "securityScan": "passed"}
    finally:
        if fd is not None: os.close(fd)
        if tmp is not None and os.path.exists(tmp): os.unlink(tmp)

parser = argparse.ArgumentParser()
parser.add_argument("--prompt", required=True)
parser.add_argument("--reference", type=Path)
parser.add_argument("--estimated-usd", type=float, required=True,
    help="copy the current estimate from Leonardo's API pricing calculator")
parser.add_argument("--execute", action="store_true")
args = parser.parse_args()
if len(args.prompt) > 1500: raise SystemExit("prompt exceeds 1500 characters")
if not math.isfinite(args.estimated_usd) or args.estimated_usd <= 0:
    raise SystemExit("estimated-usd must be finite and positive")
try: budget = float(os.environ.get("LEONARDO_MAX_USD", "0"))
except ValueError as exc: raise SystemExit("LEONARDO_MAX_USD must be numeric") from exc
if not math.isfinite(budget) or budget < 0: raise SystemExit("LEONARDO_MAX_USD must be finite and non-negative")
body = {"modelId": ORIGIN, "prompt": args.prompt, "width": 1024, "height": 1024,
        "num_images": 1, "contrast": 3.5, "public": False, "seed": 18427}
preview = dict(body)
reference_evidence = None
if args.reference:
    _, reference_evidence = inspect_local_image(args.reference)
    preview["controlnets"] = [{"initImageId": "<uploaded-at-execution>",
        "initImageType": "UPLOADED", "preprocessorId": 430, "strengthType": "Mid"}]
approval_record = {"request": preview, "reference": reference_evidence,
                   "estimatedUsd": args.estimated_usd, "maxUsd": budget}
approval_sha256 = hashlib.sha256(json.dumps(
    approval_record, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
print(json.dumps({"dryRun": not args.execute, "estimatedUsd": args.estimated_usd,
                  "maxUsd": budget, "request": preview,
                  "referenceEvidence": reference_evidence,
                  "approvalSha256": approval_sha256}, indent=2))
if not args.execute: raise SystemExit(0)
if budget <= 0 or args.estimated_usd > budget: raise SystemExit("estimate is not within approved budget")
if os.environ.get("LEONARDO_APPROVED_REQUEST_SHA256") != approval_sha256:
    raise SystemExit("approval hash does not match this exact request, source, estimate, and budget")
try: scanner_command = json.loads(os.environ["LEONARDO_IMAGE_SCAN_COMMAND_JSON"])
except (KeyError, json.JSONDecodeError) as exc:
    raise SystemExit("set LEONARDO_IMAGE_SCAN_COMMAND_JSON to an approved scanner argv array") from exc
if (not isinstance(scanner_command, list) or not scanner_command
        or not all(isinstance(value, str) and value for value in scanner_command)):
    raise SystemExit("LEONARDO_IMAGE_SCAN_COMMAND_JSON must be a non-empty string array")
key = os.environ["LEONARDO_API_KEY"]
upload_suffixes = tuple(x.strip().lower() for x in os.environ["LEONARDO_UPLOAD_HOST_SUFFIXES"].split(",") if x.strip())
output_suffixes = tuple(x.strip().lower() for x in os.environ.get(
    "LEONARDO_OUTPUT_HOST_SUFFIXES", "cdn.leonardo.ai").split(",") if x.strip())
out = Path(os.environ.get("LEONARDO_OUTPUT_DIR", "leonardo-output")).resolve(); out.mkdir(parents=True, exist_ok=True)
session = requests.Session()
client_job_key = f"leonardo-{approval_sha256[:20]}"
job_path = out / "job.json"
atomic_json(job_path, {"state": "pre_submit", "clientJobKey": client_job_key,
    "approvalSha256": approval_sha256, "requestSha256": approval_sha256,
    "modelId": ORIGIN, "estimatedUsd": args.estimated_usd,
    "reference": reference_evidence, "updatedUnix": time.time()})
if args.reference:
    init_id, uploaded_evidence = upload_init(
        session, key, args.reference.resolve(), upload_suffixes, reference_evidence
    )
    body["controlnets"] = [{"initImageId": init_id, "initImageType": "UPLOADED",
        "preprocessorId": 430, "strengthType": "Mid"}]
actual_request_sha256 = hashlib.sha256(json.dumps(
    body, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
atomic_json(job_path, {"state": "ready_to_submit", "clientJobKey": client_job_key,
    "approvalSha256": approval_sha256, "requestSha256": actual_request_sha256,
    "modelId": ORIGIN, "estimatedUsd": args.estimated_usd,
    "reference": reference_evidence, "updatedUnix": time.time()})
try:
    created = api(session, key, "POST", "/v1/generations", body)
except (requests.ConnectionError, requests.Timeout) as e:
    atomic_json(job_path, {"state": "ambiguous_create", "clientJobKey": client_job_key,
        "approvalSha256": approval_sha256, "requestSha256": actual_request_sha256,
        "modelId": ORIGIN, "estimatedUsd": args.estimated_usd,
        "reference": reference_evidence, "updatedUnix": time.time(),
        "note": "Do not replay; reconcile generations and account cost"})
    raise SystemExit(f"create outcome unknown; reconcile before retrying: {e}")
job = created.get("sdGenerationJob", created); generation_id = job.get("generationId")
if not generation_id: raise RuntimeError("create response did not contain generationId")
atomic_json(job_path, {"state": "submitted", "clientJobKey": client_job_key,
    "generationId": generation_id, "approvalSha256": approval_sha256,
    "requestSha256": actual_request_sha256, "modelId": ORIGIN,
    "estimatedUsd": args.estimated_usd,
    "createdCost": job.get("cost") or job.get("apiCreditCost"),
    "reference": reference_evidence, "updatedUnix": time.time()})
generation = poll(session, key, generation_id, out)
artifacts = []
for image in generation.get("generated_images", []):
    if image.get("nsfw"): raise RuntimeError(f"output rejected by NSFW policy: {image.get('id')}")
    artifacts.append(download(session, image["url"], image["id"], out,
                              output_suffixes, scanner_command))
if not artifacts: raise RuntimeError("COMPLETE generation returned no accepted images")
atomic_json(out / "manifest.json", {"generationId": generation_id, "status": "COMPLETE",
    "clientJobKey": client_job_key, "modelId": ORIGIN,
    "approvalSha256": approval_sha256,
    "requestSha256": actual_request_sha256,
    "promptSha256": hashlib.sha256(args.prompt.encode()).hexdigest(),
    "reference": reference_evidence,
    "estimatedUsd": args.estimated_usd,
    "actualCost": generation.get("cost") or generation.get("apiDollarCost"),
    "nsfwAccepted": False, "creativeQa": "pending", "rightsReview": "required",
    "provenance": "original provider bytes preserved; no universal C2PA claim",
    "artifacts": artifacts, "completedUnix": time.time()})
print(json.dumps(artifacts, indent=2))

Webhook ハンドラーでは、専用のコールバック Bearer シークレットに hmac.compare_digest を使用してください。アップロードのサフィックスは、ストレージプロバイダーをハードコードするのではなく、レビュー済みの署名付き URL レスポンスまたは環境ポリシーに基づいて設定してください。

失敗、再試行、重複課金のポリシー

HTTP レスポンスの失敗、生成の終端ステータス、品質上の失敗を分けて扱います。

シグナル	対応
400 検証エラー、無効な寸法、画像数の超過、プロンプトが1500文字を超過	ペイロードとモデルの組み合わせを修正してください。同じ内容のまま再試行しないでください
401/403 認証エラー	キーを修正またはローテーションしてください。同じ内容のまま再試行しないでください
403 モデレーション	回避しないでください。規約に従って内容を修正するか、拒否してください
404 ID／モデル	生成 ID、画像 ID、初期化 ID のどれかを確認し、現在のモデル提供状況を確認してください
429	アカウントの制限を守り、ジッターを加えた上限付き指数バックオフを使用してください
GET の 5xx／ネットワークエラー	ジッターを加えた上限付きバックオフを使用してください。GET は安全に再実行できます
PENDING ＋画像なし	待機を続けるか、Webhook と照合してください。成功ではありません
FAILED	ID と診断レスポンスを保持し、内容を変更して再試行する前に分類してください
COMPLETE ＋ nsfw:true	隔離または拒否してください。必要に応じて別のモデレーション層を追加してください

公式 SDK では再試行を設定できますが、公開ドキュメントには生成作成の冪等性キーについて記載がありません。タイムアウト後に POST /generations を無条件で再試行しないでください。ジョブがすでに受け付けられ、課金されている可能性があります。送信前にローカルのジョブキーを永続化し、generationId を直ちに保存して、読み取りのみを再試行してください。生成作成の結果が不明な場合は、自動送信を停止し、最近のアカウントの生成履歴と費用を照合するか、サポートに問い合わせてください。

署名付きアップロード URL の有効期間は2分です。有効期限が切れた URL を再試行するのではなく、新しい URL を要求してください。画像 ID が必要な場面で生成 ID を再利用しないでください。

コストと容量のガバナンス

Leonardo の現在の Production API は、ドル残高を使用する PAYG へ移行中です。コストはモデル、寸法、枚数、品質／機能、ガイダンス、アップスケール、またはトレーニングによって異なります。Leonardo は公開ドキュメントで、安定した完全なモデル別ドル価格表を公開していません。価格を捏造しないでください。

実行前に：

1. API Access の API 料金計算ツールで、正確な構成を指定して計算します。
2. 見積もりを記録し、ローカルのハードリミットを適用します。
3. プロンプト／参照画像の分岐が通過するまでは、num_images:1 を維持します。
4. 無人サービスでは、自動チャージを無効にするか、上限を厳しく設定します。
5. 返された cost オブジェクトを記録します。移行期間中は apiCreditCost も引き続き使用されます。

PAYG 残高に有効期限はありません。ウェブアプリのサブスクリプショントークンと API 請求は別々です。カスタムプランでは上限や割引が変更されることがあります。正式な情報はポータル／注文書で確認してください。

2026-07-10 時点で公開されているデフォルトの上限：

• 画像生成ジョブの同時実行数：10。
• トレーニングジョブの同時実行数：5。
• キュー内／保留中の画像生成：200 件、キュー内のアップスケール：100 件。
• 全体で毎分 2,000 リクエスト。
• 生成作成リクエストは毎分 100 件、ドキュメントに記載された各バリエーション／アップスケール作成ルートは毎分 100 件。

同時実行枠 1 つにつき、生成リクエストが 10 件、保留中の画像ジョブが 20 件追加されると説明されています。これらの上限はカスタマイズされる場合があります。アプリケーション側ではより低い上限を適用し、429 応答を尊重し、プロバイダーがより多くの処理を受け付ける場合でもユーザーごとの予算を保護してください。

成果物、キー、Webhook のセキュリティ

• 名前付きの Production API キーをサーバー側のシークレットマネージャーで使用し、ブラウザー／モバイルコードに埋め込まないでください。現在の FAQ によると、1 アカウントにつき 10 個のキーを保持できます。キーをローテーションし、未使用のキーは削除してください。
• すべての API 応答と Webhook 本文を機密情報として扱ってください。アカウント／キーのメタデータを含む可能性がある生のオブジェクトではなく、許可リストに基づく情報のみを保存してください。
• public:false を維持してください。生成物を非公開にしても、秘密情報、規制対象データ、または同意を得ていない生体情報／肖像素材をアップロードしてよいことにはなりません。
• API で生成された画像 URL は有効期限がなく、ウェブアプリからアクセスできると記載されています。これはストレージの SLA を意味するものではありません。速やかにダウンロードし、HTTPS／ホスト／DNS を検証してください。API 認証情報を転送せず、バイト数に上限を設け、MIME とマジックバイトの両方を検証し、アトミックに保存してハッシュを計算し、管理された永続ストレージに保存してください。
• タスクで指定された任意の URL をプロキシしないでください。署名付きアップロード用ホストと Leonardo の出力 CDN ホストにはそれぞれ別の許可リストを設け、リダイレクトとプライベート／予約済みアドレスを拒否してください。
• 必要に応じて、DELETE /api/rest/v1/generations/{id} で生成物を、DELETE /api/rest/v1/init-image/{id} でアップロードした入力を削除してください。アカウントで削除の挙動を確認し、証跡を保持してください。公開ドキュメントにはバックアップ削除の SLA が記載されていません。

権利、プライバシー、安全性、および来歴

これは運用上のガイダンスであり、法的助言ではありません。

• 入力の所有権はお客様にあります。すべての権利、ライセンス、プライバシーに関する許可、および肖像の使用許諾を得ている必要があります。顔の入れ替えやなりすましの近道として、Character Referenceやトレーニングを使用しないでください。
• 現行の利用規約（2026-01-19）では、当事者間において、有料サブスクライバーは規約に準拠したコンテンツを所有し、Leonardoは自らが有する権利を譲渡すると定めています。出力は一意とは限らず、Leonardoは権原および非侵害に関する保証を否認しています。APIのページには別途、API利用者は入力を保持し、出力の権利を得ると記載されています。適用されるのは、準拠法を定めるサブスクリプション契約書または注文書です。
• 常にpublic:falseを設定してください。利用規約では、プライベートコンテンツは義務の履行および権利の行使にのみ使用され、明示的な書面による同意がない限り、モデルのトレーニングや新製品のために使用、保持、分析、または処理されないと定めています。公開コンテンツには、トレーニングや商用利用を含む広範かつ永続的なライセンスが付与されます。APIのページには、API入力および生成画像はトレーニングに使用されないと記載されており、プライバシー関連資料では、この保護が有料のプライベートコンテンツにも適用されるとされています。これを公開コンテンツにまで一般化しないでください。
• APIで生成された画像は期限切れにならないと記載されています。一方、プライバシーポリシーには目的に基づく保持が定められており、利用規約では12か月以上非アクティブなアカウントの削除が認められています。これらの記載は、それぞれ異なる階層に関するものです。エンドポイントごとの削除やバックアップに関する公開SLAはありません。必要な場合は契約で定めてください。
• プライバシーポリシーでは、情報は米国に保管され、オーストラリアおよびLeonardo、関連会社、プロバイダー、またはパートナーが事業を展開する国で処理されると記載されています。公開されているAPIの地域パラメーターや、顧客が選択できるデータ保存地域は見つかりませんでした。規制対象のワークロードには、DPAまたは注文書を使用してください。
• 利用規約では、違法行為や権利侵害を伴う利用、誤解を招くまたは名誉を毀損するなりすまし、性的虐待や暴力、露骨なポルノ、児童搾取、同意のない描写、動物虐待、極端な流血表現、憎悪や差別、危険にさらす行為を禁止しています。プロンプトのモデレーションにより、400/403で拒否される場合があります。出力にはnsfwフラグが含まれます。アプリケーション固有の入出力モデレーションと人によるレビューを追加してください。
• Leonardoは、SOC 2 Type IおよびIIの認証を取得していると述べています。これはプロバイダーの主張として扱い、保証業務では報告書と統制マッピングを入手してください。
• 2026-07-10時点で、すべての出力にC2PA Content Credentials、不可視のウォーターマーク、またはその他の来歴マーカーが付与されることを保証する、ファーストパーティーのAPIドキュメントは見つかりませんでした。そのようなマーカーがあると主張しないでください。元のバイト列とメタデータを保持し、成果物をハッシュ化し、生成マニフェストを保存し、ポリシーや法律に従って合成コンテンツであることを開示してください。

リリースチェックリスト

• 現行のガイドまたはアカウントで、正確なネイティブモデルIDまたはv2モデルスラッグを確認済み。
• そのモデルに対してリクエストを検証済み。モデル間で共有されないフィールドや非推奨のパラメーターは使用していない。
• 非公開の1枚プレビュー、計算ツールによる見積もり、および上限予算を承認済み。
• 参照ID／タイプ、前処理ID、強度、権利、同意、および削除計画を確認済み。
• ポーリング／Webhook処理の前に、ジョブキーと生成IDを永続化済み。
• Webhookのベアラー認証検証、冪等性、機密情報の除去、およびGETによる照合を実装済み。
• 作成リクエストの結果が不明な場合、自動再試行しない。
• NSFW／ポリシーチェックとフル解像度でのクリエイティブ品質確認に合格済み。
• API認証なしで成果物をダウンロードし、検証、ハッシュ化したうえで永続的に保存済み。
• 実際のコスト、出力ID、来歴マニフェスト、プライバシー／トレーニング／地域に関する前提、および削除を記録済み。

ファーストパーティの情報源一覧

2026-07-10 にアクセス：

• クイックスタートと汎用 API リファレンス：https://docs.leonardo.ai/docs/getting-started および https://docs.leonardo.ai/me/reference/creategeneration
• ネイティブモデルガイド：https://docs.leonardo.ai/docs/lucid-origin、https://docs.leonardo.ai/me/docs/lucid-realism、および https://docs.leonardo.ai/docs/phoenix
• モデルと値：https://docs.leonardo.ai/reference/listplatformmodels および https://docs.leonardo.ai/docs/commonly-used-api-values
• 画像ガイダンスとプロンプト：https://docs.leonardo.ai/docs/generate-images-using-image-to-image-guidance および https://leonardo.ai/news/ai-image-prompts
• アップロード：https://docs.leonardo.ai/reference/uploadinitimage および https://docs.leonardo.ai/docs/how-to-upload-an-image-using-a-presigned-url
• ライフサイクル／Webhook：https://docs.leonardo.ai/reference/getgenerationbyid および https://docs.leonardo.ai/docs/guide-to-the-webhook-callback-feature
• エラー／モデレーション：https://docs.leonardo.ai/docs/api-error-messages および https://docs.leonardo.ai/docs/guide-to-handling-not-safe-for-work-image-generation-nsfw
• インペインティングとアップスケーリング：https://docs.leonardo.ai/reference/performinpaintinglcm、https://docs.leonardo.ai/me/docs/pro-upscaler-precise、https://docs.leonardo.ai/me/docs/pro-upscaler-creative、および https://docs.leonardo.ai/reference/createuniversalupscalerjob
• トレーニング：https://docs.leonardo.ai/docs/train-custom-element-and-generate-images および https://docs.leonardo.ai/reference/createmodel
• 公式 SDK：https://github.com/Leonardo-Interactive/leonardo-python-sdk および https://github.com/Leonardo-Interactive/leonardo-ts-sdk
• 制限、PAYG、および料金：https://docs.leonardo.ai/reference/limits、https://docs.leonardo.ai/docs/payg-guide、および https://docs.leonardo.ai/me/docs/pricing-and-plans-faq
• 非推奨化：https://docs.leonardo.ai/me/docs/deprecations-changes
• API の製品／データに関する主張：https://leonardo.ai/api
• 利用規約、プライバシー、およびエンタープライズ契約：https://leonardo.ai/terms-of-service、https://leonardo.ai/privacy-policy、および https://leonardo.ai/master-services-agreement