OpenAI Responses API Webhook 怎么验签：原始请求体、时间窗口与 response.completed 处理

把 OpenAI Responses API 的长任务放到后台后，服务端一般没必要一直占着连接做轮询。Webhook 可以把 response.completed、失败或取消事件推回业务接口，但回调一到就直接更新订单、发通知，风险也跟着来了：请求体可能被中间件改写，重复投递可能让同一任务被重复执行两次，没做验签的请求绝对不能直接进入业务逻辑分支。

可靠的处理顺序是：先读取原始请求体，再用 OpenAI SDK 验证签名，验证通过后解析事件；随后用事件 ID 或响应 ID 做幂等判断，最后才查询 Responses API 并更新业务状态。

先把回调链路拆成三段

一条可维护的链路至少分成三个部分：入口层接收 HTTP body，安全层验证签名，业务层根据事件类型处理结果。入口层不要把 body 交给会自动格式化 JSON 的中间件之后就丢掉原文，因为签名校验针对的是接收到的原始内容。

OpenAI 的 Webhook 事件使用 webhook-id 标识一次投递，webhook-timestamp 表示投递时间，webhook-signature 携带签名。事件类型则放在 JSON 的 type 字段中；后台 Responses 完成时，常见类型是 response.completed。

Python 中保留原始 body 再验签

下面用 Flask 写一个最小入口示例。request.get_data() 取到的是请求到达时的原始字节；先把它交给 client.webhooks.unwrap()，成功后 SDK 才会返回已解析的事件对象。

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

app = Flask(__name__)
client = OpenAI()
secret = os.environ["OPENAI_WEBHOOK_SECRET"]

@app.post("/hooks/openai")
def receive_openai_hook():
    raw_body = request.get_data()
    try:
        event = client.webhooks.unwrap(
            raw_body,
            request.headers,
            secret=secret,
        )
    except Exception:
        return {"message": "invalid signature"}, 400

    if event.type == "response.completed":
        response_id = event.data.id
        # 这里交给幂等队列，避免在 HTTP 请求中做长事务
        print("verified response", response_id)
    elif event.type in {"response.failed", "response.cancelled"}:
        print("verified non-success event", event.type)
    else:
        print("ignored event", event.type)

    return {"ok": True}, 200

如果希望把验证和解析拆开，可以先调用 verify_signature()，再对同一份原始 body 做 JSON 解析。两种写法都遵循同一个原则：原始 body 不能在验签前被重新序列化。

时间窗口不是装饰：它决定重放边界

签名正确不代表请求永远有效。官方 SDK 会检查时间戳，默认容忍窗口是 300 秒；超过窗口的旧请求应直接拒绝，服务器时间明显超前也会导致校验失败。容器里的时钟漂移、代理缓存回放、手工重放旧 body，都会在这个环节被拦截。

排查时可以按这个顺序走：

• 记录 webhook-id、接收时间和 HTTP 状态，不要记录签名密钥。
• 比对应用节点与可信时间源的差异，重点排查跨可用区节点。
• 确认反向代理没有缓存 POST 请求，也没有把旧请求重新投递到业务入口。
• 需要人工补偿时重新发起业务查询，不要把历史 body 原样当成新回调处理。

response.completed 到业务落库还差一步

收到 response.completed 后，先用 webhook-id 做投递级幂等，再用 event.data.id 做响应级幂等。前者防范同一条 Webhook 重复到达，后者防范业务重试、补偿任务和多次通知共同处理同一个 Responses 响应。

def handle_completed(event, store, responses_client):
    delivery_key = event.id
    response_id = event.data.id

    if store.has_delivery(delivery_key):
        return "duplicate delivery"
    store.save_delivery(delivery_key, response_id)

    if store.has_response(response_id):
        return "already applied"

    response = responses_client.responses.retrieve(response_id)
    store.save_response(
        response_id=response.id,
        status=response.status,
        output_text=response.output_text,
    )
    return "applied"

这里把查询结果落库放在幂等记录之后，实际项目还需要事务或唯一索引兜底。如果处理过程可能超过网关超时，入口只负责验签并写入内部队列，消费者再完成查询和业务更新。

四个容易误判的失败现场

常见问题

Webhook 验签必须自己实现 HMAC 吗？

不必优先自己拼接签名算法。Python、Node.js、Go 等官方 SDK 都提供了验证入口；只有在特殊运行环境下，才考虑使用 Standard Webhooks 兼容库，并严格保留原始 body 和三个签名相关请求头。

为什么先 json.loads 再验签会失败？

JSON 重新序列化可能改变空格、转义或字段顺序，得到的字节串就不再是服务端签名时的原文。应先验证收到的原始字符串或字节，再做解析。

response.completed 可以直接当成业务成功吗？

不能。它表示 Responses 响应完成，业务还要读取响应状态和输出内容，并根据自身规则判断是否可以更新订单、消息或任务记录。

重复 Webhook 要返回什么状态？

只要签名有效，重复事件可以快速返回成功，避免发送方持续重试；真正的业务去重由数据库唯一约束或幂等存储完成。

把安全门槛放在业务分支之前

Webhook 入口最值得坚持的顺序只有一句话：原始 body 保留好，签名先验，事件再解析，结果按两个 ID 幂等处理。这样即使回调重复、处理进程重启，或者一次查询暂时失败，也能把补偿范围限制在一条可追踪的响应记录内。