Screen Every Inbound Call at the Carrier Edge: A Fraud Firewall in 178 Lines of Python

Every inbound call is a decision. Is this caller legitimate, or is it a scammer, robocaller, or toll-fraud bot? Most systems make that decision too late — after the call reaches the application layer, consuming resources, triggering downstream logic, and potentially connecting to a human agent who wastes 5 minutes on a fake call.

The Edge Fraud Firewall flips the order. It screens every inbound call at the carrier edge — before your app sees it — using three Telnyx APIs on a single platform: Number Lookup for caller identity, AI Inference for risk classification, and Call Control for reject/forward/route decisions. Clean calls pass through. Fraud gets rejected with cause code. Suspicious calls get sent to a honeypot that wastes the scammer's time.

The entire firewall is 178 lines of Python in a single Flask file.

How It Works

When a call hits a Telnyx number, the Telnyx platform sends a call.initiated webhook to the Flask server. Before doing anything else, the server verifies the Ed25519 webhook signature — if the request is not from Telnyx, it returns 401 and stops.

For verified webhooks, the screening pipeline runs in order:

1. Blocklist check — if the caller's number is in the in-memory blocklist, the call is rejected immediately. No lookup, no AI call.
2. Number Lookup — a GET /v2/number_lookup/{phone} call returns carrier name, line type (landline/voip/mobile), and country code.
3. AI classification — the lookup data goes to an OpenAI-compatible chat completions endpoint (POST /v2/ai/chat/completions) with a system prompt that returns exactly one word: CLEAN, SUSPICIOUS, or BLOCK.
4. Route decision — based on the classification:

- CLEAN → answer the call and forward to your real number - SUSPICIOUS → answer and route to a honeypot (endless hold loop) - BLOCK → reject the call and add the number to the blocklist

The call flow

if event_type == "call.initiated":
    direction = event.get("payload", {}).get("direction")
    caller = event.get("payload", {}).get("from", "")
    if direction == "incoming":
        if caller in blocklist:
            # Reject immediately — known bad caller
            requests.post(f".../calls/{call_control_id}/actions/reject", headers=HEADERS, timeout=10)
            return jsonify({"status": "blocked"})

        lookup = lookup_number(caller)
        risk = classify_caller(caller, lookup)
        call_screening[call_control_id] = {"phone": caller, "risk": risk, "lookup": lookup, "ts": time.time()}

        if risk == "BLOCK":
            blocklist.add(caller)
            requests.post(f".../calls/{call_control_id}/actions/reject", headers=HEADERS, timeout=10)
        elif risk == "SUSPICIOUS" and HONEYPOT_CONNECTION:
            requests.post(f".../calls/{call_control_id}/actions/answer", headers=HEADERS, timeout=10,
                          json={"client_state": encode_state({"flow": "honeypot"})})
        else:
            requests.post(f".../calls/{call_control_id}/actions/answer", headers=HEADERS, timeout=10,
                          json={"client_state": encode_state({"flow": "forward"})})

The Three Screening Layers

Layer 1: Blocklist (instant reject)

The blocklist is a Python set — O(1) lookup, no API calls needed. Numbers get added automatically when the AI classifier returns BLOCK, and can also be added manually via the REST API:

@app.route("/blocklist", methods=["POST"])
def add_to_blocklist():
    data = request.get_json() or {}
    phone = data.get("phone")
    if not phone:
        return jsonify({"error": "phone required"}), 400
    blocklist.add(phone)
    return jsonify({"status": "added", "phone": phone})

# Add a number to the blocklist
curl -X POST http://localhost:5000/blocklist \
  -H "Content-Type: application/json" \
  -d '{"phone": "+12125551234"}'

# Check the current blocklist
curl http://localhost:5000/blocklist

Layer 2: Number Lookup (caller identity)

The Number Lookup API returns carrier-level data for any phone number — carrier name, line type (landline, VoIP, mobile), and country code. This data feeds into the AI classifier.

def lookup_number(phone):
    """Check number reputation via Telnyx Number Lookup."""
    try:
        resp = requests.get(
            f"https://api.telnyx.com/v2/number_lookup/{phone}",
            headers=HEADERS, timeout=10
        )
        if resp.ok:
            return resp.json().get("data", {})
    except Exception as e:
        app.logger.error("Number lookup failed: %s", e)
    return {}

A VoIP number from a high-risk country gets different treatment than a mobile number from a domestic carrier. The AI classifier uses this context.

Layer 3: AI Classification (risk scoring)

The AI classifier sends the lookup data to an OpenAI-compatible chat completions endpoint with a strict system prompt:

def classify_caller(phone, lookup_data):
    try:
        resp = requests.post(INFERENCE_URL, headers=HEADERS, timeout=15, json={
            "model": AI_MODEL,
            "messages": [
                {"role": "system", "content": "You are a fraud detection system. Analyze caller data and respond with ONLY one word: CLEAN, SUSPICIOUS, or BLOCK."},
                {"role": "user", "content": f"Caller: {phone}\nCarrier: {lookup_data.get('carrier', {}).get('name', 'unknown')}\nType: {lookup_data.get('carrier', {}).get('type', 'unknown')}\nCountry: {lookup_data.get('country_code', 'unknown')}"}
            ]
        })
        if resp.ok:
            return resp.json()["choices"][0]["message"]["content"].strip().upper()
    except Exception as e:
        app.logger.error("Classification failed: %s", e)
    return "CLEAN"

