Webhooks are available on Pro, Business, and Enterprise plans. Upgrade your plan to unlock this feature.

Overview

Webhooks allow you to receive real-time HTTP notifications when important events occur in AgentWarden. Instead of polling for changes, AgentWarden will send a POST request to your specified URL whenever an event happens.

Available Events

AgentWarden supports the following webhook events:
EventDescription
log.blockedTriggered when an action is blocked by permissions
log.pending_approvalTriggered when an action requires human approval
log.failedTriggered when an action fails during execution
approval.approvedTriggered when a pending approval is approved
approval.rejectedTriggered when a pending approval is rejected

Configuration

1. Set Up Your Webhook Endpoint

Create an HTTP endpoint in your application that can receive POST requests:
from flask import Flask, request
import hmac
import hashlib

app = Flask(__name__)

@app.route('/webhook/agentwarden', methods=['POST'])
def handle_webhook():
    payload = request.get_json()
    
    event = payload['event']
    data = payload['data']
    timestamp = payload['timestamp']
    
    # Handle different event types
    if event == 'approval.approved':
        approval_id = data['approval_id']
        action = data['action']
        context = data['context']
        
        # Execute the approved action
        execute_approved_action(action, context)
        
    elif event == 'approval.rejected':
        # Handle rejected approval
        log_rejected_action(data)
        
    elif event == 'log.blocked':
        # Alert on blocked actions
        send_alert(data)
    
    return {'status': 'received'}, 200

def execute_approved_action(action, context):
    """Execute the action that was approved"""
    if action == "stripe.refund":
        # Process the refund
        amount = context.get('amount')
        process_refund(amount)
    # ... handle other actions

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

2. Configure Webhook in Dashboard

  1. Go to SettingsWebhooks in your AgentWarden dashboard
  2. Enter your webhook URL: https://your-domain.com/webhook/agentwarden
  3. Select the events you want to subscribe to
  4. Click Save Configuration
  5. Click Test Webhook to verify it’s working

3. Make Your Endpoint Public

Your webhook endpoint must be publicly accessible. Options include:
  • Deploy to a cloud provider (AWS, Google Cloud, Heroku, etc.)
  • Use ngrok for local development: ngrok http 5000
  • Use a serverless function (AWS Lambda, Google Cloud Functions, etc.)

Webhook Payload

All webhooks follow this structure:
{
  "event": "approval.approved",
  "timestamp": "2026-02-25T17:30:00Z",
  "data": {
    // Event-specific data
  }
}

Event Payloads

{
  "event": "log.pending_approval",
  "timestamp": "2026-02-25T23:51:33.984319",
  "data": {
    "id": "a1b2c3d4-e5f6-7890-ab12-cd34ef567890",
    "agent_id": "agent_abc123def456ghi789",
    "action": "stripe.refund",
    "status": "pending_approval",
    "context": {
      "amount": 500,
      "customer_id": "cus_123",
      "reason": "Awaiting human review"
    },
    "created_at": "2026-02-25T23:51:33.974127+00:00"
  }
}
{
  "event": "approval.approved",
  "timestamp": "2026-02-25T23:51:57.042729",
  "data": {
    "approval_id": "a1b2c3d4-e5f6-7890-ab12-cd34ef567890",
    "agent_id": "agent_abc123def456ghi789",
    "action": "stripe.refund",
    "status": "approved",
    "context": {
      "amount": 500,
      "customer_id": "cus_123"
    },
    "resolved_by": "user_xyz789abc456def123",
    "resolved_at": "2026-02-25T23:51:57.036178+00:00"
  }
}
{
  "event": "approval.rejected",
  "timestamp": "2026-02-25T23:52:30.123456",
  "data": {
    "approval_id": "a1b2c3d4-e5f6-7890-ab12-cd34ef567890",
    "agent_id": "agent_abc123def456ghi789",
    "action": "stripe.refund",
    "status": "rejected",
    "context": {
      "amount": 500
    },
    "resolved_by": "user_xyz789abc456def123",
    "resolved_at": "2026-02-25T23:52:30.118765+00:00"
  }
}

