Ep 82: Webhooks & Event-Driven AI: Async Architectures That Scale

snippet
code
Your AI chatbot works great in a request/response loop. User sends message → model responds → done. But production AI systems aren't chatbots.

They're triggered by events: a new email arrives, a customer signs up, a document is uploaded, a sensor fires, a cron job ticks. The AI doesn't sit waiting for user input. It reacts.

And when 10,000 events arrive in the same second — Diwali sale going live, Black Friday traffic spike, system-wide alert — your synchronous request/response architecture collapses.

Event-driven AI solves this. Webhooks trigger processing. Message queues handle backpressure. Workers scale independently. The AI processes events at its own pace, never dropping a single one.

The Problem with Synchronous AI

Most AI applications start synchronous:

python

The simple way — synchronous, blocking

@app.route("/api/analyze", methods=["POST"])

snippet
code
def analyze():
text = request.json["text"]
    result = call_llm(text)  # Takes 2-5 seconds
    save_to_db(result)       # Takes 100ms
    send_email(result)       # Takes 500ms
    return jsonify(result)   # Total: 3-6 seconds blocking

This breaks at scale:

Concurrent Users Response Time Problem

10	3 sec	Fine
100	15 sec	Slow — thread pool exhausted
1,000	Timeout	Dead — all workers blocked on LLM calls
10,000	502	Load balancer gives up

Every request ties up a server thread waiting for the LLM. If the LLM takes 5 seconds and you have 20 workers, you can handle 4 requests/second. That's it.

⚡

Event-Driven Architecture: The Solution

Decouple the trigger from the processing:

snippet
code
Synchronous (before):

User → API → LLM → DB → Email → Response (5 seconds, blocking)

Event-driven (after):

snippet
code
User → API → Queue → Response (50ms, non-blocking)
↓
         Worker → LLM → DB → Email (5 seconds, but async)

The user gets an immediate response. The heavy processing happens in the background. If 10,000 events arrive at once, they queue up and get processed as capacity allows.

🔧

Webhooks: Events Pushing to You

A webhook is an HTTP callback — instead of polling "any new data?", the service pushes events to your URL:

python

Your webhook endpoint

@app.route("/webhooks/stripe", methods=["POST"])

snippet
code
def stripe_webhook():
event = request.json
if event["type"] == "payment_intent.succeeded":
        # Customer paid — trigger AI processing
        customer = event["data"]["object"]["customer"]
        queue.enqueue("generate_receipt", customer_id=customer)
        queue.enqueue("personalize_followup", customer_id=customer)
elif event["type"] == "dispute.created":
        # Dispute — AI analyzes and drafts response
        dispute = event["data"]["object"]
        queue.enqueue("analyze_dispute", dispute_id=dispute["id"])
return jsonify({"status": "ok"}), 200  # Respond fast!

Webhook Best Practices

Practice Why

Return 200 quickly (< 5 sec)	Sender will retry on timeout
Validate signatures	Prevent spoofed events
Idempotent processing	Events can arrive more than once
Queue for processing	Don't do heavy work in the handler
Store raw events	Debug and replay later

python

import hmac

import hashlib

def verify_webhook(payload, signature, secret):

example
code
"""Verify webhook signature to prevent spoofing."""
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.route("/webhooks/github", methods=["POST"])

snippet
code
def github_webhook():
signature = request.headers.get("X-Hub-Signature-256")
    if not verify_webhook(request.data, signature, GITHUB_SECRET):
        return "Unauthorized", 401
event = request.json
    event_type = request.headers.get("X-GitHub-Event")
# Store raw event (for replay/debugging)
    db.events.insert_one({
        "source": "github",
        "type": event_type,
        "payload": event,
        "received_at": datetime.utcnow(),
        "processed": False
    })
# Queue for AI processing
    if event_type == "pull_request" and event["action"] == "opened":
        queue.enqueue("ai_code_review", pr_url=event["pull_request"]["url"])
return "", 200

📊

Message Queues: The Backbone

Message queues decouple producers (webhook handlers) from consumers (AI workers):

Redis Queue (Simple)

python

from rq import Queue

from redis import Redis

snippet
code
redis = Redis()
queue = Queue(connection=redis)

Producer: webhook handler

def handle_new_email(email_data):

example
code
queue.enqueue(
        process_email_with_ai,
        email_data,
        job_timeout="5m",      # AI processing timeout
        retry=3,               # Retry on failure
        result_ttl=3600        # Keep result for 1 hour
    )

Consumer: AI worker

def process_email_with_ai(email_data):

example
code
# Classify the email
    classification = llm_classify(email_data["subject"], email_data["body"])

example
code
# Route based on classification
    if classification == "urgent":
        send_notification(email_data["from"], "urgent")
    elif classification == "support":
        draft = llm_draft_response(email_data)
        save_draft(email_data["id"], draft)
    elif classification == "spam":
        mark_spam(email_data["id"])

bash

Run workers

rq worker --burst # Process all queued jobs and exit

rq worker # Long-running worker

snippet
code
RabbitMQ (Production-Grade)

python

import pika

import json

Producer

snippet
code
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='ai_tasks', durable=True)

def publish_task(task_type, data):

example
code
channel.basic_publish(
        exchange='',
        routing_key='ai_tasks',
        body=json.dumps({"type": task_type, "data": data}),
        properties=pika.BasicProperties(delivery_mode=2)  # Persistent
    )

Consumer

def callback(ch, method, properties, body):

example
code
task = json.loads(body)

example
code
try:
        if task["type"] == "summarize":
            result = llm_summarize(task["data"]["text"])
            save_result(task["data"]["id"], result)
        elif task["type"] == "translate":
            result = llm_translate(task["data"]["text"], task["data"]["target_lang"])
            save_result(task["data"]["id"], result)

example
code
ch.basic_ack(delivery_tag=method.delivery_tag)  # Acknowledge success
    except Exception as e:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)  # Retry

snippet
code
channel.basic_qos(prefetch_count=1)  # One message at a time per worker

channel.basic_consume(queue='ai_tasks', on_message_callback=callback)

channel.start_consuming()

Queue Comparison for AI Workloads

Feature Redis Queue RabbitMQ AWS SQS Kafka

Complexity Simple Medium Low (managed) High

Durability Optional Yes Yes Yes

Ordering FIFO FIFO Approx FIFO Partition-ordered

Dead letter queue Plugin Built-in Built-in Manual

Throughput	100K/s	50K/s	3K/s/queue	1M+/s
Best for	Dev/small scale	Production	Serverless	High-volume streams

Architecture Patterns for Event-Driven AI

Pattern 1: Fan-Out

One event triggers multiple AI processes:

New document uploaded

example
code
├──→ Queue: "summarize" → Summarization worker → Summary saved
    ├──→ Queue: "extract"   → Entity extraction worker → Entities saved
    ├──→ Queue: "embed"     → Embedding worker → Vector stored
    └──→ Queue: "classify"  → Classification worker → Tags saved

python

def handle_document_upload(doc_id, content):

example
code
"""Fan-out: one event, multiple AI tasks."""
    queue.enqueue("summarize", doc_id=doc_id, content=content)
    queue.enqueue("extract_entities", doc_id=doc_id, content=content)
    queue.enqueue("generate_embedding", doc_id=doc_id, content=content)
    queue.enqueue("classify_document", doc_id=doc_id, content=content)

Pattern 2: Pipeline

Events flow through a chain of AI processors:

Raw text → Translate → Summarize → Extract entities → Store

example
code
↓            ↓              ↓
           Queue 1      Queue 2        Queue 3

python

Stage 1: Translate

def translate_worker(task):

example
code
translated = llm_translate(task["text"], "english")
    queue.enqueue("summarize", text=translated, original_id=task["id"])

Stage 2: Summarize

def summarize_worker(task):

example
code
summary = llm_summarize(task["text"])
    queue.enqueue("extract_entities", text=summary, original_id=task["original_id"])

