leonardo-image

Leonardo 图像制作

使用 Leonardo.Ai 官方 Production API 生成静态图像。请明确区分 Leonardo 原生模型、由 Leonardo 托管的合作伙伴模型、实时画布端点和旧版端点：它们使用不同的路由和架构。

从制作需求说明开始

收集或提出以下信息：

• 交付内容：主体、场景、风格、构图、宽高比／尺寸、输出数量和验收标准。
• 操作：文生图、图生图、参考图引导、蒙版编辑、放大、移除背景或自定义训练。
• 源媒体：本地路径或 Leonardo 图像 ID、每项是上传还是生成的，以及每个参考素材的预期用途。
• 权利：所有权／许可、肖像同意、隐私分类，以及是否允许公开显示。
• 运行方式：直接 REST、官方 Python SDK 或官方 TypeScript SDK。
• 费用：计算器估算、硬性预算、并发上限，以及是否已禁用自动充值。

默认生成一张低尺寸的私有预览图。进行任何付费调用前，说明确切路由、模型／模型 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 内部提供了部分合作伙伴图像模型；请将其标记为合作伙伴模型，并使用各自的 v2 指南，而不要将其视为 Lucid／Phoenix 的变体。

谨慎选择原生模型

以下 ID 在核查日期时为当前 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 的说法和编辑建议，并非普遍适用的基准测试结果。模型选择至关重要时，请进行有代表性的一图对比。调用 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 证实其他值可用，否则请使用模型指南中的这三个值。
• width、height：必须与模型兼容。常用值页面指出范围为 512–1536，且必须是 8 的倍数；Lucid 的官方示例使用 1920×1080；错误指南则指出 Alchemy 输入应为 32–1024，且必须是 8 的倍数。这些一手资料互相矛盾。请参考当前模型指南／Get API Code 和计算器，并用单张图片预览验证确切的组合。不要强行套用一个通用范围。
• alchemy：Phoenix 和部分旧版 SD 模型支持此项；Lucid 指南称其不受支持。对于 Lucid，请省略此项，或仅在与导出代码一致时将其设为 false。
• ultra：可选的特定模型质量／升频模式；此项会改变费用，也可能影响尺寸。
• styleUUID：可选的当前样式 UUID。样式 ID 会变化；请从模型指南或导出代码中获取。
• enhancePrompt：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/ 指南，不要参考缓存的代码片段。

除非特定指南对此有明确说明，否则不要在 v2 合作伙伴请求中发送 v1 的 modelId、num_images、styleUUID 或 controlnets。

安全地将图像导入 Leonardo

上传参考图采用两步流程：

1. 使用 { "extension": "png" } 调用 POST /api/rest/v1/init-image。文档中允许的扩展名有 png、jpg、jpeg 和 webp。
2. 解析 uploadInitImage.fields，然后在两分钟内，将这些字段和原始二进制文件通过 multipart POST 发送到返回的预签名 url。

预签名上传通常返回 200 或 204。初始图像 ID 由第 1 步签发；存储响应不会再签发另一个 ID。切勿将 Leonardo bearer token 发送到预签名存储主机。禁用重定向，根据获准的后缀名列表验证 HTTPS 和主机，并确保文件的魔数和 MIME 类型与声明的扩展名匹配。

当不再需要已上传的输入图像时，使用 DELETE /api/rest/v1/init-image/{id}。生成的图像 ID 与上传的初始图像 ID 不能互换：

• init_image_id：上传初始图像时返回的 ID。
• init_generation_image_id：已完成的 Leonardo 生成任务中返回的图像 ID。
• controlnets[].initImageType：UPLOADED 或 GENERATED，需与相应的图像 ID 配对。

参考图与图像指导

优先使用当前的 controlnets 对象，而不是已弃用的 controlNet、controlNetType 和 weighting 字段。控制对象可以包含：

