Skip to content

Webhooks

HMAC-signierte HTTP-Webhook-Benachrichtigungen aus Potato für 7 Arten von Annotationsereignissen senden, mit Wiederholungsversuchen per exponentiellem Backoff, Monitoring im Admin-Bereich und Verifikation in Python und Node.

Neu in v2.4.0

Mit Webhooks meldet Potato externen Systemen, dass ein Annotationsereignis eingetreten ist, ohne dass diese pollen müssen. So lassen sich Datenpipelines anbinden, Alarme auslösen, Dashboards aktualisieren oder nachgelagerte Verarbeitung automatisch anstoßen.

Überblick

Potato schickt einen HTTP-POST-Request an den konfigurierten Endpunkt, sobald ein unterstütztes Ereignis auftritt. Die Payloads sind JSON und mit HMAC-SHA256 signiert, sodass sich prüfen lässt, ob sie aus der eigenen Potato-Instanz stammen.

Die Zustellung läuft vollständig nicht blockierend: Annotationsanfragen werden nie verzögert, während Webhooks unterwegs sind. Fehlgeschlagene Zustellungen werden mit exponentiellem Backoff wiederholt.

Konfiguration

Einen Abschnitt webhooks in der YAML-Konfiguration ergänzen:

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

Mehrere Endpunkte

Es lassen sich mehrere Endpunkte konfigurieren, jeder mit eigenen Ereignis-Abonnements:

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

Ereignistypen

EreignisWird ausgelöst, wenn
annotation.createdein Annotator ein Label für eine Instanz abschickt
annotation.updatedein Annotator ein zuvor abgeschicktes Label ändert
item.fully_annotatedeine Instanz die geforderte Zahl an Mehrfachannotationen erreicht
task.completedalle Instanzen der Aufgabe vollständig annotiert sind
user.phase_completedein Annotator eine Phase abschließt (Ablauf im Solo-Modus)
quality.attention_check_failedein Annotator eine Aufmerksamkeitsprüfung nicht besteht
webhook.testder Test manuell über die Admin-API ausgelöst wird

Mit "*" werden alle aktuellen und künftigen Ereignistypen abonniert.

Aufbau der Payload

Alle Ereignisse teilen sich denselben Rahmen:

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

Payload von 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"
  }
}

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

Payload von 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"
  }
}

Signaturen prüfen

Ist ein secret konfiguriert, signiert Potato jeden Request nach Standard Webhooks (HMAC-SHA256). Drei Header werden mitgeschickt:

HeaderWert
webhook-ideindeutige ID der Zustellung
webhook-timestampUnix-Zeitstempel der Zustellung
webhook-signatureHMAC-SHA256-Signatur

Prüfung in 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)

Prüfung in 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)
  );
}

Verhalten bei Wiederholungen

Fehlgeschlagene Zustellungen (Antwort außerhalb von 2xx oder Zeitüberschreitung) werden automatisch wiederholt:

VersuchWartezeit
1 (erster)sofort
25 Sekunden
330 Sekunden
45 Minuten
530 Minuten
61 Stunde

Nach 6 fehlgeschlagenen Versuchen gilt die Zustellung als endgültig gescheitert. Die Warteschlange für Wiederholungen liegt in einer SQLite-Datei unter {output_dir}/.webhooks/webhook_retries.db, sodass ausstehende Wiederholungen einen Neustart des Servers überdauern.

Monitoring und Test

Admin-API

Status und Statistiken der Webhooks abfragen:

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

Antwort:

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

Einen Test-Webhook senden

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

Das löst sofort ein webhook.test-Ereignis an den genannten Endpunkt aus.

Vollständige Konfigurationsreferenz

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)

Weiterführende Literatur

  • Admin-Dashboard — Annotationsfortschritt verfolgen und Nutzer verwalten
  • Qualitätskontrolle — Aufmerksamkeitsprüfungen einrichten, die quality.attention_check_failed auslösen
  • Solo-Modus — phasenbasierter Ablauf, der user.phase_completed auslöst
  • Exportformate — andere Wege, Annotationen aus Potato herauszubekommen

Implementierungsdetails stehen in der Quelldokumentation.