官方建议用 webhook 代替轮询：创建 API Key 时填写回调地址（必须是 HTTPS）和一个 webhook API Key，Leonardo 回调时会带上
authorization: Bearer <这个 Key>。 这篇写一个接收端：
- 校验 Key（常量时间比较）；
- 同一个生成 ID 只处理一次（并发重复投递的情况也覆盖）；
- nsfw 图片不对外展示；
- 其他事件类型返回 200，但不处理。
先看清成品（box 上真实运行，使用官方 webhook 指南中的示例负载）
- 第一次投递返回
ok，再次投递返回duplicate；20 个并发投递中只有 1 个是ok，其余 19 个是duplicate。- Key 错误或缺失时返回 401；负载格式错误时返回 400。pytest 5 项通过。
合规提醒：Leonardo API 需要单独购买 API 额度，和网页版订阅是分开的。API Key 只放在服务端环境变量里，不要放进前端代码，也不要共享账号或 Key。生成内容要遵守 Leonardo 的条款和内容政策：官方会默认拦截 NSFW 提示词，并在结果里用
nsfw字段标记图片。不要尝试绕过审核，也不要生成侵犯他人肖像权、商标或版权的内容。请求体里的public字段决定图片是否出现在社区公开流中，处理敏感素材时设为 false。本文不列价格。
运行说明：接收端和测试都在 box 上真实运行过，但没有收到 Leonardo 的真实回调（需要 API Key 和公网 HTTPS 地址）。示例负载取自官方 webhook 指南，我删掉了大量与本例无关的字段。
做完之后你会得到什么？
| 成果 | 说明 |
|---|---|
webhook_server.py |
接收端，用 SQLite 存储 |
sample_webhook.json |
官方示例负载（删减版） |
test_webhook.py |
5 项测试，包括并发重复投递 |
需要准备哪些环境？
Python 3.9 以上（只用标准库）；测试需要 pip install pytest。上线时需要一个公网 HTTPS 域名和反向代理。
关键事实（来自官方 webhook 指南）
| 事实 | 内容 |
|---|---|
| 设置位置 | 创建 production API Key 时填写回调 URL（必须是 HTTPS）和 webhook callback API key |
| 鉴权方式 | 回调请求带有 authorization: Bearer $yourWebhookCallbackApiKey |
| 修改设置 | 新建一个 Key 并填写新的回调设置，切换过去后再删除旧 Key |
| 事件 | "type": "image_generation.complete"，数据在 data.object（包含 images[].url 和 nsfw） |
文件结构
63-leonardo-ai-C-src/
├── webhook_server.py
├── sample_webhook.json
└── test_webhook.py
第 1 步：接收端
"""接收 Leonardo 的 webhook 回调（标准库实现，教学用）。
依据官方《Guide to the Webhook Callback Feature》：
- 回调地址必须是 HTTPS（本地开发用 HTTP，上线时放在 HTTPS 反向代理之后）
- Leonardo 会带上请求头 authorization: Bearer <你在创建 API Key 时填写的 webhook callback API key>
- 生成完成的事件类型是 image_generation.complete，数据在 data.object 中
处理原则：
- Key 不对就返回 401，并且用 hmac.compare_digest 做常量时间比较
- 同一个生成 id 只处理一次（重复投递时直接返回 200）
- nsfw 为 true 的图片不对外展示，只记录
- 只做轻量处理并尽快返回 200，耗时任务交给队列"""
import hmac, json, os, sqlite3, sys, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
MAX_BODY = 2 * 1024 * 1024
class Store:
def __init__(self, path=":memory:"):
self.db = sqlite3.connect(path, check_same_thread=False); self.lock = threading.Lock()
self.db.execute("create table if not exists gen(id text primary key, prompt text, received_at text default current_timestamp)")
self.db.execute("create table if not exists img(id text primary key, gen_id text, url text, nsfw int)")
def add(self, obj):
"""返回 True 表示是新的生成；False 表示重复投递。"""
with self.lock, self.db:
cur = self.db.execute("insert or ignore into gen(id, prompt) values (?, ?)", (obj["id"], obj.get("prompt")))
if cur.rowcount == 0:
return False
for im in obj.get("images", []):
self.db.execute("insert or ignore into img values (?, ?, ?, ?)", (im["id"], obj["id"], im["url"], int(bool(im.get("nsfw")))))
return True
def public_images(self):
return [r[0] for r in self.db.execute("select url from img where nsfw = 0 order by id")]
def make_handler(secret, store):
class H(BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def _reply(self, code, text):
raw = text.encode(); self.send_response(code); self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(raw))); self.end_headers(); self.wfile.write(raw)
def do_POST(self):
auth = self.headers.get("authorization", "")
if not hmac.compare_digest(auth.encode(), f"Bearer {secret}".encode()):
return self._reply(401, "unauthorized")
n = int(self.headers.get("Content-Length") or 0)
if n <= 0 or n > MAX_BODY:
return self._reply(413 if n > MAX_BODY else 400, "bad length")
try:
ev = json.loads(self.rfile.read(n))
obj = ev["data"]["object"]
except (ValueError, KeyError, TypeError):
return self._reply(400, "bad payload")
if ev.get("type") != "image_generation.complete":
return self._reply(200, f"ignored {ev.get('type')}") # 其他事件：确认收到，但不处理
new = store.add(obj)
self._reply(200, "ok" if new else "duplicate")
return H
def serve(secret, port=0, store=None):
store = store or Store()
s = ThreadingHTTPServer(("127.0.0.1", port), make_handler(secret, store))
threading.Thread(target=s.serve_forever, daemon=True).start()
return s, store
if __name__ == "__main__":
secret = os.environ["LEONARDO_WEBHOOK_KEY"]
s, store = serve(secret, int(sys.argv[1]) if len(sys.argv) > 1 else 8788, Store("webhook.db"))
print(f"listening on http://127.0.0.1:{s.server_address[1]}（上线时放在 HTTPS 反向代理之后）")
threading.Event().wait()
第 2 步：测试
import copy, json, urllib.error, urllib.request
import pytest
import webhook_server as W
SAMPLE = json.load(open("sample_webhook.json", encoding="utf-8"))
@pytest.fixture
def srv():
s, store = W.serve("whk_test"); yield f"http://127.0.0.1:{s.server_address[1]}", store; s.shutdown()
def post(url, body, auth="Bearer whk_test", raw=None):
data = raw if raw is not None else json.dumps(body).encode()
req = urllib.request.Request(url, data=data, method="POST", headers={"authorization": auth, "content-type": "application/json"})
try:
with urllib.request.urlopen(req) as r:
return r.status, r.read().decode()
except urllib.error.HTTPError as e:
return e.code, e.read().decode()
def test_official_sample_accepted_once(srv):
url, store = srv
assert post(url, SAMPLE) == (200, "ok")
assert post(url, SAMPLE) == (200, "duplicate")
assert store.public_images() == [SAMPLE["data"]["object"]["images"][0]["url"]]
def test_wrong_or_missing_key(srv):
url, _ = srv
assert post(url, SAMPLE, auth="Bearer nope")[0] == 401
assert post(url, SAMPLE, auth="")[0] == 401
def test_nsfw_not_public(srv):
url, store = srv
ev = copy.deepcopy(SAMPLE); obj = ev["data"]["object"]; obj["id"] = "gen-2"
obj["images"] = [{"id": "a", "url": "https://x/a.jpg", "nsfw": False}, {"id": "b", "url": "https://x/b.jpg", "nsfw": True}]
assert post(url, ev) == (200, "ok") and store.public_images() == ["https://x/a.jpg"]
def test_other_event_and_bad_payload(srv):
url, _ = srv
assert post(url, {**SAMPLE, "type": "something.else"}) == (200, "ignored something.else")
assert post(url, None, raw=b"{not json")[0] == 400
assert post(url, {"type": "image_generation.complete"})[0] == 400
def test_concurrent_duplicates_processed_once(srv):
import collections, concurrent.futures as cf
url, _ = srv
with cf.ThreadPoolExecutor(20) as ex:
res = list(ex.map(lambda _: post(url, SAMPLE), range(20)))
assert collections.Counter(res) == {(200, "ok"): 1, (200, "duplicate"): 19}
第 3 步：运行
$ LEONARDO_WEBHOOK_KEY=whk_xxx python webhook_server.py 8788 &
$ curl -s -X POST localhost:8788 -H 'authorization: Bearer whk_xxx' -H 'content-type: application/json' -d @sample_webhook.json
ok
$ python -m pytest -q
5 passed
调试示例
1. 去掉锁之后，并发投递会把连接弄坏。 为了确认锁是不是必要的，我用 contextlib.nullcontext() 替换掉锁，再并发发送 20 个相同的回调。结果服务器端出现 sqlite3.InterfaceError: bad parameter or other API misuse 和 sqlite3.OperationalError: cannot commit - no transaction is active，客户端收到 RemoteDisconnected。也就是说，Leonardo 会认为这次投递失败了。原因是多个线程共用了同一个 SQLite 连接。恢复锁之后，20 次投递稳定得到 1 个 ok 和 19 个 duplicate，换成文件数据库也是同样的结果，这个场景已经写进测试。
2. 示例负载里的 ID 都是占位符。 官方示例中，生成 ID 和图片 ID 都是 XXXXXXXX-XXXX-...。我把生成和图片分别放在两张表里，所以两者相同也不会冲突。实际使用时，它们是不同的 UUID。
常见问题
| 现象 | 原因 | 处理 |
|---|---|---|
| 收不到回调 | 不是 HTTPS、证书链不完整，或者防火墙拦截 | 按指南用 openssl s_client 检查证书 |
| 总是返回 401 | 系统没有读取 authorization 头，或者 Key 不一致 | 先把请求记录下来检查 |
| 重复处理同一个生成 | 没有做幂等 | 用生成 ID 作为主键 |
| 想更换回调地址 | 只能新建 Key | 新建 Key，切换过去后再删除旧的 |
模板规则
- 校验 Bearer Key 时使用常量时间比较。
- 以生成 ID 作为幂等键。
- 尽快返回 200，耗时处理放到队列里。
- nsfw 图片不对外展示，只记录。
项目资源
webhook 接收端要做到两点：拒绝伪造的请求，重复投递时只处理一次。