Receive and verify IllyVoIP webhooks in Python with FastAPI

Implement an inbound FastAPI route for IllyVoIP webhooks, verify the original request bytes and store each authenticated event durably before acknowledging it. The example below is a complete receiving inbox; it does not execute a downstream business action or claim exactly-once external effects.

1. Prepare a dedicated receiver

Use a supported Python release, FastAPI and Uvicorn in a private virtual environment. Record your installed versions. Configure a separate HTTPS endpoint through Settings → Integrations → Webhooks only when the receiver is ready, and subscribe to the events the application actually needs.

Store that endpoint’s signing secret as ILLYVOIP_WEBHOOK_SECRET. It is not the account API key. Set WEBHOOK_DB to a private absolute SQLite path on durable local storage and WEBHOOK_SCOPE to a stable account/application identifier. Do not change the scope casually, because it participates in duplicate detection.

2. Save the receiving application

Save this as receiver.py. The 256 KiB body limit and five-minute timestamp tolerance are application choices; check them against your subscribed event sizes and clock policy. Apply request-rate and body-read time limits at the HTTPS reverse proxy as well.

import hashlib, hmac, json, os, re, sqlite3, time
from contextlib import asynccontextmanager, contextmanager
from fastapi import FastAPI, HTTPException, Request
from starlette.concurrency import run_in_threadpool

SECRET = os.environ["ILLYVOIP_WEBHOOK_SECRET"].encode()
DB = os.environ["WEBHOOK_DB"]  # Private absolute path on durable local storage.
SCOPE = os.environ["WEBHOOK_SCOPE"]  # Stable account/application identity.
MAX_BYTES = 262144  # Application limit; choose for your subscribed events.
if not SECRET or not SCOPE or not os.path.isabs(DB):
    raise RuntimeError("Missing secret/scope or absolute database path")

@contextmanager
def connect():
    c = sqlite3.connect(DB, timeout=1)
    try:
        c.execute("PRAGMA synchronous=FULL")
        with c:
            yield c
    finally:
        c.close()

def initialize():
    with connect() as c:
        c.execute("""CREATE TABLE IF NOT EXISTS inbox (
            scope TEXT NOT NULL, event_id TEXT NOT NULL,
            body BLOB NOT NULL, received_at INTEGER NOT NULL,
            state TEXT NOT NULL DEFAULT 'pending',
            PRIMARY KEY(scope, event_id))""")

@asynccontextmanager
async def lifespan(app):
    await run_in_threadpool(initialize)
    yield

app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None,
              openapi_url=None)

def verified(raw, header, now):
    parts = header.split(",")
    if len(parts) != 2:
        return False
    values = {}
    for part in parts:
        k, sep, v = part.strip().partition("=")
        if not sep or k in values:
            return False
        values[k] = v
    stamp, digest = values.get("t", ""), values.get("v1", "")
    if not re.fullmatch(r"[0-9]{1,12}", stamp):
        return False
    if not re.fullmatch(r"[a-fA-F0-9]{64}", digest):
        return False
    if abs(now - int(stamp)) > 300:
        return False
    expected = hmac.new(SECRET, stamp.encode() + b"." + raw,
                        hashlib.sha256).digest()
    return hmac.compare_digest(expected, bytes.fromhex(digest))

def persist(event_id, raw):
    with connect() as c:
        c.execute("""INSERT INTO inbox(scope,event_id,body,received_at)
                     VALUES(?,?,?,?) ON CONFLICT(scope,event_id) DO NOTHING""",
                  (SCOPE, event_id, raw, int(time.time())))

@app.post("/webhooks/illyvoip")
async def receive(request: Request):
    if request.headers.get("content-type", "").split(";", 1)[0].strip().lower() != "application/json":
        raise HTTPException(415, "Expected JSON")
    headers = request.headers.getlist("illyvoip-signature")
    if len(headers) != 1:
        raise HTTPException(401, "Invalid signature")
    raw = bytearray()
    async for chunk in request.stream():
        if len(raw) + len(chunk) > MAX_BYTES:
            raise HTTPException(413, "Body too large")
        raw.extend(chunk)
    raw = bytes(raw)
    if not verified(raw, headers[0], int(time.time())):
        raise HTTPException(401, "Invalid signature")
    try:
        event = json.loads(raw)
    except (ValueError, UnicodeDecodeError):
        raise HTTPException(400, "Invalid JSON")
    if (not isinstance(event, dict)
        or not isinstance(event.get("id"), str)
        or not 1 <= len(event["id"]) <= 200
        or not isinstance(event.get("type"), str)
        or not 1 <= len(event["type"]) <= 200):
        raise HTTPException(400, "Invalid event envelope")
    try:
        await run_in_threadpool(persist, event["id"], raw)
    except sqlite3.Error:
        raise HTTPException(503, "Storage unavailable")
    return {"accepted": True}

3. Run privately, then expose only the route

python3 -m venv .venv
.venv/bin/pip install fastapi uvicorn
umask 077
# Supply the three configuration variables through your secret/service manager.
.venv/bin/uvicorn receiver:app --host 127.0.0.1 --port 8015

Terminate HTTPS at your approved reverse proxy and forward the exact webhook path to the loopback listener. Preserve the body and signature header. Do not expose a development reload/debug service to the Internet. Protect the database directory and include it in the application’s normal backup and retention process.

In production, supervise the process and monitor storage failures. SQLite is a bounded single-host example; distributed receivers need an appropriately shared durable store and the same unique event identity.

4. Understand the verification boundary

IllyVoIP signs the timestamp, a period and the original JSON bytes using HMAC-SHA256. Parsing and reserializing JSON before verification can change those bytes. The sender’s timestamp belongs to the delivery attempt, not necessarily the time the underlying event occurred.

The route checks the body id and type, then persists the event without performing an action. Before a worker handles it, validate the supported event type and its actual data schema. The Illyvoip-Delivery header identifies a delivery job; it does not replace the event id for deduplication.

This is a normal FastAPI POST route. FastAPI’s app.webhooks documentation facility describes outgoing events an application sends; it is not the receiving handler shown here.

5. Test acceptance and failure cases

Use a synthetic body and a local test secret to compute valid signatures. Check valid input, one changed byte, changed JSON spacing, wrong secret, malformed digest, repeated signature headers, stale/future timestamps, oversize input and malformed JSON. Send two concurrent copies of the same event and confirm only one inbox row exists.

Make the test database unavailable and confirm the receiver returns failure, not success with a lost event. A repeated verified event already stored in the inbox should receive a successful acknowledgement without a second row.

6. Add a separate worker deliberately

The inbox state starts as pending. A worker must claim a row, validate the subscribed business event and record its processing outcome. Design each downstream effect with its own stable identity and reconciliation path. A crash after sending an external message but before marking the row complete can otherwise repeat that message.

Do not replace durable acceptance with an in-memory set or a FastAPI background task. A returned 2xx means this receiver stored the event; inspect the worker’s result separately. Keep raw payload access private and retain it only as long as your workflow needs it.

Related guides and reference

• Webhook verification in Node.js
• Review delivery attempts
• Test a local receiver

FastAPI Request access. IllyVoIP API reference. Reviewed October 12, 2026. Examples are instructional: local checks do not establish a completed paid request, live third-party integration or production PBX acceptance.