Webhooks

Get real-time notifications when events occur in Mandatum.

Overview

Webhooks allow you to receive HTTP callbacks when specific events happen:

• Request completed
• Request failed
• High cost detected
• Error rate threshold exceeded
• New prompt version created
• Daily analytics summary

Creating Webhooks

Via Dashboard

1. Navigate to Settings → Webhooks
2. Click Create Webhook
3. Configure:
   • URL: Your endpoint URL
   • Events: Which events to subscribe to
   • Filters: Optional metadata filters
   • Secret: Webhook signing secret
4. Click Create

Via API

from mandatum import Mandatum

client = Mandatum(api_key="your-api-key")

webhook = client.webhooks.create(
    url="https://your-app.com/webhooks/mandatum",
    events=[
        "request.completed",
        "request.failed",
        "prompt.version_created"
    ],
    filters={
        "metadata.environment": "production"
    },
    secret="your-webhook-secret"
)

print(f"Webhook created: {webhook.id}")

Available Events

Request Events

Event	Description	Payload
request.completed	Request successfully completed	Request details
request.failed	Request failed with error	Error details
request.slow	Request exceeded latency threshold	Performance data
request.expensive	Request exceeded cost threshold	Cost data

Prompt Events

Event	Description	Payload
prompt.created	New prompt created	Prompt details
prompt.updated	Prompt updated (new version)	Version details
prompt.deleted	Prompt deleted	Deletion info
prompt.version_created	New version created	Version details

Analytics Events

Event	Description	Payload
analytics.daily_summary	Daily analytics summary	Daily metrics
analytics.weekly_summary	Weekly analytics summary	Weekly metrics
analytics.cost_alert	Cost threshold exceeded	Cost details
analytics.error_alert	Error rate threshold exceeded	Error details

Integration Events

Event	Description	Payload
integration.created	New integration created	Integration details
integration.updated	Integration updated	Update details
integration.error	Integration execution failed	Error details

Webhook Payload

Request Completed

{
  "event": "request.completed",
  "timestamp": "2025-01-15T10:30:00Z",
  "webhook_id": "webhook_abc123",
  "data": {
    "request_id": "req_xyz789",
    "prompt_name": "customer-classifier",
    "version": 3,
    "status": "success",
    "input_variables": {
      "message": "I need help with billing"
    },
    "output": {
      "category": "billing",
      "confidence": 0.95
    },
    "metadata": {
      "user_id": "user_123",
      "session_id": "session_456",
      "environment": "production"
    },
    "usage": {
      "prompt_tokens": 50,
      "completion_tokens": 10,
      "total_tokens": 60,
      "estimated_cost_usd": 0.0012
    },
    "latency_ms": 1234,
    "created_at": "2025-01-15T10:30:00Z"
  }
}

Request Failed

{
  "event": "request.failed",
  "timestamp": "2025-01-15T10:31:00Z",
  "webhook_id": "webhook_abc123",
  "data": {
    "request_id": "req_xyz790",
    "prompt_name": "customer-classifier",
    "version": 3,
    "status": "error",
    "error": {
      "type": "llm_error",
      "message": "API rate limit exceeded",
      "code": "RATE_LIMIT_EXCEEDED"
    },
    "input_variables": {
      "message": "test"
    },
    "metadata": {
      "environment": "production"
    },
    "created_at": "2025-01-15T10:31:00Z"
  }
}

Analytics Alert

{
  "event": "analytics.cost_alert",
  "timestamp": "2025-01-15T23:00:00Z",
  "webhook_id": "webhook_abc123",
  "data": {
    "alert_name": "Daily cost threshold",
    "threshold": 100.00,
    "current_value": 125.50,
    "period": "1d",
    "details": {
      "total_requests": 15234,
      "total_cost_usd": 125.50,
      "top_prompts": [
        {
          "name": "classifier",
          "cost_usd": 45.30,
          "requests": 5432
        }
      ]
    }
  }
}

