Ideogram 的 webhook 不是订阅制。没有事件名要注册,也没有共享密钥要粘进某个设置页。你只是在一次异步请求上挂一个 webhook_url 查询参数,等这次请求里所有图片都生成完,Ideogram 把结果 POST 回来。一次请求,一次投递,外加一段必须你自己写的验签代码。

这个设计决定就是大部分接入事故的源头:它长得像别家那套 webhook,但它不是。

回调到底怎么发生

异步端点是 POST /v1/ideogram-v4/async/generate。它接受和同步路径一样的生成请求,立刻返回一个 URL-safe base64 的 generation_id,等图片就绪再把结果交给 webhook_url。回调地址必须是 HTTPS,私网主机、回环主机和云元数据服务都会被拒绝。

本站自绘示意图:一朵云轮廓向一个服务器方框发出带签名的请求,下方有一条回环的回调箭头
本站自绘示意图,非官方示例,用来说明云端发出签名请求与下方回环的回调箭头 AI Tool Blog

上面这段话里找不到事件类型,因为压根没有。请求体本身就是生成结果,所以同一个处理函数既能接回调,也能处理同步响应。

# webhook_url 是查询参数,既不是请求体字段,也不是后台设置项。
# Api-Key 是官方文档给出的鉴权请求头。
curl -s -X POST \
  "https://api.ideogram.ai/v1/ideogram-v4/async/generate?webhook_url=https%3A%2F%2Fexample.com%2Fideogram-webhook" \
  -H "Api-Key: ${IDEOGRAM_API_KEY}" \
  -F 'text_prompt=A poster on a wall that reads: "Everything you can imagine is real."' \
  -F "resolution=1024x1024" \
  -o accepted.json

# 这个 id 要留着:它既对应回调,也是轮询兜底的钥匙。
jq -r '.generation_id' accepted.json

回调里到底有什么

请求体镜像同步生成响应:generation_idcreated,以及一个 data 数组,数组每一项带 urlpromptresolutionseedis_image_safe。每次投递还会固定带一组请求头,其中三个是承重的。

请求头承载什么
X-Ideogram-Webhook-Generation-Idbase64 生成 id,与请求体字段一致
X-Ideogram-Webhook-User-Id发起请求账户的 base64 id
X-Ideogram-Webhook-Timestamp签名时刻的 Unix 秒,十进制
X-Ideogram-Webhook-Key-Id提示先试哪把签名密钥
X-Ideogram-Webhook-SignatureEd25519 签名,小写十六进制
Content-Typeapplication/json

这里有个官方页面互相打架的地方,写解析代码之前值得知道。webhooks 页给出的请求体顶层字段是 generation_id;异步端点参考页把同一个请求体描述成「镜像同步响应」,并把关联字段写成 request_id。两页都在 developer.ideogram.ai 下。别赌请求体,认请求头:两页都同意 X-Ideogram-Webhook-Generation-Id 等于你提交时拿到的那个 generation_id

证明这次投递是真的

验签靠的是一组公开密钥,不是共享密钥;靠的是原始字节,不是解析后的对象。

先把公钥取回来

https://api.ideogram.ai/v1/.well-known/jwks.json 发一个 GET。它按 JWK 形式列出 Ed25519 公钥,每把钥匙的 x 字段是 32 字节公钥,base64url 编码且不带 padding。少掉的那截 padding 是大多数人第一处踩坑的地方。把这份密钥集缓存起来,某个签名对缓存副本验不过时就刷新,以防密钥已经轮换。

重建被签名的消息

四个值,用单个换行符连接,顺序固定:generation-id 头、user-id 头、timestamp 头、以及原始请求体的 SHA-256 十六进制摘要。按 UTF-8 编码后逐把公钥校验,任一把通过就说明这次投递真实。

摘要才是事故现场。Ideogram 签的是它发出去的那串字节。先解析再重新序列化,键顺序或空白可能变化,摘要跟着变,一个完全真实的 webhook 也会验签失败。先拿原始字节:Flask 用 request.get_data(),FastAPI 与 Starlette 用 await request.body(),Django 用 request.body,Express 需要 express.raw() 中间件。

import base64
import hashlib
import requests
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

JWKS_URL = "https://api.ideogram.ai/v1/.well-known/jwks.json"
SIGNED = (
    "X-Ideogram-Webhook-Generation-Id",
    "X-Ideogram-Webhook-User-Id",
    "X-Ideogram-Webhook-Timestamp",
)

def b64url(value: str) -> bytes:
    # JWK 的 x 字段是不带 padding 的 base64url,先补齐再解码。
    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))

def public_keys() -> list[Ed25519PublicKey]:
    jwks = requests.get(JWKS_URL, timeout=5).json()
    return [Ed25519PublicKey.from_public_bytes(b64url(jwk["x"])) for jwk in jwks["keys"]]

