DEV Community: Vincent

How to Build a Resilient Webhook Relay for AI Agents

Vincent — Wed, 17 Jun 2026 16:07:16 +0000

The Problem

AI agents don't have email inboxes. They don't poll APIs every few seconds (they shouldn't). And yet, the world still communicates via webhooks, callbacks, and event streams. When you're building an AI agent that needs to react to real-world events — form submissions, payment confirmations, data ingestion pipelines — you need infrastructure that handles these incoming signals reliably.

That's what AgentForms was built to solve: form infrastructure for AI agents. But the hardest part wasn't the forms themselves — it was the webhook relay that sits between the world and your agent.

Why a Relay?

A naive approach to webhook handling looks like this:

Third-party API → Your Agent's Endpoint → Process

This has problems:

Your agent isn't always online. Containers restart. Deployments happen.
Third-party retries are unpredictable. Stripe retries with exponential backoff. GitHub retries 5 times. Some services don't retry at all.
One slow handler blocks everything. If processing a webhook takes 12 seconds, your endpoint is tied up for 12 seconds. Most webhook senders timeout after 5–10.
No delivery guarantees. If your agent crashes mid-processing, the event is lost.

The relay pattern solves all of these by decoupling receipt from processing:

Third-party API → Relay (ack immediately) → Queue → Worker (process at own pace)

The Relay Architecture

AgentForms implements this as two Docker containers: the relay and the worker. They talk over Redis.

The Relay: Fast Ack, Fast Forward

The relay's job is simple: accept the webhook, validate it, store it, and respond with 200 — ideally in under 100ms.

Here's the core relay handler in Python with Flask:

import hmac
import hashlib
import json
import redis
import time
from flask import Flask, request, jsonify

app = Flask(__name__)
redis_client = redis.Redis(host='redis', port=6379)

MAX_RETRIES = 5
RETRY_BACKOFF = [1, 5, 15, 60, 300]  # seconds

@app.route('/webhook/<service>', methods=['POST'])
def handle_webhook(service: str):
    """Accept webhook, validate signature, enqueue, respond 200."""

    # 1. Validate the signature (prevents spoofed webhooks)
    if not validate_signature(request, service):
        return jsonify({'error': 'invalid signature'}), 401

    # 2. Build the envelope
    envelope = {{
        'id': generate_uuid(),
        'service': service,
        'payload': request.get_data(),
        'headers': dict(request.headers),
        'received_at': time.time(),
        'retries': 0,
        'status': 'pending',
    }}

    # 3. Store with a retry key
    redis_client.setex(
        f'webhook:{{envelope["id"]}}',
        86400,
        json.dumps(envelope)
    )

    # 4. Push to the worker queue
    redis_client.lpush('webhook:queue', envelope['id'])

    # 5. Respond immediately
    return jsonify({{'status': 'queued'}}), 200

def validate_signature(request, service):
    """Service-specific signature validation."""
    if service == 'stripe':
        return validate_stripe_signature(request)
    elif service == 'github':
        return validate_github_signature(request)
    return False

The key design decisions here:

200 before processing. The sender gets a success response before we've touched the payload. The relay is fast because it does almost nothing.
Per-service validation. Different services use different signing mechanisms. Stripe sends a t= timestamp with its signature. GitHub sends a SHA256 HMAC.
24-hour TTL. We don't hold webhooks indefinitely. If a worker can't process one in 24 hours, something is wrong and we need alerting.

The Worker: Reliable Processing with Retries

The worker pulls from the queue and processes events. If processing fails, it retries with exponential backoff.

import redis
import json
import time
import traceback

redis_client = redis.Redis(host='redis', port=6379)
MAX_RETRIES = 5
RETRY_BACKOFF = [1, 5, 15, 60, 300]  # seconds

def process_webhook(envelope):
    """Process a single webhook event."""
    service = envelope['service']
    payload = json.loads(envelope['payload'])

    handler = ROUTERS.get(service)
    if not handler:
        print(f'No handler for service: {{service}}')
        return False

    try:
        handler(payload)
        return True
    except Exception as e:
        print(f'Processing failed: {{e}}')
        traceback.print_exc()
        return False

def worker_loop():
    """Pull from queue, process, retry on failure."""
    while True:
        result = redis_client.brpop('webhook:queue', timeout=5)
        if not result:
            continue

        _, webhook_id = result
        envelope_json = redis_client.get(f'webhook:{{webhook_id}}')

        if not envelope_json:
            continue

        envelope = json.loads(envelope_json)
        envelope['retries'] = envelope.get('retries', 0)

        if process_webhook(envelope):
            envelope['status'] = 'delivered'
            redis_client.setex(
                f'webhook:{{webhook_id}}',
                3600,
                json.dumps(envelope)
            )
        else:
            if envelope['retries'] < MAX_RETRIES:
                backoff = RETRY_BACKOFF[envelope['retries']]
                envelope['retries'] += 1
                envelope['status'] = 'retrying'
                redis_client.setex(
                    f'webhook:{{webhook_id}}',
                    86400,
                    json.dumps(envelope)
                )
                time.sleep(backoff)
                redis_client.lpush('webhook:queue', webhook_id)
            else:
                envelope['status'] = 'failed'
                redis_client.setex(
                    f'webhook:{{webhook_id}}',
                    86400,
                    json.dumps(envelope)
                )
                redis_client.lpush('webhook:dead-letter', webhook_id)
                print(f'Dead letter: {{webhook_id}}')

ROUTERS = {{
    'stripe': handle_stripe_event,
    'github': handle_github_event,
    'custom': handle_custom_event,
}}

The Docker Compose

Both services share Redis and run side by side:

version: '3.8'

services:
  redis:
    image: redis:7-alpine
    volumes:
      - redis-data:/data

  relay:
    build: ./relay
    ports:
      - "8080:8080"
    environment:
      - REDIS_HOST=redis
      - STRIPE_WEBHOOK_SECRET=***
    depends_on:
      - redis

  worker:
    build: ./worker
    environment:
      - REDIS_HOST=redis
      - AGENT_ENDPOINT=***
    depends_on:
      - redis

volumes:
  redis-data:

Why This Matters for AI Agents

AI agents are stateful by nature. They maintain context, track conversation threads, and manage long-running tasks. Webhooks are the primary way the external world interrupts that flow — a form submission arrives, a payment completes, a data pipeline finishes.

If your webhook handling is fragile, your agent is fragile. Missed webhooks mean missed events, which means inconsistent state, which means your agent makes decisions on incomplete information.

The relay pattern gives you:

Decoupled scaling. Scale the relay for burst traffic during peak form submissions. Scale the worker for sustained throughput.
Fault isolation. A slow database query in the worker doesn't timeout the webhook sender.
Audit trail. Every webhook is stored with its processing status. You can replay, debug, or forward dead-lettered events.
Retry semantics. Transient network errors get fast retries, rate limits get long backoffs.

343 Tests and Counting

Architecture is nothing without verification. AgentForms has 343 tests covering relay validation, worker processing, retry logic, and failure paths. Here's a representative test:

def test_relay_ack_before_processing(self):
    """Relay should respond 200 before worker processes."""
    response = self.client.post('/webhook/stripe',
        data=TEST_PAYLOAD,
        headers={{'Stripe-Signature': TEST_SIGNATURE}}
    )

    self.assertEqual(response.status_code, 200)

    queue_size = redis_client.llen('webhook:queue')
    self.assertGreater(queue_size, 0)

Going Further

This is the core relay architecture behind AgentForms. In production, you'd add metrics (Prometheus), tracing (OpenTelemetry), and a proper dead-letter dashboard. But the fundamental pattern — fast ack in the relay, reliable processing in the worker, Redis as the bridge — stays the same regardless of scale.

If you're building AI agents that need to handle real-world events, start with the relay. Everything else builds on top.

Visit https://agentforms.io to learn more.

How I Built Form Infrastructure for AI Agents

Vincent — Tue, 16 Jun 2026 17:14:24 +0000

AI agents are getting better at reasoning but they still struggle with one thing: structured input. They can generate text, analyze data, and make decisions — but when you need them to collect information from users in a consistent format, you hit a wall.

I built AgentForms to solve exactly that. It's form infrastructure designed from the ground up for AI agents, not humans. Here's how it works.

The Problem

Traditional form builders (Typeform, Jotform, Google Forms) are human-first. They optimize for UX, drag-and-drop editors, and pretty templates. But AI agents don't care about aesthetics — they need reliable APIs, structured data, and programmatic control.

Every time I tried to build an AI application that needed user input, I ended up writing the same boilerplate: form validation, webhook handling, email notifications, payment processing. After the third time, I decided to build it once properly.

Architecture

AgentForms runs as two Docker containers:

services:
  relay:
    build: ./relay
    ports:
      - "5060:5060"
    volumes:
      - ./app:/app/app
      - ./data:/app/data
    environment:
      - DATABASE_URL=sqlite:///app/data/agentforms.db

  worker:
    build: ./worker
    depends_on:
      - relay
      - redis
    environment:
      - REDIS_URL=redis://redis:6379

The relay handles HTTP requests. The worker processes async tasks. Redis sits between them for job queuing. Simple, testable, deployable anywhere.

Form Submission Flow

@app.route('/forms/<form_id>/submit', methods=['POST'])
def submit_form(form_id):
    form = Form.query.get_or_404(form_id)

    data = request.get_json()
    errors = validate(data, form.schema)
    if errors:
        return jsonify({'errors': errors}), 422

    submission = Submission(form_id=form_id, data=data)
    db.session.add(submission)
    db.session.commit()

    worker.enqueue('process_submission', submission_id=submission.id)

    return jsonify({'id': submission.id}), 201

Key decisions:

SQLite with WAL mode — single-server deployment, zero configuration. WAL mode handles concurrent reads without locking.
Schema validation at submission time — fail fast, return structured errors.
Worker queue for side effects — email, payments, webhooks all happen async. The API response is immediate.

OAuth Integration

One of the harder pieces was OAuth. Agents need to authenticate users, but OAuth flows are notoriously complex to implement correctly.

Here's our OAuth callback handler:

@app.route('/auth/oauth/callback', methods=['GET'])
def oauth_callback():
    code = request.args.get('code')
    provider = request.args.get('provider')

    token = oauth_providers[provider].exchange_code(code)
    user_info = oauth_providers[provider].get_user_info(token)

    user = User.query.filter_by(
        provider=provider,
        provider_id=user_info['id']
    ).first()

    if not user:
        user = User(
            provider=provider,
            provider_id=user_info['id'],
            email=user_info['email']
        )
        db.session.add(user)

    session_token = generate_session_token(user.id)
    db.session.commit()

    return redirect(f"/dashboard?token={session_token}")

We abstract each provider behind a common interface. Adding a new provider means implementing three methods: exchange_code(), get_user_info(), and refresh_token(). No changes to the core auth flow.

Stripe Billing

Payment processing is where most indie SaaS products fail. Not because the integration is hard — because handling edge cases is hard. Failed payments, webhook retries, subscription changes, refunds.

Our Stripe webhook handler:

@app.route('/billing/stripe/webhook', methods=['POST'])
def stripe_webhook():
    payload = request.get_data()
    sig_header = request.headers.get('Stripe-Signature')

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, stripe_webhook_secret
        )
    except stripe.error.SignatureVerificationError:
        return 'Invalid signature', 400

    if event['type'] == 'checkout.session.completed':
        session = event['data']['object']
        plan_id = session['metadata'].get('plan_id')
        customer_email = session['customer_email']

        activate_subscription(customer_email, plan_id)

    elif event['type'] == 'invoice.payment_failed':
        customer = event['data']['object']['customer']
        handle_failed_payment(customer)

    return 'OK', 200