Receiving Webhooks

Basic Handler

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

app = Flask(__name__)
WEBHOOK_SECRET = "your-webhook-secret"

@app.route('/webhooks/mandatum', methods=['POST'])
def mandatum_webhook():
    # Verify signature
    signature = request.headers.get('X-Mandatum-Signature')
    if not verify_signature(request.data, signature):
        return jsonify({'error': 'Invalid signature'}), 401

    # Parse payload
    payload = request.json
    event_type = payload['event']

    # Handle event
    if event_type == 'request.completed':
        handle_request_completed(payload['data'])
    elif event_type == 'request.failed':
        handle_request_failed(payload['data'])
    elif event_type == 'analytics.cost_alert':
        handle_cost_alert(payload['data'])

    return jsonify({'received': True}), 200

def verify_signature(payload: bytes, signature: str) -> bool:
    """Verify webhook signature."""
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected)

def handle_request_completed(data: dict):
    """Handle completed request."""
    print(f"Request {data['request_id']} completed")

def handle_request_failed(data: dict):
    """Handle failed request."""
    print(f"Request {data['request_id']} failed: {data['error']['message']}")

def handle_cost_alert(data: dict):
    """Handle cost alert."""
    print(f"Cost alert: ${data['current_value']:.2f} exceeds ${data['threshold']:.2f}")

TypeScript Handler

import express from 'express';
import crypto from 'crypto';

const app = express();
const WEBHOOK_SECRET = 'your-webhook-secret';

app.post('/webhooks/mandatum', express.json(), (req, res) => {
  // Verify signature
  const signature = req.headers['x-mandatum-signature'] as string;
  if (!verifySignature(JSON.stringify(req.body), signature)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  // Parse payload
  const payload = req.body;
  const eventType = payload.event;

  // Handle event
  switch (eventType) {
    case 'request.completed':
      handleRequestCompleted(payload.data);
      break;
    case 'request.failed':
      handleRequestFailed(payload.data);
      break;
    case 'analytics.cost_alert':
      handleCostAlert(payload.data);
      break;
  }

  res.json({ received: true });
});

function verifySignature(payload: string, signature: string): boolean {
  const expected = crypto
    .createHmac('sha256', WEBHOOK_SECRET)
    .update(payload)
    .digest('hex');
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expected)
  );
}

function handleRequestCompleted(data: any): void {
  console.log(`Request ${data.request_id} completed`);
}

function handleRequestFailed(data: any): void {
  console.log(`Request ${data.request_id} failed: ${data.error.message}`);
}

function handleCostAlert(data: any): void {
  console.log(`Cost alert: $${data.current_value} exceeds $${data.threshold}`);
}

Security

Signature Verification

Always verify webhook signatures:

import hmac
import hashlib

def verify_webhook_signature(
    payload: bytes,
    signature: str,
    secret: str
) -> bool:
    """
    Verify webhook signature.

    Args:
        payload: Raw request body
        signature: X-Mandatum-Signature header value
        secret: Your webhook secret

    Returns:
        True if signature is valid
    """
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(signature, expected)

IP Whitelisting

Restrict webhooks to Mandatum IP addresses:

MANDATUM_IPS = [
    '35.123.45.67',
    '35.123.45.68',
    '35.123.45.69'
]

@app.route('/webhooks/mandatum', methods=['POST'])
def mandatum_webhook():
    # Check IP address
    client_ip = request.remote_addr
    if client_ip not in MANDATUM_IPS:
        return jsonify({'error': 'Forbidden'}), 403

    # Process webhook
    # ...

HTTPS Only

Always use HTTPS for webhook URLs:

# Good
webhook = client.webhooks.create(
    url="https://your-app.com/webhooks"
)

# Bad (will be rejected)
webhook = client.webhooks.create(
    url="http://your-app.com/webhooks"
)

