Webhooks¶

Webhooks enable asynchronous response delivery for long-running agent tasks.

Overview¶

When you include a webhook_url in your request:

1. API returns immediately with 202 Accepted
2. Agent processes the request asynchronously
3. Response is delivered to your webhook URL
4. Retries on failure (up to 4 attempts)

Setting Up Webhooks¶

Per-Request Webhook¶

Include webhook_url in your message request:

curl -X POST https://agents.ag2trust.com/api/v1/agents/{id}/messages \
  -H "X-API-Key: cust_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "message": "Analyze this document...",
    "webhook_url": "https://your-server.com/webhooks/ag2trust"
  }'

Default Webhook¶

Set a default webhook URL for all async responses:

curl -X PUT https://agents.ag2trust.com/api/v1/webhook \
  -H "X-API-Key: cust_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{"webhook_url": "https://your-server.com/webhooks/ag2trust"}'

Webhook Delivery¶

Request Format¶

AG2Trust sends a POST request to your webhook URL:

POST /webhooks/ag2trust HTTP/1.1
Host: your-server.com
Content-Type: application/json
X-Webhook-Signature: sha256=abc123...
X-Callback-ID: cb_xyz789

{
  "callback_id": "cb_xyz789",
  "message_id": "msg_abc123",
  "session_id": "session_456",
  "agent_id": "550e8400-e29b-41d4-a716-446655440000",
  "timestamp": "2025-01-15T10:30:15Z",
  "status": "completed",
  "content": [
    {
      "type": "text",
      "text": "The analysis shows..."
    }
  ],
  "metadata": {
    "tokens_used": 250,
    "model": "gpt-4o",
    "duration_ms": 5430
  }
}

Payload Fields¶

Signature Verification¶

Always verify webhook signatures to ensure authenticity.

Signature Header¶

X-Webhook-Signature: sha256=a1b2c3d4e5f6...

Verification Algorithm¶

1. Get the raw request body
2. Compute HMAC-SHA256 with your API key as secret
3. Compare with signature header (timing-safe)

Implementation Examples¶

import hmac
import hashlib

def verify_webhook(request_body: bytes, signature: str, api_key: str) -> bool:
    """Verify webhook signature."""
    expected = hmac.new(
        api_key.encode(),
        request_body,
        hashlib.sha256
    ).hexdigest()

    provided = signature.replace("sha256=", "")

    return hmac.compare_digest(expected, provided)

# Flask example
@app.route("/webhooks/ag2trust", methods=["POST"])
def handle_webhook():
    signature = request.headers.get("X-Webhook-Signature", "")
    if not verify_webhook(request.data, signature, API_KEY):
        return "Invalid signature", 401

    payload = request.json
    # Process the webhook...
    return "OK", 200

const crypto = require('crypto');

function verifyWebhook(body, signature, apiKey) {
  const expected = crypto
    .createHmac('sha256', apiKey)
    .update(body)
    .digest('hex');

  const provided = signature.replace('sha256=', '');

  return crypto.timingSafeEqual(
    Buffer.from(expected),
    Buffer.from(provided)
  );
}

// Express example
app.post('/webhooks/ag2trust', express.raw({type: '*/*'}), (req, res) => {
  const signature = req.headers['x-webhook-signature'] || '';

  if (!verifyWebhook(req.body, signature, API_KEY)) {
    return res.status(401).send('Invalid signature');
  }

  const payload = JSON.parse(req.body);
  // Process the webhook...
  res.status(200).send('OK');
});

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
)

func verifyWebhook(body []byte, signature, apiKey string) bool {
    mac := hmac.New(sha256.New, []byte(apiKey))
    mac.Write(body)
    expected := hex.EncodeToString(mac.Sum(nil))

    provided := strings.TrimPrefix(signature, "sha256=")

    return hmac.Equal([]byte(expected), []byte(provided))
}

Retry Logic¶

Failed deliveries are retried with exponential backoff:

Failure Conditions¶

• HTTP status 4xx or 5xx
• Connection timeout (10 seconds)
• DNS resolution failure
• SSL/TLS errors

After All Retries Fail¶

The callback is marked as failed in the database. You can query failed callbacks through the Dashboard.

Testing Webhooks¶

Test Endpoint¶

curl -X POST https://agents.ag2trust.com/api/v1/webhook/test \
  -H "X-API-Key: cust_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{"webhook_url": "https://your-server.com/webhooks/ag2trust"}'

Response:

{
  "success": true,
  "status_code": 200,
  "latency_ms": 150
}

Local Development¶

For local testing, use a tunneling service:

# Using ngrok
ngrok http 3000

# Use the ngrok URL as your webhook
webhook_url: https://abc123.ngrok.io/webhooks/ag2trust

Webhook Requirements¶

URL Requirements¶

Expected Response¶

Return 200 OK to acknowledge receipt:

HTTP/1.1 200 OK
Content-Type: text/plain

OK

Any 2xx response is considered successful.

Handling Webhooks¶

Idempotency¶

Webhooks may be delivered more than once. Implement idempotency:

processed_callbacks = set()  # In production, use Redis/database

@app.route("/webhooks/ag2trust", methods=["POST"])
def handle_webhook():
    payload = request.json
    callback_id = payload["callback_id"]

    # Check if already processed
    if callback_id in processed_callbacks:
        return "Already processed", 200

    # Process the webhook
    process_callback(payload)

    # Mark as processed
    processed_callbacks.add(callback_id)

    return "OK", 200

Async Processing¶

For complex handling, acknowledge quickly and process async:

from celery import Celery

celery = Celery('tasks')

@app.route("/webhooks/ag2trust", methods=["POST"])
def handle_webhook():
    # Verify signature first
    if not verify_signature(request):
        return "Invalid", 401

    # Queue for async processing
    process_webhook.delay(request.json)

    # Return immediately
    return "OK", 200

@celery.task
def process_webhook(payload):
    # Heavy processing here
    ...

Webhook Status Codes¶

Your endpoint should return:

Best Practices¶

1. Always Verify Signatures¶

# Always first
if not verify_webhook(request.data, signature, api_key):
    return "Unauthorized", 401

2. Respond Quickly¶

# Good: Acknowledge and process async
@app.route("/webhook")
def webhook():
    queue.add(request.json)
    return "OK", 200

# Bad: Process in request
@app.route("/webhook")
def webhook():
    heavy_processing(request.json)  # May timeout!
    return "OK", 200

3. Handle All Status Types¶

payload = request.json

if payload["status"] == "completed":
    handle_success(payload["content"])
elif payload["status"] == "failed":
    handle_failure(payload)

4. Log Callback IDs¶

import logging

@app.route("/webhook")
def webhook():
    payload = request.json
    logging.info(f"Received callback: {payload['callback_id']}")
    # Process...

5. Set Up Monitoring¶

Monitor your webhook endpoint for:

• Response times
• Error rates
• Retry patterns
• Processing delays

Next Steps¶

• Rate Limits - Understanding limits
• Error Codes - Error handling
• Webhook Integration Guide - Complete setup guide