The critical detail: we verify the Stripe signature before processing anything. This prevents spoofed webhook attacks — a common mistake in indie SaaS implementations.

Webhook Forwarding

Users want their form submissions forwarded to their own endpoints — Slack, Discord, custom APIs.

Our webhook relay handles retries with exponential backoff:

def forward_webhook(submission_id, target_url, max_retries=3):
    submission = Submission.query.get(submission_id)
    payload = submission.data

    for attempt in range(max_retries):
        try:
            response = requests.post(
                target_url,
                json=payload,
                timeout=10,
                headers={'Content-Type': 'application/json'}
            )

            if response.status_code < 500:
                log_webhook_delivery(submission_id, response.status_code)
                return True

        except requests.exceptions.RequestException as e:
            delay = 2 ** attempt
            time.sleep(delay)

    log_webhook_failure(submission_id, target_url)
    return False

Simple, but it handles the 80% case. Server errors (5xx) get retried. Client errors (4xx) don't — your endpoint told us something's wrong with the data, so retrying won't help.

Why This Matters for AI

AI agents need structured input to be useful. A form is a contract: "give me these fields, I'll give you this result." Without that structure, agents are guessing.

AgentForms provides that structure as an API. Agents can:

Create forms programmatically
Submit data with validation
Trigger workflows on submission
Handle payments and billing

All through a consistent API. No scraping HTML, no guessing at field names, no dealing with CAPTCHAs.

What's Next

The core is built. 343 tests, Docker deployment, Stripe billing, OAuth, webhook forwarding. Now it needs users.

If you're building AI applications that need structured input, AgentForms is probably worth looking at. Starter plan is $9/month — cheaper than debugging your own webhook retry logic.

Building in public. Questions or contributions? Hit me up on Nostr or check out the docs at agentforms.io.