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 request to receive the response asynchronously:

curl -X POST https://api.ag2trust.com/api/v1/agents/{agent_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"
  }'

The API returns 202 Accepted immediately with a callback_id. The full response is delivered to your webhook URL when processing completes.

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": 42,
  "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¶

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¶

• Pool Endpoint - Primary messaging endpoint
• Rate Limits - Understanding limits
• Error Codes - Error handling
• Webhook Integration Guide - Complete setup guide