Best Practices

1. Respond Quickly

Your webhook endpoint should respond with a 200 OK status within 5 seconds. Process time-consuming tasks asynchronously:
from flask import Flask
import threading

@app.route('/webhook/agentwarden', methods=['POST'])
def handle_webhook():
    payload = request.get_json()
    
    # Respond immediately
    # Process in background
    thread = threading.Thread(target=process_webhook, args=(payload,))
    thread.start()
    
    return {'status': 'received'}, 200

def process_webhook(payload):
    # Time-consuming processing here
    pass

2. Handle Retries

AgentWarden will retry failed webhook deliveries up to 3 times with exponential backoff. Make your endpoint idempotent:
processed_events = set()

def handle_webhook():
    payload = request.get_json()
    event_id = payload.get('data', {}).get('approval_id')
    
    # Skip if already processed
    if event_id in processed_events:
        return {'status': 'already_processed'}, 200
    
    # Process event
    process_event(payload)
    processed_events.add(event_id)
    
    return {'status': 'received'}, 200

3. Validate Webhook Source

Verify requests are actually from AgentWarden by checking the User-Agent header:
@app.route('/webhook/agentwarden', methods=['POST'])
def handle_webhook():
    user_agent = request.headers.get('User-Agent')
    
    if user_agent != 'AgentWarden-Webhook/1.0':
        return {'error': 'Invalid source'}, 403
    
    # Process webhook
    ...

4. Log Webhook Events

Keep a log of webhook events for debugging:
import logging

logger = logging.getLogger(__name__)

@app.route('/webhook/agentwarden', methods=['POST'])
def handle_webhook():
    payload = request.get_json()
    
    logger.info(f"Received webhook: {payload['event']}")
    logger.debug(f"Webhook data: {payload}")
    
    # Process event
    ...

Troubleshooting

Webhook Not Receiving Events

  1. Check URL accessibility: Ensure your endpoint is publicly accessible
  2. Verify event subscription: Confirm you’ve subscribed to the event in Settings
  3. Check endpoint logs: Look for incoming requests and errors
  4. Test the webhook: Use the “Test Webhook” button in dashboard

Webhook Timing Out

  1. Respond within 5 seconds
  2. Move long-running tasks to background workers
  3. Use async processing (queues, threads, etc.)

Duplicate Events

  1. Implement idempotency using event IDs
  2. Store processed event IDs in database or cache
  3. Return 200 OK for duplicate events

Example: Complete Approval Workflow

Here’s a complete example of handling approval workflows with webhooks:
from flask import Flask, request
from agentwarden import AgentWarden
import os

app = Flask(__name__)
guard = AgentWarden(api_key=os.getenv('AGENTWARDEN_API_KEY'))

@app.route('/webhook/agentwarden', methods=['POST'])
def handle_webhook():
    payload = request.get_json()
    event = payload['event']
    
    if event == 'approval.approved':
        data = payload['data']
        action = data['action']
        context = data['context']
        agent_id = data['agent_id']
        
        # Execute the approved action
        try:
            if action == 'stripe.refund':
                result = process_stripe_refund(context)
                
                # Log success
                guard.log(agent_id, action, 'success', {
                    **context,
                    'result': result
                })
                
        except Exception as e:
            # Log failure
            guard.log(agent_id, action, 'failed', {
                **context,
                'error': str(e)
            })
    
    return {'status': 'received'}, 200

def process_stripe_refund(context):
    import stripe
    stripe.api_key = os.getenv('STRIPE_API_KEY')
    
    refund = stripe.Refund.create(
        charge=context['charge_id'],
        amount=context['amount']
    )
    
    return {'refund_id': refund.id}

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

Next Steps