Stage 3: Extract entities

def extract_worker(task):

example
code
entities = llm_extract(task["text"])
    save_result(task["original_id"], entities=entities)

Pattern 3: Saga (With Compensation)

Complex workflows where failures need rollback:

python

class OrderProcessingSaga:

example
code
"""Multi-step AI workflow with compensation on failure."""

example
code
async def execute(self, order):
        try:
            # Step 1: AI validates the order
            validation = await queue.enqueue("validate_order", order)
            if not validation["valid"]:
                return {"error": validation["reason"]}

example
code
# Step 2: AI generates recommendations
            recs = await queue.enqueue("generate_recommendations", order)

example
code
# Step 3: AI prices the bundle
            pricing = await queue.enqueue("calculate_pricing", order, recs)

example
code
# Step 4: Confirm order
            confirmation = await queue.enqueue("confirm_order", order, pricing)

example
code
return confirmation

example
code
except StepFailed as e:
            # Compensate: undo completed steps
            if e.step >= 3:
                await queue.enqueue("reverse_pricing", order)
            if e.step >= 2:
                await queue.enqueue("clear_recommendations", order)
            raise

💡

Real-World Example: AI Email Assistant

Putting it all together — an event-driven AI email processing system:

python

Architecture:

Gmail webhook → API → Redis Queue → AI Workers → Actions

1. Webhook receiver

@app.route("/webhooks/gmail", methods=["POST"])

snippet
code
def gmail_webhook():
notification = request.json
    # Gmail sends push notifications about new emails
    history_id = notification["historyId"]
# Quickly acknowledge and queue
    queue.enqueue("fetch_and_process_email", history_id=history_id)
    return "", 200

2. Fetch and classify worker

def fetch_and_process_email(history_id):

example
code
emails = gmail_api.get_new_emails(since=history_id)

example
code
for email in emails:
        # Classify with AI
        category = llm_classify(email["subject"], email["snippet"])

example
code
# Route to specialized queues
        routing = {
            "urgent": "urgent_email_queue",
            "support": "support_email_queue",
            "newsletter": "newsletter_queue",
            "spam": None  # Drop
        }

example
code
target_queue = routing.get(category)
        if target_queue:
            queue.enqueue(target_queue, email_id=email["id"], category=category)

3. Specialized workers

def urgent_email_worker(email_id, category):

example
code
email = gmail_api.get_email(email_id)
    summary = llm_summarize(email["body"], max_words=50)
    send_push_notification(f"🚨 Urgent: {summary}")

def support_email_worker(email_id, category):

example
code
email = gmail_api.get_email(email_id)
    draft = llm_draft_response(
        email["body"],
        tone="professional",
        include_faq=True
    )
    gmail_api.create_draft(reply_to=email_id, body=draft)
    send_notification(f"📝 Draft reply created for: {email['subject']}")

🔬

Scaling Workers

The beauty of event-driven: scale each component independently:

bash

Scale AI workers based on queue depth

Kubernetes HPA (Horizontal Pod Autoscaler)

apiVersion: autoscaling/v2

kind: HorizontalPodAutoscaler

metadata:

name: ai-worker-hpa

spec:

scaleTargetRef:

example
code
kind: Deployment
    name: ai-worker

minReplicas: 2

maxReplicas: 20

metrics:

type: External

example
code
external:
      metric:
        name: redis_queue_length
        selector:
          matchLabels:
            queue: ai_tasks
      target:
        type: AverageValue
        averageValue: "10"  # Scale up when > 10 msgs per worker

Queue Depth Workers Processing Capacity

< 20	2	2 LLM calls/sec
20-100	5	5 LLM calls/sec
100-500	10	10 LLM calls/sec
500+	20	20 LLM calls/sec

Workers scale with demand. No over-provisioning during quiet hours. No drops during spikes.

🛡️

Try It Yourself

python

Quick event-driven AI pipeline with Redis Queue

Install: pip install rq redis openai

