Build a Webhook Aggregator with Fanout in Flask and the Telnyx Python SDK

Webhook endpoints have a way of starting simple and becoming a problem. The first version is a Flask route that receives a POST, does something, and returns 200. Then Telnyx redelivers an event because your server was slow, and the same call gets answered twice. Then a burst of call and SMS events lands in the same second, and your business logic blocks on the network. Then you need to explain what happened at 3am, and your logs are not enough.

This sample is a small, deliberate answer to that. It takes raw Telnyx webhooks and runs them through a six-stage pipeline: receive, verify, dedup, log, fan out, process. Your business logic never sees a duplicate, never sees a tampered event, and never blocks on the network.

Code: https://github.com/team-telnyx/telnyx-code-examples/tree/main/webhook-aggregator-fanout

Architecture

Telnyx webhook POST
        |
        v
[1] Receive   — Flask route reads raw body + headers
        |
        v
[2] Verify    — Ed25519 signature check via Telnyx SDK v4
        |
        v
[3] Dedup     — TTL-based in-memory KV store (default 300s)
        |
        v
[4] Log       — SQLite insert (INSERT OR IGNORE for race safety)
        |
        v
[5] Fan out   — Route to call queue or SMS queue by event type
        |
        v
[6] Process   — Drain queue: answer call + play greeting, or reply to SMS

The entire pipeline lives in one file: app.py. No Redis, no Celery, no message broker. The storage primitives are intentionally boring — SQLite for the audit log, a Python dict for the dedup KV — so you can run the whole thing locally without infrastructure.

Verify before you trust

The first thing the webhook handler does is verify the Ed25519 signature. The Telnyx Python SDK v4 makes this a single call:

import telnyx
from telnyx.lib.webhooks_ed25519 import unwrap_with_ed25519

telnyx_client = telnyx.Telnyx(
    api_key=os.getenv("TELNYX_API_KEY"),
    public_key=os.getenv("TELNYX_PUBLIC_KEY"),
)

@app.route("/webhooks", methods=["POST"])
def webhook_handler():
    raw_body = request.get_data(as_text=True)
    verified_event = unwrap_with_ed25519(
        telnyx_client, raw_body, request.headers
    )

unwrap_with_ed25519 reads the Telnyx-Ed25519-Signature and Telnyx-Ed25519-Timestamp headers, verifies the signature against your Telnyx public key, and returns a typed UnwrapWebhookEvent. If the signature is invalid or the timestamp is outside the replay window, it raises — and the handler returns 401 before any business logic runs.

This matters because the public key is not a secret. Anyone who intercepts the webhook traffic can read it. What they cannot do is produce a valid signature without your Telnyx private key. Verification is the difference between trusting the payload and trusting the source.

Deduplicate with a TTL KV store

Telnyx retries webhook delivery if your server does not respond within the timeout. That means the same event can arrive two, three, or five times within a few minutes. Without deduplication, every retry triggers your business logic again — a call gets answered twice, a customer gets three identical SMS replies.

The sample uses an in-memory dict with a TTL:

dedup_store = {}  # {event_id: (timestamp, timestamp)}
DEDUP_TTL_SECONDS = int(os.getenv("DEDUP_TTL_SECONDS", "300"))

def is_duplicate(event_id):
    current_time = time.time()
    # Clean up expired entries
    expired = [k for k, (ts, _) in dedup_store.items()
               if current_time - ts > DEDUP_TTL_SECONDS]
    for key in expired:
        del dedup_store[key]
    if event_id in dedup_store:
        return True
    dedup_store[event_id] = (current_time, current_time)
    return False

Every check sweeps expired entries first, so the store never grows unbounded. The default TTL is 300 seconds, which covers Telnyx's retry window. The event ID comes from the Telnyx event payload when present, and falls back to a SHA-256 of the sorted payload when it does not.

The dedup check happens before the SQLite log and before the fanout. A duplicate event returns {"status": "duplicate"} with a 200 — because returning a non-2xx would make Telnyx retry again.

Log to SQLite before you act

Every non-duplicate event is written to SQLite before any action fires:

