Webhooks

Real-time notifications for task state transitions

Overview

Webhooks enable real-time push notifications when tasks complete, fail, or trigger refunds. Instead of polling for task status, receive instant HTTP callbacks.

Webhooks are optional. You can still use polling via GET /v3/tasks/{task_id}.

Setup

Configure webhook URL per-task or set a global webhook for your agent:

Per-Task Webhook

curl -X POST https://api.vapagent.com/v3/tasks \
  -H "Authorization: Bearer vape_xxx" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "image_generation",
    "params": {
      "description": "A mountain landscape"
    },
    "webhook_url": "https://your-server.com/webhook"
  }'

Global Webhook (Future)

Coming soon: Set default webhook URL in agent settings.

Event Types

VAP sends webhooks for the following events:

Event Trigger Description
task.completed Task finished successfully Image/video generated, balance burned
task.failed Task failed after all retries Generation failed, balance refunded
task.refunded Balance refunded after failure Reserved amount returned to balance

Payload Schema

All webhooks follow this structure:

task.completed

{
  "event": "task.completed",
  "timestamp": "2026-01-09T12:00:00Z",
  "data": {
    "task_id": "550e8400-e29b-41d4-a716-446655440000",
    "status": "completed",
    "result": {
      "image_url": "https://replicate.delivery/xezq/example.webp"
    },
    "estimated_cost": "0.18",
    "actual_cost": "0.0527",
    "duration_seconds": 45,
    "created_at": "2026-01-09T11:59:15Z",
    "completed_at": "2026-01-09T12:00:00Z"
  }
}

task.failed

{
  "event": "task.failed",
  "timestamp": "2026-01-09T12:00:00Z",
  "data": {
    "task_id": "550e8400-e29b-41d4-a716-446655440000",
    "status": "failed",
    "error": {
      "code": "provider_error",
      "message": "Image generation failed"
    },
    "estimated_cost": "0.18",
    "refund_amount": "0.18",
    "attempts": 3,
    "created_at": "2026-01-09T11:59:15Z",
    "failed_at": "2026-01-09T12:00:00Z"
  }
}

task.refunded

{
  "event": "task.refunded",
  "timestamp": "2026-01-09T12:00:01Z",
  "data": {
    "task_id": "550e8400-e29b-41d4-a716-446655440000",
    "refund_amount": "0.18",
    "new_balance": "4.91"
  }
}

Signature Verification

All webhooks include a signature in the X-VAP-Signature header to verify authenticity:

POST /webhook HTTP/1.1
Host: your-server.com
Content-Type: application/json
X-VAP-Signature: sha256=a1b2c3d4e5f6...
X-VAPvent: task.completed

{...webhook payload...}

Verification (Python)

import hmac
import hashlib

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

    Args:
        payload: Raw request body (bytes)
        signature: Value from X-VAP-Signature header
        secret: Your API key (used as HMAC secret)

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

    return hmac.compare_digest(f"sha256={expected}", signature)

# Usage in Flask
@app.route('/webhook', methods=['POST'])
def webhook():
    signature = request.headers.get('X-VAP-Signature')
    payload = request.get_data()
    secret = os.environ['VAPE_API_KEY']

    if not verify_webhook_signature(payload, signature, secret):
        return 'Invalid signature', 401

    event = request.json
    handle_event(event)
    return '', 200

Verification (Node.js)

const crypto = require('crypto');

function verifyWebhookSignature(payload, signature, secret) {
  const expected = crypto
    .createHmac('sha256', secret)
    .update(payload)
    .digest('hex');

  return crypto.timingSafeEqual(
    Buffer.from(`sha256=${expected}`),
    Buffer.from(signature)
  );
}

// Usage in Express
app.post('/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.headers['x-vap-signature'];
  const payload = req.body;
  const secret = process.env.VAPE_API_KEY;

  if (!verifyWebhookSignature(payload, signature, secret)) {
    return res.status(401).send('Invalid signature');
  }

  const event = JSON.parse(payload);
  handleEvent(event);
  res.status(200).send();
});

Security: Always verify webhook signatures to prevent malicious requests.

Retry Policy

If your webhook endpoint returns non-2xx or times out, VAP retries with exponential backoff:

Attempt Delay Cumulative Time
1 Immediate 0s
2 30 seconds 30s
3 2 minutes 2m 30s
4 10 minutes 12m 30s
5 1 hour 1h 12m 30s

After 5 failed attempts, the webhook is marked as failed and no further retries occur.

Response Requirements

Your webhook endpoint must:

  • Return HTTP 2xx status code (200, 201, 202, 204)
  • Respond within 30 seconds
  • Handle duplicate deliveries (webhooks may be sent multiple times)

Example Endpoint

from flask import Flask, request

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    # 1. Verify signature
    signature = request.headers.get('X-VAP-Signature')
    if not verify_webhook_signature(request.get_data(), signature, SECRET):
        return 'Invalid signature', 401

    # 2. Parse event
    event = request.json
    event_type = event['event']
    task_id = event['data']['task_id']

    # 3. Process event (idempotent!)
    if event_type == 'task.completed':
        image_url = event['data']['result']['image_url']
        process_completed_task(task_id, image_url)
    elif event_type == 'task.failed':
        handle_failed_task(task_id)

    # 4. Return 200 quickly
    return '', 200

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

Best Practices

1. Process Asynchronously

Return 200 immediately, then process the webhook in a background job:

@app.route('/webhook', methods=['POST'])
def webhook():
    # Verify signature
    if not verify_webhook_signature(...):
        return 'Invalid', 401

    # Queue for background processing
    event = request.json
    queue.enqueue(process_webhook, event)

    # Return immediately
    return '', 200

2. Handle Duplicates

Use task_id to deduplicate:

def process_webhook(event):
    task_id = event['data']['task_id']

    # Check if already processed
    if redis.exists(f'processed:{task_id}'):
        return

    # Process event
    handle_event(event)

    # Mark as processed
    redis.setex(f'processed:{task_id}', 86400, '1')  # 24h TTL

3. Log All Events

Store webhook events for debugging and audit:

@app.route('/webhook', methods=['POST'])
def webhook():
    # Log raw event
    db.webhooks.insert({
        'received_at': datetime.now(),
        'event_type': request.json['event'],
        'payload': request.json,
        'signature': request.headers.get('X-VAP-Signature')
    })

    # Process...
    return '', 200

Testing Webhooks

Local Testing with ngrok

  1. Install ngrok: npm install -g ngrok
  2. Start local server: python app.py (port 5000)
  3. Expose via ngrok: ngrok http 5000
  4. Use ngrok URL in webhook_url: https://abc123.ngrok.io/webhook

Mock Webhook Event

Test your handler with a manual POST:

curl -X POST http://localhost:5000/webhook \
  -H "Content-Type: application/json" \
  -H "X-VAP-Signature: sha256=test" \
  -d '{
    "event": "task.completed",
    "timestamp": "2026-01-09T12:00:00Z",
    "data": {
      "task_id": "test-task-id",
      "status": "completed",
      "result": {
        "image_url": "https://example.com/test.jpg"
      }
    }
  }'

Need Help?

If webhooks aren't working: