Skip to content

Webhook

Potatoから7種類のアノテーションイベントについてHMAC署名付きのHTTP webhook通知を送ります。指数バックオフによる再送、管理画面での監視、PythonとNodeでの検証に対応しています。

v2.4.0の新機能

webhookを使うと、アノテーションのイベントが起きたときに、ポーリングなしでPotatoから外部システムへ通知できます。データパイプラインとの接続、アラートの発火、ダッシュボードの更新、後続処理の自動起動などに使えます。

概要

対応しているイベントが発生するたびに、Potatoは設定したエンドポイントへHTTP POSTリクエストを送ります。ペイロードは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アノテーターがフェーズを完了したとき（ソロモードのワークフロー）
quality.attention_check_failedアノテーターが注意チェックに失敗したとき
webhook.testテスト用に管理APIから手動で発火させたとき

"*"を指定すると、現在および将来のすべてのイベント種別を購読します。

ペイロードの形式

すべてのイベントは共通のエンベロープを持ちます。

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

annotation.createdのペイロード

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のペイロード

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のペイロード

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）を使って各リクエストに署名します。3つのヘッダーが付きます。

ヘッダー値
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回失敗すると、その配信は恒久的な失敗として記録されます。再送キューは{output_dir}/.webhooks/webhook_retries.dbのSQLiteに永続化されるため、サーバーを再起動しても保留中の再送は残ります。

監視とテスト

管理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)

参考資料

実装の詳細については、ソースドキュメントを参照してください。