Originally published on lowlatencyclub.ai
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 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.
The entire firewall is 178 lines of Python in a single Flask file.
How It Works
When a call hits a Telnyx number, the 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:
- Blocklist check — if the caller's number is in the in-memory blocklist, reject immediately. No lookup, no AI call.
- Number Lookup — returns carrier name, line type (landline/voip/mobile), and country code.
-
AI classification — the lookup data goes to an OpenAI-compatible chat completions endpoint with a system prompt that returns exactly one word:
CLEAN,SUSPICIOUS, orBLOCK. -
Route decision — based on the classification:
-
CLEAN→ answer and forward to your real number -
SUSPICIOUS→ answer and route to a honeypot (endless hold loop) -
BLOCK→ reject and add to the blocklist
-
The AI Classifier
The system prompt is deliberately constrained — one word only:
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"
One-word output keeps inference latency to a single token and makes parsing trivial — no JSON, no regex, just a string comparison.
If the AI call fails for any reason, the function defaults to CLEAN. This is a safe-fail — if the screening service is unavailable, legitimate callers still get through.
The Honeypot
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 and plays a hold message, then loops indefinitely:
# call.answered → honeypot flow
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"})})
# call.speak.ended → loop forever
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.",
"client_state": encode_state({"flow": "honeypot_hold"})})
The scammer stays on the line, burning their time and resources.
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 {}
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 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. If verification fails, the request is rejected 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
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.
Why Screen at the Carrier Edge?
Fraud screening typically runs at the application layer — after the call is already connected. By that point, the fraud has already consumed SIP trunks, media servers, and agent time.
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 called over the public internet. The screen-then-route decision is fast enough to sit inline on every inbound call.
Going to Production
This sample uses in-memory state. For production:
- Replace the in-memory blocklist with Redis
- Add rate limiting to the AI classification call
- Set up monitoring on classification latency and block rate
- Use
gunicorn -w 4 app:appas a process manager - Configure failover webhook URLs in the Telnyx Portal
Resources
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
Top comments (0)