Skip to content

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-signatureHMAC-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（首次）立即
25 秒
330 秒
45 分鐘
530 分鐘
61 小時

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)

延伸閱讀

  • 管理後臺 —— 監控標註進度並管理使用者
  • 品質控制 —— 配置觸發 quality.attention_check_failed 的注意力檢查
  • Solo 模式 —— 觸發 user.phase_completed 的分階段流程
  • 匯出格式 —— 從 Potato 取出標註資料的其他方式

有關實現詳情，請參閱源文件。