用签名Webhook替代任务轮询

大多数我接入过的视频 API 都要求轮询。你 POST 一个任务，拿到一个 id，然后写个循环，每隔几秒问一次“好了吗”，直到返回结果，或者你猜定的超时时间耗尽。这个循环正是 bug 滋生的地方：重试导致重复处理、任务在无响应时永远挂起，以及调度器在什么都没发生时悄悄耗尽 rate limit。

在构建 OpenShorts——一款将长视频剪辑成竖屏短视频的开源工具——时，我们做了三个 API 设计决策，消除了其中的大部分痛苦。下面是这些决策，以及背后的思考和代码。

1. 一个请求进来，一个签名 webhook 出去

整个自动化闭环只有两条 HTTP 消息。启动一个任务：

curl -X POST https://api.openshorts.app/api/process \
  -H "Authorization: Bearer osk_..." \
  -H "Content-Type: application/json" \
  -d '{"url": "https://youtube.com/watch?v=...",
       "acknowledged": true,
       "webhook_url": "https://your-server.com/hooks/openshorts",
       "webhook_secret": "your-shared-secret"}'

你会立即收到一个 job id。几分钟后，当视频片段完成剪辑、添加字幕并归档时，你的 webhook 会收到且只会收到一次 POST，其中包含片段标题和持久有效的下载链接。

有一点比听起来更重要：任务失败时也会触发 webhook。如果失败是静默的，每个消费者都不得不重新发明一套超时机制，而且每个人设定的超时时间都不一样。只有把失败变成一个会送达的事件，而不是“没有事件发生”，pipeline 才能做到无状态。

2. 为回调签名，并使用常量时间比较

如果传入了 webhook_secret，投递请求会携带一个格式为 sha256=<hex> 的 X-OpenShorts-Signature header，其值是原始 request body 的 HMAC-SHA256。

import hmac, hashlib

expected = "sha256=" + hmac.new(secret, raw_body, hashlib.sha256).hexdigest()
if not hmac.compare_digest(expected, request.headers["X-OpenShorts-Signature"]):
    return 401

有两个细节经常被忽略，值得再强调一遍：

对 raw body 计算 hash，不要对解析后再重新序列化的 JSON 计算。键的顺序和空白字符经过 framework 的一次往返处理后不会保持不变。

使用 compare_digest，不要使用 ==。普通的字符串比较遇到第一个不同的 byte 就会提前返回，从而泄露你猜对了多少 prefix。

3. 把 MCP 当作一等接口，而不是一层包装

我们在 mcp.openshorts.app/mcp 部署了一个 MCP server，让 Agent 可以直接驱动 pipeline。只需一行命令即可连接：

claude mcp add --transport http openshorts https://mcp.openshorts.app/mcp \
  --header "Authorization: Bearer osk_..."

任何支持 Streamable HTTP 的 client 都能以相同方式工作。server 会通过 protocol 描述自身，包括 tool schema，因此无需配置其他内容。

六个 tool 覆盖了整个 pipeline：process_video、get_job_status、list_clips、get_quota、add_subtitles 和 publish_clip。

get_quota 是我最坚持要提供的一个。一个无法看到自身预算的 Agent，会毫不犹豫地启动一个它根本无法完成的任务，而你要等到二十分钟后收到失败 webhook 时才会发现。把剩余余额作为 tool 暴露出来，可以让 model 在消费额度之前先检查，将 runtime failure 转变为 planning decision。

我们拒绝做的另一件事，是把 MCP server 构建成 REST API 某个便利子集的包装层。每个 tool 调用的都是 web app 使用的同一套 pipeline，共用同一个 account、minutes 和 job history。一旦 Agent 接口只是一个子集，你就有两个产品需要保持同步，而 Agent 那个产品永远会落后。

计量方式是一项 API 设计决策

这是我之前没想到会如此重要的部分，结果它反而最为关键。

这个领域的大多数工具会单独计量 Agent call，按照来源视频时长或每次 operation 收费。这种计费方式有其合理性，但使用体验极差。Agent loop 天生就具有探索性：它会检查状态、列出片段、重试那个效果不佳的结果。如果每次 call 都有成本，那么正确的工程应对方式就会变成编写数量更少、规模更大、也更脆弱的 call，而这恰好与你希望 Agent 具备的特性背道而驰。

我们让 API call 与 dashboard 共用同一个固定的分钟余额。不设独立计量，也不按 call 收费。self-hosted edition 则完全没有计量机制，正因如此，持续运行的 pipeline 才具备可负担性。

调度方式随你现有的基础设施而定

由于启动一个任务只需要一次 POST，因此你不需要 scheduler integration。无论是一行 cron、一个 GitHub Action，还是一个 n8n HTTP Request node，都不需要专用 connector 就能工作。我们没有提供官方 n8n template，因为前面所述的两步结构就是完整的 integration。

代码位于 GitHub 的 mutonby/openshorts，核心部分采用 MIT license，self-hosted edition 提供与 cloud 相同的 MCP endpoint。

如果你正在构建任何涉及 webhook 和 Agent 的系统，总结起来很简单：将失败作为事件送达；对 body 签名并使用常量时间比较；把预算作为 tool 暴露出来；不要对 Agent 思考所需的 call 收费。

对于后续操作，你可以考虑屏蔽此人和/或举报滥用行为。