Webhooks

from tork_governance import TorkClient

client = TorkClient()

# Create a webhook endpoint
webhook = client.webhooks.create(
    url="https://your-server.com/webhooks/tork",
    events=[
        "policy.violation",
        "approval.requested",
        "approval.decided",
        "budget.threshold_reached",
        "circuit_breaker.triggered"
    ],
    secret="whsec_your_signing_secret"  # For signature verification
)

print(f"Webhook ID: {webhook.id}")
print(f"Signing secret: {webhook.secret}")

{
  "id": "evt_abc123def456",
  "type": "policy.violation",
  "created_at": "2024-01-15T10:30:00Z",
  "data": {
    "agent_id": "agent-123",
    "policy_name": "block-pii-sharing",
    "action": "BLOCK",
    "reason": "PII detected in output",
    "content_preview": "The user's SSN is...",
    "metadata": {
      "user_id": "user_456",
      "session_id": "sess_789"
    }
  },
  "environment": "production"
}

import hmac
import hashlib
from flask import Flask, request

app = Flask(__name__)
WEBHOOK_SECRET = "whsec_your_signing_secret"

@app.route("/webhooks/tork", methods=["POST"])
def handle_webhook():
    # Get the signature from headers
    signature = request.headers.get("X-Tork-Signature")
    timestamp = request.headers.get("X-Tork-Timestamp")

    # Compute expected signature
    payload = f"{timestamp}.{request.data.decode()}"
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()

    # Verify signature
    if not hmac.compare_digest(signature, f"sha256={expected}"):
        return "Invalid signature", 401

    # Process the event
    event = request.json
    print(f"Received event: {event['type']}")

    return "OK", 200