Skip to content

웹훅

Potato에서 7가지 주석 이벤트 유형에 대해 HMAC 서명된 HTTP 웹훅 알림을 보냅니다 — 지수 백오프 재시도, 관리자 모니터링, Python/Node 검증 포함.

v2.4.0 신규 기능

웹훅을 사용하면 주석 이벤트가 발생할 때 Potato가 외부 시스템에 알릴 수 있습니다 — 폴링 없이도 가능합니다. 데이터 파이프라인에 연결하고, 알림을 트리거하고, 대시보드를 업데이트하거나, 후속 처리를 자동으로 시작하세요.

개요

Potato는 지원되는 이벤트가 발생할 때마다 구성된 엔드포인트로 HTTP POST 요청을 보냅니다. 페이로드는 JSON이며 HMAC-SHA256으로 서명되므로 해당 페이로드가 본인의 Potato 인스턴스에서 왔는지 확인할 수 있습니다.

웹훅 전달은 완전히 논블로킹입니다 — 웹훅이 전송 중이어도 주석 요청이 지연되지 않습니다. 실패한 전달은 지수 백오프로 재시도됩니다.

설정

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)를 사용하여 각 요청에 서명합니다. 세 가지 헤더가 포함됩니다:

헤더값
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

웹훅 상태와 통계를 확인합니다:

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"
      }
    }
  ]
}

테스트 웹훅 보내기

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를 트리거하는 주의력 검사를 구성합니다
  • 솔로 모드 — user.phase_completed를 트리거하는 단계 기반 워크플로
  • 내보내기 형식 — Potato에서 주석을 추출하는 대안적인 방법

구현 세부 정보는 원본 문서를 참조하세요.