​
Global Webhooks

Global webhooks in HITL.sh provide real-time notifications about all system events across your entire account. Unlike loop-specific webhooks, global webhooks capture events from all loops, making them ideal for centralized monitoring, analytics, and integration systems.

​
What are Global Webhooks?

Global webhooks are HTTP endpoints that receive notifications about:

Global Webhook Events

  • Account-wide Events: User management, billing, and system updates
  • All Loop Events: Requests, responses, and loop status changes
  • System Events: Performance metrics, errors, and maintenance updates
  • Integration Events: API usage, rate limiting, and authentication events

​
When to Use Global Webhooks

​
Use Cases

Centralized Monitoring

Monitor all HITL.sh activity from a single endpoint for comprehensive oversight.

Analytics Systems

Collect data from all loops for unified analytics and reporting.

Audit Logging

Maintain comprehensive audit trails for compliance and security purposes.

Integration Hubs

Route events to multiple downstream systems and services.

​
Comparison with Loop-Specific Webhooks

​
Setting Up Global Webhooks

​
1. Create Webhook Endpoint

First, create an endpoint in your system to receive webhook notifications: 
from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret"

@app.route('/webhooks/hitl/global', methods=['POST'])
def handle_global_webhook():
    # Verify webhook signature
    signature = request.headers.get('X-HITL-Signature')
    if not verify_signature(request.data, signature):
        return jsonify({"error": "Invalid signature"}), 401
    
    payload = request.json
    event_type = payload["event"]
    
    # Route to appropriate handler based on event type
    if event_type.startswith("request."):
        handle_request_event(payload)
    elif event_type.startswith("loop."):
        handle_loop_event(payload)
    elif event_type.startswith("user."):
        handle_user_event(payload)
    elif event_type.startswith("system."):
        handle_system_event(payload)
    
    return jsonify({"status": "success"}), 200

def verify_signature(payload, signature):
    expected_signature = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected_signature, signature)

​
2. Configure Webhook in HITL.sh

Set up the global webhook through the API: 
import requests

def create_global_webhook(webhook_url, events=None):
    if events is None:
        events = ["*"]  # Receive all events
    
    webhook_data = {
        "url": webhook_url,
        "events": events,
        "type": "global",  # Specify global webhook
        "secret": "your_webhook_secret",
        "enabled": True
    }
    
    response = requests.post(
        "https://api.hitl.sh/v1/webhooks",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        },
        json=webhook_data
    )
    
    return response.json()

​
3. Test Webhook Configuration

Verify your webhook is working correctly: 
def test_webhook(webhook_id):
    # Send a test event to verify webhook delivery
    response = requests.post(
        f"https://api.hitl.sh/v1/webhooks/{webhook_id}/test",
        headers={"Authorization": f"Bearer {API_KEY}"}
    )
    
    if response.status_code == 200:
        print("Webhook test successful")
        return True
    else:
        print(f"Webhook test failed: {response.text}")
        return False

​
Event Types and Payloads

​
Request Events

Events related to request lifecycle:

​
Loop Events

Events related to loop management:

​
User Events

Events related to user management:

​
System Events

Events related to system operations:

​
Event Filtering

​
Selective Event Subscription

Subscribe to specific event types to reduce noise: 
def create_filtered_webhook(webhook_url):
    # Only receive request-related events
    events = [
        "request.created",
        "request.completed",
        "request.escalated"
    ]
    
    webhook_data = {
        "url": webhook_url,
        "events": events,
        "type": "global",
        "secret": "your_webhook_secret",
        "enabled": True
    }
    
    response = requests.post(
        "https://api.hitl.sh/v1/webhooks",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        },
        json=webhook_data
    )
    
    return response.json()

​
Event Pattern Matching

Use wildcards for flexible event filtering: 
# Receive all request events
events = ["request.*"]

# Receive all events from specific loops
events = ["loop.xyz789.*"]

# Receive all events except system events
events = ["*", "!system.*"]

​
Webhook Processing

​
Event Routing

Route different event types to appropriate handlers: 
def route_webhook_event(payload):
    event_type = payload["event"]
    
    # Route based on event category
    if event_type.startswith("request."):
        route_request_event(payload)
    elif event_type.startswith("loop."):
        route_loop_event(payload)
    elif event_type.startswith("user."):
        route_user_event(payload)
    elif event_type.startswith("system."):
        route_system_event(payload)
    else:
        handle_unknown_event(payload)

def route_request_event(payload):
    event_type = payload["event"]
    
    if event_type == "request.created":
        handle_new_request(payload["data"])
    elif event_type == "request.completed":
        handle_completed_request(payload["data"])
    elif event_type == "request.escalated":
        handle_escalated_request(payload["data"])

​
Data Transformation

