回调没人接、结果全靠猜 - Webhook 与异步任务接入的四个翻车现场
先说失败案例
接入 AI 能力的异步接口时，很多独立开发者和小团队会踩进同一个坑。下面四个场景，如果你正在做类似的事情，大概率已经中招了。
翻车一：用同步思维调异步接口。 你调了一个文档摘要的批量任务接口，拿到一个 task_id，然后写了个 while True 循环不停轮询。本地测试一切正常，上线后任务一多，你的服务器 CPU 全耗在轮询上，对面接口也开始限流。更糟的是，任务平均要跑两三分钟，你的请求超时设置早就到了，用户看到的是白屏。
翻车二：Webhook 回调地址是个“死胡同”。 你注册时随手填了个回调 URL，指向你本地开发的 localhost:8000。任务跑完了，ThisToken.AI 的服务器试图把结果推给你——推给谁？推给它自己？结果全部丢失，你只能靠轮询接口一个一个捞，捞回来还不知道哪条对应哪个任务。
翻车三：回调收到了，但没人认领。 你好不容易把回调打通了，Webhook 发来的 POST 请求打进了你的服务，但你的代码里压根没有处理这个事件的逻辑。请求落进了日志，没人解析，任务状态永远是“处理中”。用户等半天，去后台一看，结果其实早就到了。
翻车四：回调验签跳过，接口裸奔。 你觉得“反正就我一个项目在用，验不验签无所谓”。结果回调地址被人扫到，伪造的请求带着假数据打进来，你的系统照单全收，写入数据库。出问题时你连这条数据是真是假都分不清。
这四个坑的共同点是：把异步任务当同步任务用，把 Webhook 当摆设。下面是正确的做法。
正确路径：跑通 ThisToken.AI 的异步任务 + Webhook
第一步：注册并获取 API Key
- 打开 https://api.thistoken.ai/register ，用邮箱注册一个账号（价格与套餐以官网价格页为准）。
- 登录后进入控制台，在「API Keys」页面创建一个 Key，复制并妥善保存——它只显示一次。
- 在「Webhook 设置」里填写你的回调地址，必须是公网可访问的 HTTPS 地址。本地开发时推荐用 ngrok 之类的内网穿透工具临时映射一个。
第二步：跑通第一段代码
以 Python 为例，先装依赖：
pip install requests flask下面这段代码包含两部分：提交异步任务，以及用 Flask 接收 Webhook 回调。
import os
import hmac
import hashlib
import requests
from flask import Flask, request, jsonify
API_KEY = os.environ["THISTOKEN_API_KEY"]
BASE_URL = "https://api.thistoken.ai/v1"
WEBHOOK_SECRET = "your-webhook-secret" # 控制台 Webhook 设置页可查看
app = Flask(__name__)
results = {} # 生产环境请用数据库
def submit_task(text: str) -> str:
"""提交一个异步摘要任务，返回 task_id"""
resp = requests.post(
f"{BASE_URL}/tasks/summarize",
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"text": text,
"callback_url": "https://your-domain.com/webhook/thistoken",
},
timeout=30,
)
resp.raise_for_status()
return resp.json()["task_id"]
@app.route("/webhook/thistoken", methods=["POST"])
def webhook():
# 1. 验签：确认请求确实来自 ThisToken.AI
signature = request.headers.get("X-Signature", "")
expected = hmac.new(
WEBHOOK_SECRET.encode(),
request.get_data(),
hashlib.sha256,
).hexdigest()
if not hmac.compare_digest(signature, expected):
return jsonify({"error": "invalid signature"}), 401
# 2. 解析事件，认领结果
payload = request.get_json()
if payload.get("event") == "task.completed":
task_id = payload["task_id"]
results[task_id] = payload["result"]
print(f"任务 {task_id} 完成，已入库")
# 3. 快速返回 200，告诉对方"收到了"
return "", 200
if __name__ == "__main__":
task_id = submit_task("这里放一段需要摘要的长文本……")
print(f"任务已提交：{task_id}，等待回调…")
app.run(port=8000)几个关键点，对照前面的翻车现场看：
- 拿到 task_id 就返回，不要死等结果。
submit_task发出请求后立刻结束，把等待的活儿交给 Webhook。 - 回调地址必须是公网可达的 HTTPS 地址。 本地调试用内网穿透，上线后换成正式域名。
- 验签不可省略。 用 HMAC 比对
X-Signature，且用hmac.compare_digest而非==，避免时序攻击。具体的签名算法和 Header 名称以官方文档为准，思路是一致的。 - 回调处理要快。 收到请求后先落库、快速返回 200，复杂处理放到后台任务里做。回调端超时重试是平台侧的常见机制，你的接口拖太久会收到重复推送，所以处理逻辑要做幂等（同一个 task_id 重复写入时先检查是否已存在）。
- 轮询是兜底，不是主路。 保留一个按 task_id 查询状态的接口调用，用于 Webhook 丢失后的补偿排查，但别把它当主要手段。
上线前的检查清单
- 回调地址在公网可访问，且配置了 HTTPS；
- 验签逻辑已启用，密钥没有硬编码进代码仓库；
- 回调处理做了幂等，重复推送不会产生脏数据；
- 任务失败事件（如
task.failed）也有对应的处理和告警； - 有一个手动查询任务状态的兜底入口。
把这套流程跑通之后，异步任务的接入就从“碰运气”变成了“有据可查”：每次提交有 task_id，每次回调有验签和落库，出了问题能顺着日志查到具体环节。
如果你还没注册，可以从这里开始：https://api.thistoken.ai/register
---
不想折腾多家供应商的接入差异？在 https://api.thistoken.ai/register 注册，用一个 base_url 调用所有模型。