worker.py

from rq import Worker, Queue

from redis import Redis

from openai import OpenAI

snippet
code
redis = Redis()
client = OpenAI()

def ai_summarize(text, doc_id):

example
code
"""AI summarization worker."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Summarize in 2-3 sentences."},
            {"role": "user", "content": text}
        ]
    )
    summary = response.choices[0].message.content
    redis.hset(f"doc:{doc_id}", "summary", summary)
    print(f"Summarized doc {doc_id}: {summary[:50]}...")
    return summary

producer.py

from rq import Queue

from redis import Redis

snippet
code
queue = Queue(connection=Redis())

Simulate webhook events

snippet
code
documents = [
{"id": 1, "text": "Long article about climate change..."},
    {"id": 2, "text": "Research paper on transformer architectures..."},
    {"id": 3, "text": "News article about Indian elections..."},

]

for doc in documents:

example
code
queue.enqueue("worker.ai_summarize", doc["text"], doc["id"])
    print(f"Queued doc {doc['id']}")

Run: python producer.py (queue tasks)

Run: rq worker (process tasks)

📦

Practical Takeaways

Synchronous AI doesn't scale — every request blocks a thread waiting for the LLM

Webhooks push events to you — always return 200 fast, queue the heavy work

Message queues decouple trigger from processing — backpressure handled automatically

Fan-out, pipeline, and saga — three patterns for different AI workflow shapes

Scale workers independently — 2 workers at night, 20 during traffic spikes

Always verify webhooks, always be idempotent — events can be spoofed or duplicated

🚀

What's Next?

Episode 83: Semantic Caching — Your event-driven system processes thousands of LLM calls. Many are similar: "What's the weather in Mumbai?" and "Mumbai weather today?" are basically the same question. Semantic caching matches by meaning, not exact string — cutting LLM costs by 30-60%.

← Previous

Ep 81: A2A (Agent-to-Agent)

Next: Episode 83 — Semantic Caching

Exact-match caching is boring. Same question, same answer. Semantic caching matches similar questions — and that changes everything.

This is part of a 98-episode series covering AI engineering from tokens to production deployment.

The Problem with Synchronous AI

Most AI applications start synchronous:

The simple way — synchronous, blocking

This breaks at scale:

Concurrent Users Response Time Problem

Event-Driven Architecture: The Solution

Decouple the trigger from the processing:

Webhooks: Events Pushing to You

Your webhook endpoint

Webhook Best Practices

Practice Why

Message Queues: The Backbone

Redis Queue (Simple)

Producer: webhook handler

Consumer: AI worker

Run workers

Producer

Consumer

Queue Comparison for AI Workloads

Feature Redis Queue RabbitMQ AWS SQS Kafka

Complexity Simple Medium Low (managed) High

Durability Optional Yes Yes Yes

Ordering FIFO FIFO Approx FIFO Partition-ordered

Dead letter queue Plugin Built-in Built-in Manual

Architecture Patterns for Event-Driven AI

Pattern 1: Fan-Out

One event triggers multiple AI processes:

New document uploaded

Pattern 2: Pipeline

Events flow through a chain of AI processors:

Stage 1: Translate

Stage 2: Summarize

Stage 3: Extract entities

Pattern 3: Saga (With Compensation)

Complex workflows where failures need rollback:

Real-World Example: AI Email Assistant

Architecture:

Gmail webhook → API → Redis Queue → AI Workers → Actions

1. Webhook receiver

2. Fetch and classify worker

3. Specialized workers

Scaling Workers

Scale AI workers based on queue depth

Kubernetes HPA (Horizontal Pod Autoscaler)

Queue Depth Workers Processing Capacity

Try It Yourself

Quick event-driven AI pipeline with Redis Queue

Install: pip install rq redis openai

worker.py

producer.py

Simulate webhook events

Run: python producer.py (queue tasks)

Run: rq worker (process tasks)

Practical Takeaways

Ep 83: Semantic Caching

Next: Episode 83 — Semantic Caching