The prompt is deliberately constrained — one word only. This keeps inference latency low (a single token, not a paragraph of explanation) and makes the output trivial to parse: CLEAN, SUSPICIOUS, or BLOCK.

If the AI call fails for any reason (timeout, bad model, network error), the function defaults to CLEAN. This is a safe-fail: better to let a call through than block a legitimate customer.

The Honeypot: Wasting Scammer Time

When the AI classifies a call as SUSPICIOUS, the firewall routes it to a honeypot instead of rejecting or forwarding. The honeypot answers the call, plays a hold message, and loops indefinitely:

elif event_type == "call.answered":
    state = decode_state(event.get("payload", {}).get("client_state", ""))
    flow = state.get("flow", "forward")
    if flow == "honeypot":
        requests.post(f".../calls/{call_control_id}/actions/speak",
                      json={"payload": "Please hold while I connect you to a specialist.",
                            "voice": "female", "language": "en-US",
                            "client_state": encode_state({"flow": "honeypot_hold"})})

When the TTS finishes, the call.speak.ended event fires, and the server plays another hold message:

elif event_type == "call.speak.ended":
    state = decode_state(event.get("payload", {}).get("client_state", ""))
    if state.get("flow") == "honeypot_hold":
        time.sleep(2)
        requests.post(f".../calls/{call_control_id}/actions/speak",
                      json={"payload": "All specialists are currently busy. Please continue to hold.",
                            "voice": "female", "language": "en-US",
                            "client_state": encode_state({"flow": "honeypot_hold"})})

The scammer stays on the line, burning their time and resources. Every minute spent on the honeypot is a minute not spent scamming a real person.

State Management with client_state

Telnyx Call Control is webhook-driven — each call action triggers a new webhook event. To track which flow a call is in (forward vs honeypot), the app uses client_state, a base64-encoded JSON blob that Telnyx passes back in the next webhook:

def encode_state(data):
    return base64.b64encode(json.dumps(data).encode()).decode()

def decode_state(b64):
    try:
        return json.loads(base64.b64decode(b64).decode())
    except Exception:
        return {}

When answering a suspicious call, the app encodes {"flow": "honeypot"} into client_state. When the call.answered webhook arrives, it decodes client_state to determine which path to take. No external database needed — the state travels with the webhook.

Webhook Signature Verification

The first thing the webhook handler does is verify the request is actually from Telnyx:

try:
    client.webhooks.unwrap(request.get_data(as_text=True), headers=dict(request.headers))
except Exception:
    return jsonify({"error": "invalid signature"}), 401

The Telnyx Python SDK verifies the Ed25519 signature against the raw request body and the telnyx-signature-ed25519 and telnyx-timestamp headers. If verification fails, the request is rejected with 401 before any processing happens.

Setup

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/edge-fraud-firewall-python
cp .env.example .env
pip install -r requirements.txt

Fill in .env:

TELNYX_API_KEY=KEY0123456789
TELNYX_PUBLIC_KEY=your_public_key_here
FORWARD_NUMBER=+1xxxxxxxxxx
AI_MODEL=moonshotai/Kimi-K2.6

Start the server and expose it for webhooks:

python app.py
ngrok http 5000

Configure the webhook URL in the Telnyx Portal: set the Call Control Application webhook URL to https://<your-ngrok-id>.ngrok.io/webhooks/voice.

Call your Telnyx number. The firewall screens the call before it reaches your forward number.

Why Screen at the Carrier Edge?

Fraud screening typically runs at the application layer — after the call has already been answered, routed, and connected. By that point, the fraud has already consumed your infrastructure: SIP trunks, media servers, agent time, downstream API calls.

Screening at the carrier edge means the decision happens on the same network that carries the call. Number Lookup and AI Inference run on Telnyx infrastructure, not a third-party API you call over the public internet. The screen-then-route decision is fast enough to sit inline on every inbound call — no perceptible delay for legitimate callers.

REST API

The firewall also exposes management endpoints:

Going to Production

This sample uses in-memory state (blocklist as a Python set, call_screening as a dict with TTL cleanup). For production:

• Replace in-memory state with Redis or a database
• Add rate limiting to the AI classification call
• Set up monitoring and alerting on classification latency and block rate
• Use gunicorn -w 4 app:app as a process manager
• Configure failover webhook URLs in the Telnyx Portal

Related Examples

• number-lookup-fraud-screener-python — Score and screen callers with Number Lookup reputation data
• fraud-alert-verification-python — Verify suspicious activity with an outbound confirmation call
• edge-geo-smart-router-python — Route inbound calls at the edge based on caller geography
• route-phone-calls-to-ai-agent-python — Forward screened, verified calls to an AI voice agent