智能体 Webhook 怎么接?验证 OpenAI 签名并防止重复处理事件

接收 OpenAI Webhook 时保留原始正文验签,快速响应,并用持久化唯一键防止事件重复处理。

Webhook 让 OpenAI 在批任务、后台响应等事件发生时主动通知你的服务器。可靠接入需要三件事:用原始请求体验证签名、尽快返回 2xx、按事件 ID 或业务资源 ID 做幂等,防止重试造成重复入库或重复发消息。

配置密钥并保留原始正文

在项目中创建 Webhook 后,把签名密钥保存到服务端的 OPENAI_WEBHOOK_SECRET。以 Flask 为例:

智能体 Webhook 怎么接?验证 OpenAI 签名并防止重复处理事件

import os
from flask import Flask, request, Response
from openai import OpenAI, InvalidWebhookSignatureError

app = Flask(__name__)
client = OpenAI(webhook_secret=os.environ["OPENAI_WEBHOOK_SECRET"])
processed = set()  # 演示用;生产环境应使用持久化唯一约束

@app.post("/openai/webhook")
def webhook():
    try:
        event = client.webhooks.unwrap(
            request.get_data(as_text=True), request.headers
        )
    except InvalidWebhookSignatureError:
        return Response("invalid signature", status=400)

    event_key = getattr(event, "id", None) or f"{event.type}:{event.data.id}"
    if event_key in processed:
        return Response(status=200)

    processed.add(event_key)
    # 生产环境把后续处理放入队列,再返回 200
    print(event.type, event.data.id)
    return Response(status=200)

为什么要先验签

只有签名验证通过,才能把请求当作可信的 OpenAI 事件。某些框架会先把 JSON 解析成对象或改变正文,可能导致验签失败;应按 SDK 文档把原始字符串和请求头传给 webhooks.unwrap。

用重复事件测试幂等

  1. 保存一条测试事件的原始正文和必要请求头。
  2. 在受控环境连续投递两次。
  3. 确认签名均通过,但业务记录只新增一次。
  4. 让处理函数故意超时,确认平台重试不会造成重复副作用。

内存集合只适合演示。生产中应给事件键建立数据库唯一索引,并把耗时工作放入队列。先持久化接收结果再返回成功,可避免服务器崩溃后事件无迹可查。

本文依据 2026 年 10 月 1 日官方 Webhooks 指南,未连接你的公网地址、框架中间件或签名密钥。

相关问题

返回 200 就表示业务完成了吗?

不一定。通常只表示事件已可靠接收;后续队列任务还要单独记录成功或失败。

能只按 response_id 去重吗?

要根据事件类型和业务语义设计。不同事件可能引用同一资源,事件键需避免误合并。

参考:OpenAI Webhooks

Ai菜鸟网。发布者:AI小管家,转载请注明出处:https://www.alyyhw.com/31266.html

赞 (0)
AI小管家的头像AI小管家
AI 内容审核工具怎么接入?用 OpenAI Moderation 检查文本与图片
上一篇 1天前
AI 工具怎么生成图片?用 OpenAI 图片 API 创建、保存并核对结果
下一篇 1天前

相关推荐

联系我们

联系我们

1

在线咨询: QQ交谈

邮件:admin@example.com

工作时间:周一至周五,9:30-18:30,节假日休息

关注微信
关注微信
分享本页
返回顶部