Webhooks

Configure an HTTPS URL in Settings → Integrations → Webhooks (/settings/integration?tab=webhooks). When a tool column finishes for a row, Kuration may send your endpoint a tool_output_ready event via POST.

Slack Incoming Webhooks (notifications) use the same Integrations screen but are vendor-specific — see Slack notifications.

---

Event: tool_output_ready

Emitted when tool output has been produced for a cell (delivery is asynchronous and best-effort; failures do not roll back row processing).

Request your server receives

Method: POST
Content-Type: application/json

Headers

Header	Value
Content-Type	application/json
User-Agent	Kuration-Webhook/1.0
X-Kuration-Signature	sha256=<hex> — HMAC-SHA256 over the canonical JSON body (UTF-8), using your webhook secret.

The canonical signed string is json.dumps(payload_dict, separators=(',', ':')).encode('utf-8') (no spaces after :), while the HTTP body may use different JSON whitespace. GET the raw bytes with request.get_data(), parse JSON, then re-serialize with the same separators and UTF-8-encode before HMAC:

canonical = json.dumps(data, separators=(",", ":")).encode("utf-8")
expected = hmac.new(secret.encode("utf-8"), canonical, hashlib.sha256).hexdigest()

This matches generate_webhook_signature in the platform codebase.

Implement verification against the prefix form

The header looks like sha256=abcd.... Strip the sha256= prefix (or compare the full header to sha256=" + hex_digest) before calling hmac.compare_digest. Comparing plain hex to the full header string will fail.

Payload schema

JSON object:

{
  "event": "tool_output_ready",
  "project_id": "98e79cf8-87c9-49b0-9899-58edb6f312345",
  "row_id": "Jxl8KwcI6bAHH8wdQ123",
  "col_id": "466527f1-ef79-4cd6-bf03-c878681d1234",
  "value": "Technology",
  "status": "success",
  "timestamp": "2024-01-20T14:45:00.000000"
}

Field	Type	Description
event	string	Always tool_output_ready for this hook.
project_id	string	Project UUID.
row_id	string	Row identifier.
col_id	string	Tool column UUID.
value	string	Serialized tool output value (meaning depends on tool).
status	string	Outcome marker (e.g. success, error, skipped, etc.—matches internal tool run type).
timestamp	string	ISO-style timestamp string.

Configure and rotate the secret from the integrations page; treat it like a password (environment variables, secret manager).

---

Flask receiver example

from flask import Flask, request, jsonify
import hmac
import hashlib
import json

app = Flask(__name__)

WEBHOOK_SECRET = "YOUR_WEBHOOK_SECRET"  # from Kuration settings


def normalize_signature(header_value: str) -> str:
    """Return hex digest portion from X-Kuration-Signature (sha256=<hex>)."""
    if not header_value:
        return ""
    value = header_value.strip()
    if value.startswith("sha256="):
        return value[len("sha256=") :]
    return value


@app.route("/webhook", methods=["POST"])
def handle_webhook():
    signature_header = request.headers.get("X-Kuration-Signature", "")
    signature = normalize_signature(signature_header)

    if not signature:
        return jsonify({"error": "Missing signature"}), 401

    raw = request.get_data()
    try:
        data = json.loads(raw.decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError):
        return jsonify({"error": "Invalid JSON"}), 400

    canonical = json.dumps(data, separators=(",", ":")).encode("utf-8")
    expected = hmac.new(
        WEBHOOK_SECRET.encode("utf-8"),
        canonical,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return jsonify({"error": "Invalid signature"}), 401

    print(
        f"Received {data.get('event')} project={data.get('project_id')} "
        f"row={data.get('row_id')} col={data.get('col_id')}"
    )

    # Process data here; respond quickly — see below.

    return jsonify({"success": True}), 200


if __name__ == "__main__":
    app.run(port=5000)

---

Verification helper only

Useful if you integrate with frameworks other than Flask but still verify the MAC over raw bytes:

import hmac
import hashlib
import json


def verify_webhook_signature(raw_body: bytes, signature_header: str, secret: str) -> bool:
    """True if signature_header matches sha256 over canonical JSON (compact separators)."""
    if not signature_header:
        return False
    trimmed = signature_header.strip()
    if trimmed.startswith("sha256="):
        provided = trimmed[len("sha256=") :]
    else:
        provided = trimmed
    try:
        data = json.loads(raw_body.decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError):
        return False
    canonical = json.dumps(data, separators=(",", ":")).encode("utf-8")
    expected = hmac.new(secret.encode("utf-8"), canonical, hashlib.sha256).hexdigest()
    return hmac.compare_digest(provided, expected)


# sig = request.headers.get("X-Kuration-Signature", "")
# verify_webhook_signature(request.get_data(), sig, WEBHOOK_SECRET)

---

Security and delivery notes

• Verify every request with hmac.compare_digest (constant-time).
• Store the secret in environment variables or a secret manager — never commit it.
• Respond quickly: outbound delivery uses a ~5 second timeout; offload heavy work to a queue/async task and return 2xx promptly.
• Use HTTPS for your public URL.

---

Slack notifications

To send alerts to Slack, create an Incoming Webhook in Slack’s API dashboard, then connect it under Integrations → Webhooks (/settings/integration?tab=webhooks) using Manage / Connect on the Slack card. That path is distinct from tool_output_ready deliveries: follow Slack’s documentation for webhook URL rotation and message formatting.

---

Source of truth

Signing and outbound POSTs are implemented in webhook_service.py (generate_webhook_signature, send_webhook). Payload fields follow WebhookPayload in models/webhook.py. If behavior changes, update this page and the in-app webhook documentation together.