Webhook

从 Potato 发送带 HMAC 签名的 HTTP webhook 通知，覆盖 7 类标注事件，支持指数退避重试、管理端监控以及 Python/Node 验签。

v2.4.0 新增

Webhook 让 Potato 在标注事件发生时通知外部系统，不需要轮询。可以对接数据管线、触发告警、更新仪表盘，或自动启动下游处理。

概览

只要有受支持的事件触发，Potato 就会向你配置的端点发送一个 HTTP POST 请求。payload 是 JSON，并用 HMAC-SHA256 签名，你可以据此验证请求确实来自你的 Potato 实例。

Webhook 投递完全是非阻塞的——标注请求绝不会因为 webhook 还在发送中而被拖慢。投递失败会按指数退避重试。

配置

在 YAML 配置中加一个 webhooks 段：

yaml

webhooks:
  enabled: true
  endpoints:
    - name: "my-pipeline"
      url: "https://your-system.example.com/potato-events"
      secret: "your-signing-secret"        # optional but recommended
      events:
        - annotation.created
        - item.fully_annotated
        - task.completed
      active: true
      timeout: 10

多个端点

你可以配置多个端点，每个订阅不同的事件：

yaml

webhooks:
  enabled: true
  endpoints:
    - name: "data-pipeline"
      url: "https://pipeline.example.com/annotations"
      secret: ${WEBHOOK_SECRET_1}
      events:
        - annotation.created
        - item.fully_annotated
    - name: "slack-alerts"
      url: "https://hooks.slack.com/services/..."
      events:
        - task.completed
        - quality.attention_check_failed
    - name: "catch-all"
      url: "https://logging.example.com/potato"
      events:
        - "*"          # subscribe to all events

事件类型

事件	触发时机
`annotation.created`	标注者为某条实例提交了标签
`annotation.updated`	标注者修改了之前提交的标签
`item.fully_annotated`	某条实例达到了所需的标注重叠数
`task.completed`	任务中的所有实例都已完成标注
`user.phase_completed`	标注者完成了一个阶段（Solo 模式流程）
`quality.attention_check_failed`	标注者未通过注意力检查
`webhook.test`	通过管理 API 手动触发，用于测试

用 "*" 可以订阅当前及以后所有的事件类型。

Payload 格式

所有事件共用同一个外层结构：

json

{
  "event_id": "evt_01HXYZ...",
  "event_type": "annotation.created",
  "timestamp": "2026-03-17T14:23:01Z",
  "task_name": "sentiment-study",
  "data": {
    ...
  }
}

`annotation.created` 的 payload

json

{
  "event_type": "annotation.created",
  "data": {
    "annotator_id": "user123",
    "instance_id": "doc_042",
    "annotation": {
      "sentiment": "positive",
      "confidence": "high"
    },
    "submitted_at": "2026-03-17T14:23:01Z"
  }
}

`item.fully_annotated` 的 payload

json

{
  "event_type": "item.fully_annotated",
  "data": {
    "instance_id": "doc_042",
    "annotator_count": 3,
    "annotations": [
      {"annotator_id": "user1", "sentiment": "positive"},
      {"annotator_id": "user2", "sentiment": "positive"},
      {"annotator_id": "user3", "sentiment": "neutral"}
    ]
  }
}

`task.completed` 的 payload

json

{
  "event_type": "task.completed",
  "data": {
    "task_name": "sentiment-study",
    "total_instances": 500,
    "total_annotations": 1500,
    "completed_at": "2026-03-17T15:00:00Z"
  }
}

验证签名

配置了 secret 之后，Potato 会按 Standard Webhooks 规范（HMAC-SHA256）对每个请求签名。请求中包含三个 header：

Header	值
`webhook-id`	本次投递的唯一 ID
`webhook-timestamp`	投递时的 Unix 时间戳
`webhook-signature`	HMAC-SHA256 签名

Python 验签

python

import hmac
import hashlib
import time
 
def verify_webhook(payload_bytes: bytes, headers: dict, secret: str) -> bool:
    webhook_id = headers.get("webhook-id", "")
    timestamp = headers.get("webhook-timestamp", "")
    signature = headers.get("webhook-signature", "")
 
    # Reject stale requests (older than 5 minutes)
    if abs(time.time() - int(timestamp)) > 300:
        return False
 
    signed_content = f"{webhook_id}.{timestamp}.{payload_bytes.decode()}"
    expected = hmac.new(
        secret.encode(),
        signed_content.encode(),
        hashlib.sha256
    ).hexdigest()
 
    return hmac.compare_digest(f"v1,{expected}", signature)

Node.js 验签

javascript

const crypto = require('crypto');
 
function verifyWebhook(payload, headers, secret) {
  const webhookId = headers['webhook-id'];
  const timestamp = headers['webhook-timestamp'];
  const signature = headers['webhook-signature'];
 
  // Reject stale requests
  if (Math.abs(Date.now() / 1000 - parseInt(timestamp)) > 300) {
    return false;
  }
 
  const signedContent = `${webhookId}.${timestamp}.${payload}`;
  const expected = crypto
    .createHmac('sha256', secret)
    .update(signedContent)
    .digest('hex');
 
  return crypto.timingSafeEqual(
    Buffer.from(`v1,${expected}`),
    Buffer.from(signature)
  );
}

重试行为

投递失败（响应非 2xx 或超时）会自动重试：

尝试次数	延迟
1（首次）	立即
2	5 秒
3	30 秒
4	5 分钟
5	30 分钟
6	1 小时

6 次都失败后，该次投递被标记为永久失败。重试队列持久化在 SQLite 中，位于 {output_dir}/.webhooks/webhook_retries.db——待重试的任务在服务器重启后仍然保留。

监控与测试

管理 API

查看 webhook 状态和统计：

bash

# List all webhooks and their delivery statistics
curl -H "X-API-Key: $ADMIN_API_KEY" \
  http://localhost:8000/admin/api/webhooks

响应：

json

{
  "endpoints": [
    {
      "name": "my-pipeline",
      "url": "https://...",
      "events": ["annotation.created"],
      "active": true,
      "stats": {
        "total_emitted": 1240,
        "total_failed": 3,
        "pending_retries": 0,
        "last_success": "2026-03-17T14:23:01Z"
      }
    }
  ]
}

发送测试 webhook

bash

curl -X POST -H "X-API-Key: $ADMIN_API_KEY" \
  http://localhost:8000/admin/api/webhooks/test \
  -H "Content-Type: application/json" \
  -d '{"endpoint_name": "my-pipeline"}'

这会立即向指定端点发出一个 webhook.test 事件。

完整配置参考

yaml

webhooks:
  enabled: true
  endpoints:
    - name: string           # unique name for this endpoint
      url: string            # HTTPS URL to POST to
      secret: string         # optional HMAC secret for signature verification
      events:                # list of event types, or ["*"] for all
        - annotation.created
      active: true           # set false to disable without removing
      timeout: 10            # request timeout in seconds (default: 10)
      max_retries: 6         # max retry attempts (default: 6)

Webhook

概览

配置

多个端点

事件类型

Payload 格式

annotation.created 的 payload

item.fully_annotated 的 payload

task.completed 的 payload

验证签名

Python 验签

Node.js 验签

重试行为

监控与测试

管理 API

发送测试 webhook

完整配置参考

延伸阅读

`annotation.created` 的 payload

`item.fully_annotated` 的 payload

`task.completed` 的 payload