Transform webhook data for your systems: 
def transform_request_data(webhook_data):
    """Transform webhook data to internal format"""
    return {
        "id": webhook_data["request_id"],
        "loop_id": webhook_data["loop_id"],
        "status": webhook_data.get("status", "unknown"),
        "created_at": webhook_data["created_at"],
        "priority": webhook_data.get("priority", "normal"),
        "content_type": webhook_data.get("content_type", "unknown")
    }

def transform_loop_data(webhook_data):
    """Transform loop event data"""
    return {
        "id": webhook_data["loop_id"],
        "name": webhook_data.get("name", ""),
        "status": webhook_data.get("status", "unknown"),
        "event_type": webhook_data["event"],
        "timestamp": webhook_data["timestamp"]
    }

​
Error Handling and Retries

​
Webhook Delivery Failures

Handle webhook delivery failures gracefully:

Retry Logic

  • Implement exponential backoff
  • Set maximum retry attempts
  • Log failed deliveries for investigation
  • Alert on persistent failures

Fallback Mechanisms

  • Store failed events for later processing
  • Implement alternative notification methods
  • Use polling as backup for critical events
  • Maintain event ordering when possible

​
Dead Letter Queues

Implement dead letter queues for failed webhooks: 
import redis
import json

class DeadLetterQueue:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.queue_key = "hitl_webhook_dlq"
    
    def add_failed_event(self, event_data, error_reason):
        """Add failed webhook event to dead letter queue"""
        dlq_entry = {
            "event_data": event_data,
            "error_reason": error_reason,
            "failed_at": datetime.utcnow().isoformat(),
            "retry_count": 0
        }
        
        self.redis.lpush(self.queue_key, json.dumps(dlq_entry))
    
    def process_dlq(self):
        """Process events from dead letter queue"""
        while True:
            entry = self.redis.brpop(self.queue_key, timeout=1)
            if entry:
                dlq_entry = json.loads(entry[1])
                self.retry_event(dlq_entry)
    
    def retry_event(self, dlq_entry):
        """Retry processing a failed event"""
        try:
            # Attempt to process the event again
            process_webhook_event(dlq_entry["event_data"])
        except Exception as e:
            # Increment retry count
            dlq_entry["retry_count"] += 1
            
            if dlq_entry["retry_count"] < 3:
                # Re-add to queue for another attempt
                self.add_failed_event(
                    dlq_entry["event_data"],
                    str(e)
                )
            else:
                # Log final failure
                log_final_failure(dlq_entry)

​
Security and Authentication

​
Webhook Signature Verification

Always verify webhook signatures: 
def verify_webhook_signature(payload, signature, secret):
    """Verify webhook signature using HMAC-SHA256"""
    expected_signature = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected_signature, signature)

def secure_webhook_handler(request):
    """Secure webhook handler with signature verification"""
    # Get signature from headers
    signature = request.headers.get('X-HITL-Signature')
    if not signature:
        return jsonify({"error": "Missing signature"}), 401
    
    # Verify signature
    if not verify_webhook_signature(
        request.data, 
        signature, 
        WEBHOOK_SECRET
    ):
        return jsonify({"error": "Invalid signature"}), 401
    
    # Process webhook
    return process_webhook(request.json)

​
Rate Limiting

Implement rate limiting to prevent abuse: 
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    app,
    key_func=get_remote_address,
    default_limits=["100 per minute"]
)

@app.route('/webhooks/hitl/global', methods=['POST'])
@limiter.limit("100 per minute")
def handle_global_webhook():
    # Webhook processing logic
    pass

​
Monitoring and Analytics

​
Webhook Performance Metrics

Track webhook delivery performance:

Delivery Metrics

  • Success rate percentage
  • Average delivery time
  • Failure rate by event type
  • Retry success rates

System Metrics

  • Webhook queue length
  • Processing latency
  • Error frequency
  • Resource utilization

​
Alerting and Notifications

Set up alerts for webhook issues:
1

Failure Alerts

Alert when webhook delivery fails repeatedly.
2

Performance Alerts

Alert when webhook processing times exceed thresholds.
3

Volume Alerts

Alert when webhook volume is unusually high or low.

​
Best Practices

​
Webhook Design

Idempotency

  • Handle duplicate events gracefully
  • Use event IDs for deduplication
  • Implement idempotent processing logic

Event Ordering

  • Maintain event order when possible
  • Use timestamps for sequencing
  • Handle out-of-order events gracefully

​
Performance Optimization

1

Async Processing

Process webhooks asynchronously to avoid blocking.
2

Batch Processing

Group related events for efficient processing.
3

Connection Pooling

Reuse HTTP connections for better performance.

​
Troubleshooting

​
Common Issues

​
Next Steps

Ready to set up global webhooks for your HITL.sh account?

Loop-Based Webhooks

Learn about loop-specific webhook configurations.

API Reference

Detailed API documentation for webhook management.

Integration Examples

See how to use webhooks with popular integration platforms.