Filtering

By Metadata

Only receive webhooks for specific metadata:

# Only production requests
webhook = client.webhooks.create(
    url="https://your-app.com/webhooks",
    events=["request.completed"],
    filters={
        "metadata.environment": "production"
    }
)

# Only specific user
webhook = client.webhooks.create(
    url="https://your-app.com/webhooks",
    events=["request.completed"],
    filters={
        "metadata.user_id": "user_123"
    }
)

By Prompt

Only receive webhooks for specific prompts:

webhook = client.webhooks.create(
    url="https://your-app.com/webhooks",
    events=["request.completed"],
    filters={
        "prompt_name": "customer-classifier"
    }
)

Retry Policy

Mandatum automatically retries failed webhook deliveries:

Attempt	Delay
1	Immediate
2	1 minute
3	5 minutes
4	15 minutes
5	1 hour

After 5 failed attempts, the webhook delivery is marked as failed.

Handling Retries

Return 200 OK quickly to avoid retries:

@app.route('/webhooks/mandatum', methods=['POST'])
def mandatum_webhook():
    # Verify and parse
    payload = request.json

    # Queue for async processing
    queue.enqueue(process_webhook, payload)

    # Return immediately
    return jsonify({'received': True}), 200

def process_webhook(payload: dict):
    """Process webhook asynchronously."""
    # Heavy processing here
    # ...

Monitoring

Webhook Logs

View webhook delivery logs:

# Get webhook logs
logs = client.webhooks.get_logs(
    webhook_id="webhook_abc123",
    limit=50
)

for log in logs:
    print(f"{log.timestamp}: {log.event} - {log.status}")
    if log.status == 'failed':
        print(f"  Error: {log.error}")

Webhook Health

Check webhook health:

# Get webhook health
health = client.webhooks.get_health(
    webhook_id="webhook_abc123"
)

print(f"Success rate: {health.success_rate:.1%}")
print(f"Average latency: {health.avg_latency_ms}ms")
print(f"Failed deliveries (24h): {health.failed_count_24h}")

Common Use Cases

Slack Notifications

Send alerts to Slack:

def handle_request_failed(data: dict):
    """Send failed request alert to Slack."""
    import requests

    slack_webhook = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"

    message = {
        "text": f"🚨 Request failed: {data['prompt_name']}",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Request ID:* {data['request_id']}\n*Error:* {data['error']['message']}"
                }
            }
        ]
    }

    requests.post(slack_webhook, json=message)

Custom Analytics

Send data to your analytics platform:

def handle_request_completed(data: dict):
    """Track request in analytics."""
    import segment

    segment.track(
        user_id=data['metadata'].get('user_id'),
        event='LLM Request Completed',
        properties={
            'prompt_name': data['prompt_name'],
            'version': data['version'],
            'latency_ms': data['latency_ms'],
            'cost_usd': data['usage']['estimated_cost_usd'],
            'environment': data['metadata'].get('environment')
        }
    )

Incident Management

Create incidents on failures:

def handle_request_failed(data: dict):
    """Create PagerDuty incident."""
    from pdpyras import EventsAPISession

    session = EventsAPISession('YOUR_INTEGRATION_KEY')

    session.trigger(
        summary=f"Mandatum request failed: {data['prompt_name']}",
        source='mandatum',
        severity='error',
        custom_details={
            'request_id': data['request_id'],
            'error': data['error']['message'],
            'prompt_name': data['prompt_name']
        }
    )

Testing

Test Webhook

Send a test event to your webhook:

# Send test event
client.webhooks.send_test(
    webhook_id="webhook_abc123",
    event="request.completed"
)

Local Testing

Use tools like ngrok for local testing:

# Start ngrok
ngrok http 3000

# Use ngrok URL for webhook
https://abc123.ngrok.io/webhooks/mandatum

Next Steps

• Learn about metadata
• Explore custom code
• View API reference