def log_event(event_id, event_type, payload):
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute(
        "INSERT OR IGNORE INTO webhook_events "
        "(event_id, event_type, payload, received_at, processed_at) "
        "VALUES (?, ?, ?, ?, ?)",
        (event_id, event_type, json.dumps(payload),
         datetime.now(timezone.utc).isoformat(),
         datetime.now(timezone.utc).isoformat())
    )
    conn.commit()
    conn.close()

INSERT OR IGNORE on the UNIQUE event_id column makes the log idempotent — even if two identical events slip past the in-memory dedup in a race, SQLite enforces uniqueness at the storage layer.

The log exists so that when something goes wrong at 3am, you have a real table: event_id, event_type, payload, received_at, processed_at. You can query it, export it, or build a dashboard on top of it. It is operational data, not a log file you grep.

Fan out by event type

After verification, dedup, and logging, the event is routed to one of two in-memory queues based on its type:

ACTION_TYPES = ["call", "sms"]
action_queues = defaultdict(list)

# In webhook_handler:
if "call" in event_type.lower():
    enqueue_action("call", event)
elif "message" in event_type.lower() or "sms" in event_type.lower():
    enqueue_action("sms", event)

The fanout is a router, not a dispatcher. It decides which queue gets the event; it does not execute the action inline. The queue is drained in the same request for this sample, but in production you would move the drain to a separate worker process.

The queue separation matters because call actions and SMS actions have different latency profiles and different failure modes. A call that takes 800ms to answer should not block an SMS reply that could fire in 50ms. Separating the queues lets you prioritize, retry, and monitor each action type independently.

Process: answer calls, reply to SMS

The queue drain uses the Telnyx Python SDK v4 client to take the actual action:

def process_call_action(event_data):
    payload = event_data.get("data", {}).get("payload", {})
    call_control_id = payload.get("call_control_id")
    telnyx_client.calls.answer(call_control_id=call_control_id)
    telnyx_client.calls.playback_start(
        call_control_id=call_control_id,
        audio_url="https://example.com/greeting.mp3"
    )

def process_sms_action(event_data):
    payload = event_data.get("data", {}).get("payload", {})
    from_number = payload.get("from")
    to_number = payload.get("to")
    text = payload.get("text", "")
    telnyx_client.messages.create(
        from_=to_number,
        to=from_number,
        text=f"Thanks for your message! We received: {text[:50]}..."
    )

Note the field access: event_data.get("data", {}).get("payload", {}). The Telnyx webhook envelope wraps the payload inside data.payload, and the SDK v4 typed event follows that structure. If you are migrating from SDK v2, this is the most common breakage point — v2 gave you the flat payload, v4 gives you the wrapped envelope.

Run the demo without Telnyx credentials

The sample ships with a single-file demo launcher (demo/demo_server.py) that runs the entire pipeline without Telnyx credentials or ngrok. It generates an Ed25519 keypair at startup, injects the public key into the Telnyx client, stubs the calls and messages API surfaces, and serves a dashboard at http://localhost:5555/.

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/webhook-aggregator-fanout
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python demo/demo_server.py

Open the dashboard, click "Send Call Webhook" or "Send SMS Webhook," and watch the six-stage pipeline execute live. The dashboard shows the dedup KV, the SQLite event log, and the fanout queues in real time.

To test deduplication, send the same event twice — the first returns {"status": "success"}, the second returns {"status": "duplicate"}, and only one row appears in the events table.

Run it for real

cp .env.example .env
# Set TELNYX_API_KEY and TELNYX_PUBLIC_KEY in .env
flask --app app run --port 5000

Point your Telnyx webhook URL to https://your-public-url/webhooks (use ngrok for local testing) and configure your call and messaging profiles to deliver events there.

Where to take it next

For production, three things:

1. Move the queue drain to a worker. The sample drains queues in the same request so the pipeline is visible end to end. In production, enqueue to Redis or a real queue and drain from a separate process.
2. Add a dead-letter queue. If an action fails, it should not disappear. Log it, retry with backoff, and surface it for manual intervention.
3. Add monitoring on the queue depths. If the call queue is growing faster than it drains, you have a capacity problem before you have a customer problem.

The core pattern is simple: verify before you trust, deduplicate before you act, log before you process, and fan out before you execute. Everything else is infrastructure.