{
  "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 指南将相同的 ID 标注为适用于 Origin，而通用指南表格则将其标注为适用于 Lucid Realism；请根据所选模型的指南进行确认。角色参考不是换脸，也不保证生成完全相同的角色。对于内容和角色参考，文档中说明的强度选项为 Low、Mid、High；风格参考还可使用 Ultra 和 Max。多个风格参考可以使用 influence 作为相对比例；数值之和不必为 1。

对于经典的图生图，请在 init_image_id 和 init_generation_image_id 中恰好使用一个，并设置 init_strength。强度越高，通常允许的偏差越大。让源图与所需输出的宽高比保持兼容，以减少变形。

调整时，每次只使用一种参考图角色：

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 创建架构还公开了 canvasRequest、canvasRequestType（INPAINT|OUTPAINT|SKETCH2IMG|IMG2IMG）、canvasInitId 和 canvasMaskId，这些参数在调用 POST /api/rest/v1/canvas-init-image 后可用，但当前公开指南并未完整说明模型兼容性矩阵。投入生产前，请使用导出的代码或向支持团队确认。

放大选择

对于当前的 Pro Upscalers，使用一张图像引用调用 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 三者之一，upscaleMultiplier（旧版指南中为 1–2）、creativityStrength、可选的提示词以及风格。对于 HTTP 方法，应以 API 参考为准，而不是误写为 GET 的操作示例。其指南规定 Universal Upscaler 的输出上限为 20 MP。旧版 variation 端点可能会拒绝上传的图像；Pro v2 Upscalers 明确支持上传的图像。

自定义 Elements 和模型

只有在可复用的身份、风格或产品能带来足够收益，且成本和数据风险可接受时，才进行训练。

Element（LoRA）流程：

1. POST /api/rest/v1/datasets。
2. 对每张图像，调用 POST /api/rest/v1/datasets/{datasetId}/upload，然后完成两分钟有效的预签名分段上传。
3. 调用 POST /api/rest/v1/elements，并提供数据集、触发词（instance_prompt）、SDXL 配置和训练参数。
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）、强度和分辨率。不要混淆 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. 以安全尺寸生成单张私有草稿。
2. 每次只更改一个变量：提示词、模型、风格、引用强度或种子。
3. 将已批准的图像 ID 用于下一步引用或编辑。
4. 只有在构图、文字、身份和权利审查通过后才放大。
5. 记录模型 ID/slug、完整请求、种子、引用 ID 和哈希、生成 ID、成本、输出 ID/哈希以及决定。

以全分辨率检查提示词遵循情况；构图；手部/解剖结构及物体几何形状；角色/产品身份；文字/logo 准确性；光照/透视一致性；引用内容泄漏；接缝/光晕；NSFW 标记；隐私/肖像权/知识产权；以及最终像素、颜色、alpha 和文件格式。

异步生命周期：优先使用 Webhook，以轮询进行核对

GET /api/rest/v1/generations/{generationId} 会返回 generations_by_pk。文档记录的状态包括 PENDING、COMPLETE 和 FAILED。状态为待处理时，即使响应为 200 且 generated_images: []，也不代表成功。

Leonardo 建议使用 Webhook，而不是频繁轮询。创建 Production API key 时，请配置 HTTPS 回调 URL 和回调 API key。Leonardo 会发送：

• authorization: Bearer <callback-secret>（与 Production API key 分开的独立密钥）。
• 事件类型，例如 image_generation.complete。
• 包含状态和图像 URL 的生成数据。

使用恒定时间的 Bearer 比较、TLS、字节数上限、严格的 JSON/内容类型解析，以及幂等键（例如事件类型加生成 ID）。将工作加入队列并快速返回，然后使用你自己的 API 凭据按 ID 获取生成内容。记录日志前先脱敏：已发布的示例负载包含 apiKey 对象和回调详情。Leonardo 公布了回调源 IP，但这些 IP 可能会变动，因此它们应作为密钥验证的补充，而非替代。

公开文档未说明 HMAC 签名、投递重试计划、顺序保证或重放窗口。因此，请定期通过 GET 核对卡住的任务，并确保 Webhook 处理具有幂等性。若要轮换回调配置，请使用新的 URL/密钥创建新的 Production API key，切换流量，然后删除旧密钥。

完整的先试运行后执行 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 处理程序，请使用 hmac.compare_digest 比较其独立的回调 Bearer 密钥。请根据经过审核的预签名响应/环境策略设置上传文件后缀，而不要将存储服务提供商硬编码。

失败、重试和重复计费策略

区分 HTTP 响应失败、终态生成状态和质量失败。

信号	响应
400 验证失败、尺寸无效、图像过多、提示词 >1500	修正负载/模型组合；不要原样重试
401/403 身份验证失败	修正/轮换密钥；不要原样重试
403 内容审核	不要绕过；根据条款修改或拒绝
404 ID/模型	核实 ID 是生成 ID、图像 ID 还是 init ID，并确认模型当前是否可用
429	遵守账户限制；使用带抖动的有界指数退避
5xx/GET 网络错误	使用带抖动的有界退避；GET 可以安全重复
PENDING + 空图像	继续等待或核对 Webhook；这不代表成功
FAILED	保留 ID 和诊断响应，然后在更改后重试前进行分类
COMPLETE + nsfw:true	隔离/拒绝；需要时添加单独的内容审核层

官方 SDK 支持配置重试，但公开文档未说明创建生成任务时可使用幂等键。超时后不要盲目重试 POST /generations，因为任务可能已被接受并计费。提交前先持久化本地任务键，并立即保存 generationId；只重试读取操作。若创建结果不明确，请停止自动提交，并核对近期账户生成记录、费用或联系支持。

预签名上传 URL 的有效期为两分钟；URL 过期时应重新申请，而不是重试已过期的 URL。需要图像 ID 时，切勿重复使用生成 ID。

成本与容量治理

Leonardo 当前的 Production API 正在过渡到按量付费（PAYG）和美元余额模式。成本取决于模型、尺寸、数量、质量／功能、引导、放大或训练。Leonardo 不会在公开文档中发布稳定且完整的逐模型美元价格表；请勿自行编造。

执行前：

1. 在 API Access 中使用准确配置打开 API 定价计算器。
2. 记录估算费用并设置本地硬性上限。
3. 在提示词／参考图分支通过验证前，将 num_images:1 保持为默认值。
4. 对无人值守服务禁用自动充值，或严格限制充值上限。
5. 记录返回的 cost 对象；迁移期间仍会保留 apiCreditCost。

PAYG 余额不会过期。Web 应用订阅代币与 API 计费是分开的。自定义套餐可以调整限额和折扣；应以门户／订单表为准。

截至 2026-07-10，公布的默认限额：

• 最多同时运行 10 个图像生成任务。
• 最多同时运行 5 个训练任务。
• 排队／待处理的图像生成任务最多 200 个；排队等待放大的任务最多 100 个。
• 总体限额为每分钟 2,000 个请求。
• 创建生成请求的限额为每分钟 100 个；每个已记录的变体／放大器创建路由也均为每分钟 100 个。

据称，一个并发单位可增加 10 个生成请求和 20 个待处理图像任务。这些限额可能会自定义。请在应用侧设置更低的限额，遵循 429 响应，并保护每位用户的预算，即使服务提供方接受更多任务也不例外。

产物、密钥与 Webhook 安全

• 在服务端密钥管理器中使用命名的 Production API 密钥；切勿将其嵌入浏览器／移动端代码。当前 FAQ 表示，一个账号最多可持有 10 个密钥。请轮换密钥并删除未使用的密钥。
• 将所有 API 响应和 Webhook 请求体视为敏感信息。仅持久化经过允许列表筛选的数据，不要保存可能包含账号／密钥元数据的原始对象。
• 保持 public:false。生成内容设为私有，并不代表可以上传机密、受监管数据，或未经同意的生物识别／肖像素材。
• 文档说明，API 生成的图像 URL 不会过期，且可在 Web 应用中访问。这并非存储 SLA。请及时下载，验证 HTTPS／主机／DNS，不要转发 API 凭据，限制字节数，核验 MIME 类型及文件魔数，以原子方式保存、计算哈希，并存储在受控的持久化存储中。
• 不要代理任务提供的任意 URL。为预签名上传主机和 Leonardo 输出 CDN 主机分别维护允许列表；拒绝重定向以及私有／保留地址。
• 必要时使用 DELETE /api/rest/v1/generations/{id} 删除生成内容，并使用 DELETE /api/rest/v1/init-image/{id} 删除已上传的输入。请在账号中验证删除行为并保留证据；公开文档未说明备份删除 SLA。

权利、隐私、安全与来源信息

这是操作指南，不构成法律建议。

• 你保留输入内容的所有权，并且必须拥有所有权利、许可、隐私授权和肖像同意。请勿将角色参考或训练作为换脸／冒充他人的捷径。
• 当前《条款》（2026-01-19）规定，在双方之间，付费订阅者拥有合规内容，Leonardo 会转让其拥有的任何相关权益。输出内容可能不具有唯一性；Leonardo 不保证所有权或不侵权。API 页面另行说明，API 客户保留输入内容，并获得输出内容的权利。以适用的订阅／订单表单为准。
• 始终设置 public:false。《条款》规定，私有内容仅用于履行义务／行使权利；未经明确书面同意，不会用于模型训练／新产品，也不会为此被保留、分析或处理。公开内容则授予广泛的永久许可，其中包括训练和商业用途。API 页面称 API 输入和生成的图像不会用于训练；隐私材料也将此保护与付费私有内容相对应。请勿将其泛化至公开内容。
• 文档说明，生成的 API 图像不会过期；而隐私政策规定了基于目的的保留期限，《条款》允许删除不活跃账户（12+ 个月）。这些声明针对不同层面。公开文档中没有针对特定 API 端点的删除／备份 SLA；如有需要，请在合同中约定。
• 隐私政策称，信息存储在美国，并在澳大利亚以及 Leonardo、其关联方、服务提供商或合作伙伴运营所在的任何国家／地区进行处理。未发现公开的 API 区域参数或客户可选的数据驻留选项。对于受监管的工作负载，请参阅 DPA／订单表单。
• 《条款》禁止非法／侵权用途、误导性或诽谤性冒充、性虐待／暴力、露骨色情、剥削儿童、未经同意的描绘、虐待动物、极端血腥内容、仇恨／歧视，以及危及他人安全的行为。提示词审核可能以 400/403 拒绝请求；输出包含 nsfw 标记。请添加适用于具体应用的输入／输出审核和人工审核。
• Leonardo 表示其已通过 SOC 2 Type I 和 II 认证。请将此视为服务提供商的声明；如需进行保障评估，请索取报告和控制措施映射。
• 截至 2026-07-10，未找到任何第一方 API 文档保证每项输出都带有 C2PA Content Credentials、不可见水印或其他来源标记。请勿声称存在此类标记。保留原始字节／元数据，对产物进行哈希，保存生成清单，并按照政策／法律披露合成内容的来源。

发布检查清单

• 已在当前指南／账户中确认准确的原生模型 ID 或 v2 模型 slug。
• 已根据该模型验证请求；未使用跨模型字段或已弃用参数。
• 已批准私有单图预览、计算器估算和硬性预算。
• 已核实参考 ID／类型、预处理 ID、强度、权利、同意和删除计划。
• 已在轮询／Webhook 处理之前持久化作业键和生成 ID。
• 已实现 Webhook bearer 验证、幂等性、脱敏和 GET 对账。
• 不会自动重试结果不明确的创建请求。
• 已通过 NSFW／政策检查和全分辨率创意审核。
• 已在无需 API 认证的情况下下载、验证、哈希并持久保存产物。
• 已记录实际成本、输出 ID、来源清单、隐私／训练／区域假设及删除情况。

第一方来源登记

访问日期：2026-07-10：

• 快速入门和通用 API 参考：https://docs.leonardo.ai/docs/getting-started and https://docs.leonardo.ai/me/reference/creategeneration
• 原生模型指南：https://docs.leonardo.ai/docs/lucid-origin, https://docs.leonardo.ai/me/docs/lucid-realism, and https://docs.leonardo.ai/docs/phoenix
• 模型和值：https://docs.leonardo.ai/reference/listplatformmodels and https://docs.leonardo.ai/docs/commonly-used-api-values
• 图像引导和提示词：https://docs.leonardo.ai/docs/generate-images-using-image-to-image-guidance and https://leonardo.ai/news/ai-image-prompts
• 上传：https://docs.leonardo.ai/reference/uploadinitimage and https://docs.leonardo.ai/docs/how-to-upload-an-image-using-a-presigned-url
• 生命周期/网络钩子：https://docs.leonardo.ai/reference/getgenerationbyid and https://docs.leonardo.ai/docs/guide-to-the-webhook-callback-feature
• 错误/内容审核：https://docs.leonardo.ai/docs/api-error-messages and 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, and https://docs.leonardo.ai/reference/createuniversalupscalerjob
• 训练：https://docs.leonardo.ai/docs/train-custom-element-and-generate-images and https://docs.leonardo.ai/reference/createmodel
• 官方 SDK：https://github.com/Leonardo-Interactive/leonardo-python-sdk and https://github.com/Leonardo-Interactive/leonardo-ts-sdk
• 限额、按量付费和定价：https://docs.leonardo.ai/reference/limits, https://docs.leonardo.ai/docs/payg-guide, and 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, and https://leonardo.ai/master-services-agreement