def is_authentic(headers: dict, raw_body: bytes) -> bool:
    """raw_body 必须是收到的原始字节,不能是重新序列化过的字典。"""
    digest = hashlib.sha256(raw_body).hexdigest()
    message = "\n".join([*(headers[name] for name in SIGNED), digest]).encode("utf-8")
    signature = bytes.fromhex(headers["X-Ideogram-Webhook-Signature"])
    # Key-Id 只是提示,不是要求:逐把试,密钥轮换才伤不到你。
    for key in public_keys():
        try:
            key.verify(signature, message)
            return True
        except InvalidSignature:
            continue
    return False

一个扛得住重试的处理函数

返回 2xx 才算确认。别的状态码,或者超时,Ideogram 就重试。这一句话决定了端点的形状:必须幂等,而且对同一次投递的重复到达也必须答对。

用 generation id 作为工作单元

同一个 generation_id 可能到达不止一次,端点慢一下就会触发重试。把完成状态挂在 generation id 上,已经处理过的 id 当作空操作,同时照旧返回 2xx 让重试停下。在已经存着你任务的那张表里加一个唯一键就够了。

import os
import time

import requests
from flask import Flask, request

app = Flask(__name__)
BASE = "https://api.ideogram.ai/v1"
HEADERS = {"Api-Key": os.environ["IDEOGRAM_API_KEY"]}
DONE: set[str] = set()

def finished(generation_id: str) -> dict:
    while True:
        payload = requests.get(
            f"{BASE}/generations/{generation_id}", headers=HEADERS, timeout=30
        ).json()
        if payload.get("status") != "pending":
            return payload
        time.sleep(5)

@app.post("/ideogram-webhook")
def ideogram_webhook():
    raw = request.get_data()  # 原始字节;先解析会把摘要弄坏
    if not is_authentic(dict(request.headers), raw):
        return {"error": "invalid signature"}, 400

    generation_id = request.headers["X-Ideogram-Webhook-Generation-Id"]
    if generation_id in DONE:
        return {"ok": True, "duplicate": True}
    DONE.add(generation_id)

    payload = request.get_json()
    for index, image in enumerate(payload["data"]):
        if not image["is_image_safe"]:
            continue
        # 图片链接会过期,所以字节要在这次请求里就落盘。
        with open(f"{generation_id}-{index}.png", "wb") as handle:
            handle.write(requests.get(image["url"], timeout=60).content)
    return {"ok": True}

if __name__ == "__main__":
    finished(os.environ["GENERATION_ID"])

webhook 一直没来的时候

官方写得很直白:投递不保证送达。重试次数有限,用完就丢弃,所以端点宕得够久,就会彻底听不到某次已经成功的生成。而且没有任何地方会报错。

拿同一个 generation id 去轮询

留着提交时拿到的 id,按节奏请求 GET /v1/generations/{generation_id}。生成完成后它返回同一个 data 载荷,所以一个处理函数就能同时服务两条路径。

官方没有公布的数字

三个官方页面讲了这个机制,没有一个给运维细节配数字。第三方教程给了数字的,那是它自己的估计。

  • 重试次数与节奏。 只说「有限次」,没有具体数字。
  • 时间戳容忍窗口。 时间戳在签名消息里,所以重放是相关的,但官方没给可接受的时钟偏移。
  • 密钥轮换周期。 密钥集可缓存可刷新,没给刷新间隔。
  • 回调超时阈值。2xx 或超时会触发重试,阈值是多少没写。
  • 图片链接存活时间。 链接会过期、必须下载,官方没给 TTL。
  • 投递来源地址。 官方给出的唯一过滤条件就是 HTTPS,加上拒绝私网、回环与元数据主机。

API 端点参考 覆盖了其余接口面,Ideogram 4.0 提示词指南 讲的是这条异步路由接受的 json_prompt 对象。两个官方页面都同意的事实值得重复一遍:text_promptjson_prompt 互斥,而 rendering_speed=FLASH 目前对 Ideogram 4.0 返回 400。异步生成按张计费,价格拆解 里有每张的单价。

常见问题

请求体里有事件类型字段吗? 没有。官方文档没有定义任何事件名,请求体就是生成结果本身,不是信封。

需要和 Ideogram 共享一个密钥吗? 不需要。验签用的是 JWKS 端点上的 Ed25519 公钥,你这边没有密钥可轮换,也没有密钥可泄露。

能先解析 JSON 再验签吗? 不能。签名覆盖的是原始字节的摘要,重新序列化会改变字节,哪怕这次投递完全真实。

端点挂了会怎样? 这次投递就丢了。用提交时拿到的 id 去轮询 GET /v1/generations/{generation_id};如果你想先看模型跑起来,上手指南 是浏览器那一侧的做法。

先把轮询这条路铺好,再挂回调,验签放在解析之前。三个决定就覆盖了规范真正承诺的全部内容。