DEV Community: Harpreet Singh Seehra

AI Voice Agent with Function Calling — Calling Real APIs Mid-Conversation in Python

Harpreet Singh Seehra — Fri, 07 Aug 2026 13:22:25 +0000

A phone-call voice agent that calls real APIs (weather, orders, balances) in the middle of a conversation. Speech recognition, LLM tool-calling, and Telnyx text-to-speech in one Flask file. One API key for calls, AI inference, and TTS.

The Problem: Voice Agents That Lie

You have an AI voice agent. It picks up calls. It greets callers. It responds in natural language.

But the moment someone asks, "What's the weather in San Francisco?", "Where is my order 12345?", or "What's my account balance?" — the agent hallucinates. It guesses. It invents a plausible-sounding answer with zero real data.

Voice agents without function calling are LLMs talking to thin air. They have no way to reach out for live information mid-conversation. They just pattern-match.

The ai-voice-agent-with-function-calling-python example fixes this in one Flask file. It wires three Telnyx capabilities — speech recognition via gather_using_ai, LLM tool-calling via AI Inference, and text-to-speech via speak — into a single webhook-driven loop. Callers ask questions in plain speech. The agent calls real functions. It speaks real answers back.

What It Does

Call your agent number. It picks up, greets you, and waits. You speak a request: "What's the weather in San Francisco?" The voice agent transcribes your speech via gather_using_ai, sends the transcript to Telnyx AI Inference with three tool definitions (check_weather, lookup_order, check_account_balance), and the model decides whether to call a tool or respond directly.

If the model calls a tool — say check_weather — your Python function runs, returns its JSON result to the model, and the model synthesizes a one-sentence spoken answer. speak() reads it aloud in natural voice. The conversation loop continues. You ask about an order. The agent calls lookup_order. You ask about your balance. The agent calls check_account_balance. No hallucinations, no guessing — every tool-backed answer is backed by real function output.

Step	What happens	Telnyx API
1	Caller dials agent number	Inbound call → `call.initiated` webhook
2	Agent answers + greets	`answer()` + `speak()` (TTS)
3	TTS ends → start listening	`call.speak.ended` → `gather_using_ai()`
4	Caller speaks a request	Speech-to-text via `gather_using_ai`
5	Transcript returned	`call.ai_gather.ended` webhook
6	Transcript → AI Inference (with tools)	`POST /v2/ai/chat/completions`
7	Model returns `tool_calls`	Loop: execute functions, re-infer
8	Model returns final text	`speak()` reads it aloud
9	Caller hangs up	`call.hangup` → cleanup

The Architecture

One Flask file. One webhook endpoint. In-memory conversation state keyed by call_control_id. No database, no Redis, no background workers. Telnyx owns the telephony, speech recognition, and TTS layers. Your code owns the conversation state and the function implementations.

Inbound call → Telnyx webhook → /webhooks/voice
                              ↓
                 call.initiated → answer() + greet (speak)
                              ↓
                 call.speak.ended → gather_using_ai()
                              ↓
                 call.ai_gather.ended → transcript
                              ↓
                 Transcript → AI Inference (with TOOLS array)
                              ↓
                 Model returns tool_calls?
                  ├── yes → execute_function() → re-infer → loop
                  └── no  → final text response
                              ↓
                 speak() reads response aloud
                              ↓
                 call.speak.ended → gather_using_ai() → loop
                              ↓
                 call.hangup → cleanup

The Tools: OpenAI-Style Function Calling

The agent has three tools, defined in the OpenAI function-calling schema:

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "check_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "lookup_order",
            "description": "Look up order status by order number",
            "parameters": {
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "check_account_balance",
            "description": "Check account balance by account number",
            "parameters": {
                "type": "object",
                "properties": {"account_id": {"type": "string"}},
                "required": ["account_id"],
            },
        },
    },
]

The model sees these tools every inference call. When the user asks "What's the weather in San Francisco?", the model returns a tool_calls array containing check_weather with {"city": "San Francisco"}. Your code runs execute_function("check_weather", {"city": "San Francisco"}), returns the JSON result, and re-sends the conversation to the model. The model then synthesizes a spoken answer: "The weather in San Francisco is 72°F and partly cloudy with 45% humidity."

The mock implementations in execute_function are intentionally simple — replace them with real API calls to your weather provider, order management system, or billing platform:

def execute_function(name, args):
    if name == "check_weather":
        return json.dumps({"city": args.get("city"), "temp": "72F",
                           "condition": "Partly cloudy", "humidity": "45%"})
    elif name == "lookup_order":
        return json.dumps({"order_id": args.get("order_id"), "status": "shipped",
                           "eta": "June 20", "carrier": "FedEx"})
    elif name == "check_account_balance":
        return json.dumps({"account_id": args.get("account_id"), "balance": "$1,234.56",
                           "due_date": "July 1"})
    return json.dumps({"error": "Unknown function"})

The Conversation Loop

The webhook handler is the heartbeat. Each event transitions the state machine:

@app.route("/webhooks/voice", methods=["POST"])
def handle_voice():
    # Verify the Telnyx Ed25519 signature before trusting the event.
    try:
        client.webhooks.unwrap(request.get_data(as_text=True), headers=dict(request.headers))
    except Exception:
        return jsonify({"error": "invalid signature"}), 401

    payload = request.get_json()
    data = payload.get("data", {})
    p = data.get("payload", {})
    event_type = data.get("event_type")
    ccid = p.get("call_control_id")
    call = active_calls.get(ccid)

    if event_type == "call.initiated" and p.get("direction") == "incoming":
        active_calls[ccid] = {
            "caller": p.get("from"),
            "conversation": [{"role": "system", "content": SYSTEM_PROMPT}],
            "_ts": time.time(),
        }
        client.calls.actions.answer(ccid)
        return jsonify({"status": "answering"}), 200

    elif event_type == "call.answered":
        client.calls.actions.speak(ccid, payload=GREETING, voice=VOICE, language="en-US")
        return jsonify({"status": "greeting"}), 200

    elif event_type == "call.speak.ended" and call:
        if call.get("processed"):
            call["processed"] = False
        client.calls.actions.gather_using_ai(
            ccid,
            parameters={
                "type": "object",
                "properties": {
                    "user_request": {
                        "type": "string",
                        "description": "What the caller said — their full spoken request.",
                    }
                },
                "required": ["user_request"],
            },
            voice=VOICE,
            language="en-US",
            user_response_timeout_ms=15000,
        )
        return jsonify({"status": "listening"}), 200

    elif event_type == "call.ai_gather.ended" and call:
        if call.get("processed"):
            return jsonify({"status": "ok"}), 200
        call["processed"] = True

        result = p.get("result", {})
        speech = result.get("user_request", "") if isinstance(result, dict) else ""

        if not speech:
            for msg in reversed(p.get("message_history", [])):
                if msg.get("role") == "user":
                    speech = msg.get("content", "")
                    break

        if not speech:
            client.calls.actions.speak(ccid, payload=REPROMPT, voice=VOICE, language="en-US")
            return jsonify({"status": "reprompting"}), 200

        call["conversation"].append({"role": "user", "content": speech})
        response = call_inference(call["conversation"])
        call["conversation"].append({"role": "assistant", "content": response})

        client.calls.actions.speak(ccid, payload=response, voice=VOICE, language="en-US")
        return jsonify({"status": "responding"}), 200

    elif event_type == "call.hangup":
        active_calls.pop(ccid, None)
        return jsonify({"status": "ended"}), 200

    return jsonify({"status": "ok"}), 200

Three things to note. First, the signature verification via client.webhooks.unwrap() — never trust an unverified webhook. Second, the call["processed"] dedup guard — Telnyx retries webhooks, and without it you would speak the same response twice. Third, the call.ai_gather.ended handler extracts speech from result.user_request with a message_history fallback, because the gather result shape varies by SDK version.

The Inference Function: Tool-Calling Loop

The call_inference function is the AI brain. It sends the conversation to the model with the TOOLS array. If the model returns tool_calls, the function executes each one, appends the results to the conversation, and recurses. If the model returns a plain text response, that's the spoken answer.

def call_inference(messages, max_tokens=300, _depth=0, _max_depth=5):
    if _depth >= _max_depth:
        return "I'm having trouble processing that request right now."

    payload = {
        "model": AI_MODEL,
        "messages": messages,
        "temperature": 0.5,
        "tools": TOOLS,
    }
    try:
        resp = requests.post(
            INFERENCE_URL,
            headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
                     "Content-Type": "application/json"},
            json=payload,
            timeout=30,
        )
    except Exception as e:
        app.logger.error("Inference request failed: %s", e)
        return "I couldn't reach the AI service just now. Please try again."

    try:
        resp.raise_for_status()
    except Exception as e:
        app.logger.error("Inference HTTP error: %s — %s", e, resp.text[:200])
        return "The AI service returned an error. Please try again."

    choice = resp.json()["choices"][0]
    msg = choice["message"]

    if msg.get("tool_calls"):
        for tc in msg["tool_calls"]:
            fn = tc["function"]
            try:
                fn_args = json.loads(fn.get("arguments", "{}"))
            except json.JSONDecodeError:
                fn_args = {}
            result = execute_function(fn["name"], fn_args)
            messages.append(msg)
            messages.append({"role": "tool", "tool_call_id": tc["id"], "content": result})
        return call_inference(messages, max_tokens, _depth=_depth + 1, _max_depth=_max_depth)

    return _strip_fences(msg["content"])

The _max_depth=5 recursion guard prevents infinite loops if the model keeps calling tools forever. The _strip_fences() helper strips

```json

code fences from the response so TTS doesn't read them aloud — without this, your voice agent would say "triple backslash triple json" before every answer.

What the Original Sample Got Wrong

This example was the most broken upstream sample in the set. The original app.py had six critical bugs that prevented the app from working at all. We fixed all of them in the fix PR.

Bug 1: gather() is DTMF-only. The original code called gather(input_type="speech", end_silence_timeout_secs=3, language_code="en-US"). None of those parameters are valid — gather() only collects DTMF keypresses. Passing speech params causes a TypeError. The fix is gather_using_ai(), which transcribes free-form speech and fires call.ai_gather.ended instead of call.gather.ended.

Bug 2: voice="female" is invalid. The speak() action requires voice in <Provider>.<Model>.<VoiceId> format (e.g., Telnyx.KokoroTTS.af). Passing "female" causes an API error.

Bug 3: call_inference() UnboundLocalError. If requests.post() raised an exception, the variable resp was never assigned. The next line, resp.raise_for_status(), then crashed with UnboundLocalError: local variable 'resp' referenced before assignment. The fix wraps both calls in try/except blocks and returns a spoken error string.

Bug 4: No recursion depth guard. The original call_inference() recursed on tool_calls with no max depth. A misbehaving model could loop forever. The fix adds _depth and _max_depth=5.

Bug 5: base_url not overridden. The Telnyx Python SDK reads the TELNYX_BASE_URL environment variable, which in some environments routes API calls to a proxy. The fix is explicit base_url="https://api.telnyx.com/v2" in the client constructor.

Bug 6: Markdown fences in TTS. Some models wrap JSON in

```json

fences. Without stripping, speak() reads the fences aloud. The fix is _strip_fences().

One API Key for Voice, AI, and TTS

The entire app uses a single TELNYX_API_KEY:

Voice (Call Control) — answer(), speak(), gather_using_ai() via the Telnyx SDK
AI Inference — POST /v2/ai/chat/completions via requests with Bearer auth
Text-to-speech — handled by speak() (uses the same API key, no separate TTS provider)
Webhook signature verification — client.webhooks.unwrap() validates Ed25519 signatures

No third-party speech-to-text provider. No separate LLM API key. No separate TTS provider. One network, one key, one bill.

The `gather_using_ai` vs `gather` Distinction

This is the most subtle and important distinction in the Call Control API. Two methods with similar names, completely different capabilities:

Method	Input type	Webhook event	Use case
`gather()`	DTMF only (keypad digits)	`call.gather.ended`	"Press 1 for sales, 2 for support" IVR menus
`gather_using_ai()`	Free-form speech	`call.ai_gather.ended`	Natural language voice agents
`gather_using_speak()`	DTMF + TTS prompt	`call.gather.ended`	Spoken IVR prompts with DTMF response

The upstream sample used gather() with speech params that don't exist — a classic copy-paste-from-docs mistake. The fix is gather_using_ai(), which returns transcribed speech in the result.user_request field of the call.ai_gather.ended payload.

Environment Variables

TELNYX_API_KEY=your_api_key_here
TELNYX_PUBLIC_KEY=your_public_key_here
AI_MODEL=moonshotai/Kimi-K2.6
AGENT_NUMBER=+15551234567
CONNECTION_ID=your_connection_id
PORT=5000

TELNYX_API_KEY — your Telnyx API v2 key (Portal → API Keys)
TELNYX_PUBLIC_KEY — your Telnyx public key (used for webhook signature verification)
AI_MODEL — any model on Telnyx AI Inference (default: moonshotai/Kimi-K2.6)
AGENT_NUMBER — the phone number callers dial
CONNECTION_ID — your Call Control Application ID (Portal → Call Control → Applications)
PORT — HTTP port for the Flask server

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-voice-agent-with-function-calling-python
cp .env.example .env   # add TELNYX_API_KEY, TELNYX_PUBLIC_KEY, AGENT_NUMBER, CONNECTION_ID
pip install -r requirements.txt
python app.py           # starts on http://localhost:5000

Expose your local server with ngrok and configure the webhook URL in your Call Control Application:

ngrok http 5000
# Copy the HTTPS URL → Portal → Call Control → Application → Webhook URL
# Set to: https://<id>.ngrok.io/webhooks/voice

Call your agent number. You'll hear the greeting. Ask: "What's the weather in San Francisco?" The agent will call the check_weather tool and speak the result. Ask: "Where is order 12345?" The agent will call lookup_order. Ask: "What's my account balance for account 67890?" The agent will call check_account_balance. Every answer is backed by real function output, not LLM hallucination.

Key links:

Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/ai-voice-agent-with-function-calling-python
Telnyx Portal: https://portal.telnyx.com

Browser Phone Calls with Real-Time AI Coaching

Harpreet Singh Seehra — Thu, 06 Aug 2026 13:39:27 +0000

Click-to-Call WebRTC with AI Assist — a split-screen browser app where you make real phone calls from your browser and get live AI coaching tips in a sidebar. WebRTC telephony credentials, browser SpeechRecognition, and Telnyx AI Inference. One API key for calls and AI. No phone needed.

The Problem: Sales Calls Without Feedback

You are on a sales call. You are pitching, handling objections, trying to close. You have no idea if you are talking too fast, missing buying signals, or failing to ask for the next step. Your manager is not on the line. There is no coach whispering in your ear. You are flying blind.

The existing solutions are clunky. Call recording platforms store the audio for later review, but "later" means after the deal is lost. AI note-takers join as a third party on the call, which creeps out prospects and sometimes breaks compliance. Real-time coaching tools exist, but they require integrations with your dialer, your CRM, and your conferencing platform. Three vendors, three API keys, three bills.

The Click-to-Call WebRTC with AI Assist example fixes this in 75 lines of Python. Open a browser tab. Enter a phone number. Click Call. You are talking to a real phone number from your browser, and an AI coach is watching your transcript and feeding you actionable tips every 8 seconds in a live sidebar.

What It Does

Open http://localhost:5000 in your browser. You see a split-screen UI: a call panel on the left and an AI coaching sidebar on the right.

Enter a phone number and click Call. The backend creates a WebRTC telephony credential via POST /v2/telephony_credentials. The frontend uses the Telnyx WebRTC SDK to connect via SIP. The WebRTC call connects — you are now talking to a real phone number from your browser, no desk phone, no softphone app, just the browser.

While you talk, the browser SpeechRecognition API transcribes your speech in real time. Every 8 seconds, the transcript is sent to the /coaching endpoint, which calls Telnyx AI Inference and returns one actionable coaching tip. The tip appears in the right sidebar with a timestamp — live AI coaching during the call.

Click Hang Up to end. The call disconnects, transcription stops, coaching stops.

Step	What happens	Tech
1	User clicks Call	Frontend sends number to backend
2	Backend creates WebRTC credential	`POST /v2/telephony_credentials`
3	Frontend connects via WebRTC	`@telnyx/webrtc` SDK, SIP registration
4	Call connects	WebRTC audio to PSTN
5	Browser transcribes speech	`SpeechRecognition` API, real-time
6	Every 8s, transcript sent to AI	`POST /v2/ai/chat/completions`
7	AI returns coaching tip	Model: `moonshotai/Kimi-K2.6`
8	Tip appears in sidebar	DOM update with timestamp
9	User clicks Hang Up	WebRTC disconnect, cleanup

The Architecture

Everything lives in one Flask file (~75 lines) and one HTML template. No database, no Redis, no WebSocket server. The browser handles speech recognition. The backend handles WebRTC credentials and AI inference.

Browser opens http://localhost:5000
        ↓
User enters number, clicks Call
        ↓
Frontend POST /webrtc/token → Backend creates telephony credential
        ↓
Backend returns SIP username, password, caller_number
        ↓
Frontend TelnyxRTC client.connect() via WebRTC
        ↓
WebRTC call connects to PSTN number
        ↓
Browser SpeechRecognition starts → live transcript in left panel
        ↓
Every 8s: transcript POST /coaching → AI Inference → coaching tip
        ↓
Tip appears in right sidebar with timestamp
        ↓
User clicks Hang Up → client.disconnect() → cleanup

The Backend: Flask in 75 Lines

The backend has four endpoints. Here is the full app.py:

import os, time, requests
from flask import Flask, request, jsonify, render_template

app = Flask(__name__)

TELNYX_API_KEY = os.getenv("TELNYX_API_KEY")
AI_MODEL = os.getenv("AI_MODEL", "moonshotai/Kimi-K2.6")
WEBRTC_CONNECTION_ID = os.getenv("WEBRTC_CONNECTION_ID")
CALLER_NUMBER = os.getenv("CALLER_NUMBER")
INFERENCE_URL = "https://api.telnyx.com/v2/ai/chat/completions"

@app.route("/")
def index():
    return render_template("index.html")

@app.route("/webrtc/token", methods=["POST"])
def webrtc_token():
    """Create a WebRTC telephony credential and return SIP credentials."""
    resp = requests.post(
        "https://api.telnyx.com/v2/telephony_credentials",
        headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
                 "Content-Type": "application/json"},
        json={"connection_id": WEBRTC_CONNECTION_ID,
              "expires_secs": 3600},
        timeout=10)
    resp.raise_for_status()
    data = resp.json()["data"]
    return jsonify({
        "sip_username": data["sip_username"],
        "sip_password": data["sip_password"],
        "caller_number": CALLER_NUMBER
    })

@app.route("/coaching", methods=["POST"])
def coaching():
    """Send transcript to AI Inference and return a coaching tip."""
    transcript = request.json.get("transcript", "")
    if not transcript:
        return jsonify({"tip": "Start speaking to get coaching tips."})

    messages = [
        {"role": "system", "content":
            "You are a sales coach. Review the call transcript and give ONE "
            "actionable tip to improve the call. Be specific, concise, and "
            "practical. Focus on: asking better questions, handling objections, "
            "closing techniques, or tone adjustments. Return only the tip, no preamble."},
        {"role": "user", "content": f"Call transcript so far:\n{transcript}"}
    ]

    resp = requests.post(INFERENCE_URL,
        headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
                 "Content-Type": "application/json"},
        json={"model": AI_MODEL, "messages": messages,
              "max_tokens": 150, "temperature": 0.7},
        timeout=15)
    resp.raise_for_status()
    tip = resp.json()["choices"][0]["message"]["content"]
    return jsonify({"tip": tip, "timestamp": time.strftime("%H:%M:%S")})

@app.route("/health")
def health():
    return jsonify({"status": "ok"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.getenv("PORT", 5000)))

That is the entire backend. Four endpoints, one inference call, one credential creation call. No session management, no database, no background workers.

WebRTC Telephony Credentials

The /webrtc/token endpoint creates a temporary SIP credential via POST /v2/telephony_credentials:

resp = requests.post(
    "https://api.telnyx.com/v2/telephony_credentials",
    headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
             "Content-Type": "application/json"},
    json={"connection_id": WEBRTC_CONNECTION_ID,
          "expires_secs": 3600},
    timeout=10)

The connection_id links the credential to your Telnyx WebRTC connection (configured in the portal). The credential expires in 1 hour. The response contains sip_username and sip_password, which the frontend uses to register with Telnyx's SIP server over WebRTC.

This is the key abstraction: Telnyx handles the SIP infrastructure, the STUN/TURN servers, the NAT traversal, and the PSTN gateway. Your code just asks for a credential and passes it to the WebRTC SDK.

AI Coaching Endpoint

The /coaching endpoint sends the transcript to Telnyx AI Inference and returns a coaching tip:

messages = [
    {"role": "system", "content":
        "You are a sales coach. Review the call transcript and give ONE "
        "actionable tip to improve the call. Be specific, concise, and "
        "practical. Focus on: asking better questions, handling objections, "
        "closing techniques, or tone adjustments. Return only the tip, no preamble."},
    {"role": "user", "content": f"Call transcript so far:\n{transcript}"}
]

resp = requests.post(INFERENCE_URL,
    headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
             "Content-Type": "application/json"},
    json={"model": AI_MODEL, "messages": messages,
          "max_tokens": 150, "temperature": 0.7},
    timeout=15)

The system prompt is specific: one tip, actionable, no preamble. Temperature is 0.7 — high enough for creative, varied coaching tips, low enough that they stay relevant. Max tokens 150 — enough for a sentence or two, not enough for the model to ramble.

The default model is moonshotai/Kimi-K2.6, but you can override via the AI_MODEL environment variable. Any model available on Telnyx AI Inference works.

The Frontend: Browser WebRTC + SpeechRecognition

The frontend is a single HTML template served by Flask. It loads the Telnyx WebRTC SDK from CDN and uses the browser's native SpeechRecognition API.

WebRTC Connection with TelnyxRTC

<script src="https://unpkg.com/@telnyx/webrtc@2/dist/telnyx.js"></script>
<script>
const client = new TelnyxRTC({
    login: sipUsername,
    password: sipPassword,
    ringtoneFile: null,
    ringbackFile: null
});

client.on('telnyx.ready', () => {
    statusBadge.textContent = 'Ready';
    statusBadge.className = 'badge ready';
});

client.on('telnyx.error', (error) => {
    statusBadge.textContent = 'Error: ' + error.message;
});

async function startCall() {
    const resp = await fetch('/webrtc/token', {method: 'POST'});
    const creds = await resp.json();
    sipUsername = creds.sip_username;
    sipPassword = creds.sip_password;
    callerNumber = creds.caller_number;

    client.login = sipUsername;
    client.password = sipPassword;
    client.connect();

    client.on('telnyx.ready', () => {
        const call = client.newCall({
            destinationNumber: document.getElementById('phoneNumber').value,
            callerNumber: callerNumber
        });
        activeCall = call;
        startTranscription();
        startCoachingLoop();
        startTimer();
    });
}
</script>

The TelnyxRTC client handles SIP registration, ICE negotiation, and media setup. You pass it the SIP username and password from /webrtc/token, call connect(), and wait for the telnyx.ready event. Then you create a new call with newCall({destinationNumber, callerNumber}).

Browser SpeechRecognition for Live Transcript

const recognition = new (window.SpeechRecognition || window.webkitSpeechRecognition)();
recognition.continuous = true;
recognition.interimResults = true;
recognition.lang = 'en-US';

let transcript = '';

recognition.onresult = (event) => {
    let interim = '';
    for (let i = event.resultIndex; i < event.results.length; i++) {
        const result = event.results[i];
        if (result.isFinal) {
            transcript += result[0].transcript + ' ';
            document.getElementById('transcript').textContent = transcript;
        } else {
            interim += result[0].transcript;
        }
    }
    document.getElementById('interim').textContent = interim;
};

function startTranscription() {
    transcript = '';
    recognition.start();
}

The SpeechRecognition API is built into Chrome and Safari. It streams partial results in real time (interimResults: true) and finalizes phrases when the speaker pauses. The transcript accumulates in a string that is displayed in the left panel and sent to the coaching endpoint every 8 seconds.

The AI Coaching Loop

let coachingInterval;

function startCoachingLoop() {
    coachingInterval = setInterval(async () => {
        if (!transcript.trim()) return;
        const resp = await fetch('/coaching', {
            method: 'POST',
            headers: {'Content-Type': 'application/json'},
            body: JSON.stringify({transcript})
        });
        const data = await resp.json();
        addCoachingTip(data.tip, data.timestamp);
    }, 8000);
}

function addCoachingTip(tip, timestamp) {
    const sidebar = document.getElementById('coachingSidebar');
    const div = document.createElement('div');
    div.className = 'coaching-tip';
    div.innerHTML = `<span class="timestamp">${timestamp}</span> ${tip}`;
    sidebar.appendChild(div);
    sidebar.scrollTop = sidebar.scrollHeight;
}

Every 8 seconds, the current transcript is sent to /coaching. The AI reviews the conversation so far and returns one tip. The tip is appended to the sidebar with a timestamp. The sidebar auto-scrolls so the latest tip is always visible.

The 8-second interval is a tradeoff. Shorter intervals mean more responsive coaching but more API calls and higher latency (the AI needs context to give good tips). Longer intervals mean fewer calls but delayed feedback. Eight seconds hits the sweet spot for sales calls: enough context for meaningful tips, frequent enough to feel live.

Call Timer and Status Badges

let callStartTime;
let timerInterval;

function startTimer() {
    callStartTime = Date.now();
    timerInterval = setInterval(() => {
        const elapsed = Math.floor((Date.now() - callStartTime) / 1000);
        const mins = String(Math.floor(elapsed / 60)).padStart(2, '0');
        const secs = String(elapsed % 60).padStart(2, '0');
        document.getElementById('timer').textContent = `${mins}:${secs}`;
    }, 1000);
}

The call timer starts when the call connects and updates every second. Status badges show Ready, Connecting, On Call, and Ended states. These are small UX touches that make the app feel like a real dialer.

One API Key for WebRTC and AI

The entire app uses a single TELNYX_API_KEY:

WebRTC credentials — POST /v2/telephony_credentials (via requests.post with Bearer token)
AI Inference — POST /v2/ai/chat/completions (via requests.post with Bearer token)
Call delivery — handled by Telnyx's PSTN gateway (no additional API calls)

The SDK handles the WebRTC media layer. The two requests.post calls handle credentials and AI directly. No third-party transcription service, no separate LLM provider, no conferencing platform. One network, one key, one bill.

Environment Variables

TELNYX_API_KEY=your_api_key_here
AI_MODEL=moonshotai/Kimi-K2.6
WEBRTC_CONNECTION_ID=your_connection_id
CALLER_NUMBER=+12155551234
PORT=5000

TELNYX_API_KEY — your Telnyx API v2 key
AI_MODEL — any model on Telnyx AI Inference (default: moonshotai/Kimi-K2.6)
WEBRTC_CONNECTION_ID — the connection ID from your Telnyx Portal WebRTC configuration
CALLER_NUMBER — the caller ID for outbound calls (must be a Telnyx number on your account)
PORT — HTTP port (default: 5000)

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/click-to-call-webrtc-with-ai-assist-python
cp .env.example .env   # add TELNYX_API_KEY, WEBRTC_CONNECTION_ID, CALLER_NUMBER
pip install -r requirements.txt
python app.py           # starts on http://localhost:5000

Open http://localhost:5000 in Chrome or Safari. Enter a phone number. Click Call. Start talking. Watch the coaching tips appear.

Key links:

Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/click-to-call-webrtc-with-ai-assist-python
Telnyx Portal: https://portal.telnyx.com
WebRTC docs: https://developers.telnyx.com/docs/voice/webrtc
AI Inference docs: https://developers.telnyx.com/docs/inference

From Phone Call to Formatted Email in 80 Lines of Python — AI Voice Memo Cleanup with Telnyx

Harpreet Singh Seehra — Wed, 05 Aug 2026 10:44:52 +0000

AI Voice Memo to Email — an 80-line Flask webhook that answers a phone call, gathers a spoken memo, runs it through AI Inference to clean up grammar and extract structure, and delivers a formatted email. One API key for voice, AI, and messaging. No third-party services.

The Voice Memo Problem

Voice memos are the fastest way to capture a thought — you speak, you're done. But what you get is a rambling audio blob that nobody (including you) wants to read later. The raw transcript is worse: no punctuation, false starts, filler words, and no structure. You still have to manually clean it up before it's useful as an email, a status update, or a meeting summary.

The existing solutions split the problem across multiple services. A transcription service converts audio to text. An LLM API cleans up the text. An email service sends the result. Three vendors, three API keys, three bills, three points of failure.

The AI Voice Memo to Email example does all of it on one network — Telnyx Call Control handles the phone call, Telnyx AI Inference cleans up the transcript, and Telnyx Messaging delivers the email. One API key. One Flask file. About 80 lines of Python.

What It Does

You call a Telnyx number. The app answers, speaks a greeting, and starts listening. You dictate your memo — a status update, a meeting summary, a bug report, whatever — and press # when you're done. The app sends the transcript to AI Inference with a prompt that returns structured JSON: a subject line, a formatted body, and a list of action items. The app sends that as an email to your default address and confirms back on the call: "Memo saved and emailed. Subject: [inferred subject]. Goodbye!"

Step	Event	Action
1	`call.initiated` (incoming)	Answer the call, create session
2	`call.answered`	TTS: "Voice memo. Speak your memo after the tone. Press pound when finished."
3	`call.speak.ended`	Start speech gather (120s timeout, `#` terminates)
4	`call.gather.ended`	Send transcript to AI Inference → get structured JSON → send email → TTS confirmation
5	`call.hangup`	Clean up session

The memo is also stored in memory and accessible via GET /memos — so even if email delivery isn't configured, the formatted memo is still retrievable.

The Architecture

Everything lives in one Flask file. No database, no Redis, no Celery. Call state is tracked in an in-memory dict keyed by call_control_id. Memos are stored in a list. A background thread cleans up expired sessions every 5 minutes (1-hour TTL).

Caller dials your Telnyx number
        ↓
Telnyx sends call.initiated webhook → /webhooks/voice
        ↓
app calls answer() → creates session in active_calls[ccid]
        ↓
Telnyx sends call.answered → app calls speak() with greeting
        ↓
Telnyx sends call.speak.ended → app calls gather(input_type="speech", terminating_digit="#")
        ↓
Caller dictates memo, presses #
        ↓
Telnyx sends call.gather.ended with speech transcript
        ↓
app sends transcript to AI Inference → gets JSON {subject, body, action_items}
        ↓
app sends email via Telnyx Messaging API
        ↓
app calls speak() with confirmation: "Memo saved and emailed. Subject: X. Goodbye!"
        ↓
Telnyx sends call.hangup → app removes session

The Call Flow State Machine

The webhook handler is a state machine driven by Telnyx events. Each event triggers the next action:

@app.route("/webhooks/voice", methods=["POST"])
def handle_voice():
    # Verify the Telnyx Ed25519 signature before trusting the event.
    try:
        client.webhooks.unwrap(request.get_data(as_text=True), headers=dict(request.headers))
    except Exception:
        return jsonify({"error": "invalid signature"}), 401

    payload = request.get_json()
    event_type = payload.get("data", {}).get("event_type")
    data = payload.get("data", {})
    p = data.get("payload", {})
    ccid = p.get("call_control_id")

    if event_type == "call.initiated" and p.get("direction") == "incoming":
        active_calls[ccid] = {"caller": p.get("from"), "raw_text": [], "start": time.time()}
        client.calls.actions.answer(ccid)
        return jsonify({"status": "answering"}), 200

    elif event_type == "call.answered":
        client.calls.actions.speak(ccid,
            payload="Voice memo. Speak your memo after the tone. Press pound when finished.",
            voice="female", language_code="en-US")
        return jsonify({"status": "greeting"}), 200

    elif event_type == "call.speak.ended":
        client.calls.actions.gather(ccid,
            input_type="speech", end_silence_timeout_secs=5, timeout_secs=120,
            language_code="en-US", terminating_digit="#")
        return jsonify({"status": "recording"}), 200

    elif event_type == "call.gather.ended":
        call = active_calls.get(ccid)
        speech = p.get("speech", {}).get("result", "")
        if call and speech:
            call["raw_text"].append(speech)
            # ... AI cleanup + email + confirmation
        return jsonify({"status": "processed"}), 200

    elif event_type == "call.hangup":
        active_calls.pop(ccid, None)
        return jsonify({"status": "ended"}), 200

The state machine has five transitions, one per event. The call.initiated handler checks direction == "incoming" to avoid processing outbound call legs. The call.speak.ended handler is what advances from greeting to gathering — Telnyx fires this event when TTS playback finishes, so you know the caller has heard the greeting before the gather starts.

The gather uses end_silence_timeout_secs=5 — if the caller stops speaking for 5 seconds, the gather ends automatically. The timeout_secs=120 caps the total gather at 2 minutes. The terminating_digit="#" lets the caller explicitly signal "I'm done" by pressing pound.

AI-Powered Memo Cleanup

The core of the app is a single inference call that turns rambling speech into structured JSON:

def call_inference(messages, max_tokens=400):
    resp = requests.post(INFERENCE_URL,
        headers={"Authorization": f"Bearer {TELNYX_API_KEY}", "Content-Type": "application/json"},
        json={"model": AI_MODEL, "messages": messages,
              "max_tokens": max_tokens, "temperature": 0.3},
        timeout=15)
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

The system prompt asks for three fields — subject, body, and action_items:

formatted = call_inference([
    {"role": "system", "content":
        "Clean up this voice memo into a well-formatted email. "
        "Fix grammar, add structure (paragraphs, bullets if needed). "
        "Return JSON: subject (string, inferred from content), "
        "body (string, the formatted memo), "
        "action_items (list of strings)."},
    {"role": "user", "content": speech}
])
memo = json.loads(formatted)

Temperature is 0.3 — low enough that the same memo produces roughly the same output every time, but high enough that the AI can infer a reasonable subject line from the content. The max_tokens=400 cap is sufficient for a typical voice memo.

If the AI response isn't valid JSON, the except block saves the raw speech and speaks a simpler confirmation — the caller still gets their memo saved, just without the email:

except Exception:
    memos.append({"raw": speech, "caller": call["caller"],
                  "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ")})
    client.calls.actions.speak(ccid, payload="Memo saved. Goodbye!",
        voice="female", language_code="en-US")

Graceful degradation — the call is never wasted. If AI fails, the raw transcript is preserved. If email fails, the formatted memo is preserved. The caller always gets a confirmation.

Email Delivery via Telnyx Messaging

After the memo is formatted, the app sends it as an email through the Telnyx Messaging API:

def send_email(to, subject, body):
    try:
        requests.post("https://api.telnyx.com/v2/messages",
            headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
                     "Content-Type": "application/json"},
            json={"from": {"email_address": f"memo@{MEMO_NUMBER.replace('+','')}.telnyx.com"},
                  "to": [{"email_address": to}],
                  "subject": subject, "body": body, "type": "email"},
            timeout=15)
    except Exception as e:
        app.logger.error("Email send failed: %s", e)

The same TELNYX_API_KEY that answers the call and runs the AI inference also sends the email — one key, one bill, one network. The email send is wrapped in a try/except because email delivery may require additional Telnyx setup. If it fails, the memo is still saved and retrievable via GET /memos.

Webhook Signature Verification

Every Telnyx webhook is signed with an Ed25519 key. The app verifies the signature before processing the event:

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

The webhooks.unwrap() method from the Telnyx Python SDK handles the Ed25519 verification internally — it reads the telnyx-signature-ed25519 and telnyx-timestamp headers, reconstructs the signed payload, and verifies the signature against the public key. The raw body is verified, not the parsed JSON — because JSON parsing is not canonical, and the signature would fail.

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-voice-memo-to-email-python
cp .env.example .env   # add TELNYX_API_KEY, MEMO_NUMBER, DEFAULT_EMAIL
pip install -r requirements.txt
python app.py           # starts on http://localhost:5000

Then:

ngrok http 5000

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

Call your Telnyx number. Speak your memo. Press #. Check your email.

Check saved memos:

curl http://localhost:5000/memos | python3 -m json.tool

Key links:

Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/ai-voice-memo-to-email-python
Telnyx Portal: https://portal.telnyx.com
Call Control docs: https://developers.telnyx.com/docs/voice/call-control
AI Inference docs: https://developers.telnyx.com/docs/inference
Messaging docs: https://developers.telnyx.com/docs/messaging

Build a Real-Time AI Translation Bridge for Phone Calls in 141 Lines of Python

Harpreet Singh Seehra — Tue, 28 Jul 2026 10:04:01 +0000

Connect two callers who speak different languages on the same phone call. Each caller speaks in their own language and hears the other in their own language — live, on the same call, no interpreter in the loop.

No Google Translate. No AWS Translate. No DeepL. No third-party translation API. Just Telnyx Voice Call Control (for the phone call) and Telnyx AI Inference (for the translation). Same API key, same platform, same billing.

141 lines of Python. Here's how it works.

What It Does

You POST two phone numbers and two languages:

curl -X POST http://localhost:5000/bridge \
  -H "Content-Type: application/json" \
  -d '{"number_a": "+13125550001", "lang_a": "English",
       "number_b": "+5215550002", "lang_b": "Spanish"}'

The app calls caller A. When they answer, it calls caller B. When B answers, the translation bridge goes live:

Caller A speaks in English
Telnyx transcribes the speech → sends to your webhook
Your app sends the transcript to AI Inference: "Translate from English to Spanish"
The translated text is spoken to caller B via TTS in Spanish
Caller B speaks in Spanish → transcribed → translated to English → spoken to caller A
Loop until either caller hangs up

Both callers stay on the same phone call. Neither installs anything. Translation happens server-side, transparently.

The Architecture

Everything lives in one Flask file. No database, no Redis, no message queue. Bridge state lives in an in-memory dict, keyed by a bridge ID that travels in Telnyx's client_state field on every webhook event.

POST /bridge → call A → A answers → call B → B answers → bridge active
     ↓
gather A speech → translate → TTS to B → gather B speech → translate → TTS to A → loop
     ↓
hangup → hangup other caller → bridge ended

The client_state field on every Telnyx action carries the bridge ID and which side (a or b) the call belongs to. When a webhook fires, you decode the base64 state, look up the bridge, and know exactly which call leg you're handling. No session store — the call is self-describing.

The Language Code Bug

The original sample had all TTS and speech recognition hardcoded to language_code="en-US" — even for Spanish. The call "worked" (audio flowed, no errors) but the translation was useless:

Spanish TTS with en-US: the TTS engine reads Spanish words with English phonemes. Unintelligible.
Spanish STT with en-US: the STT engine listens for English sounds. Spanish speech recognition drops.

The fix — a language name → BCP-47 code mapping:

LANG_CODES = {
    "english": "en-US", "spanish": "es-US", "french": "fr-FR",
    "german": "de-DE", "italian": "it-IT", "portuguese": "pt-BR",
    "hindi": "hi-IN", "arabic": "ar-SA", "chinese": "zh-CN",
    "japanese": "ja-JP", "korean": "ko-KR", "russian": "ru-RU",
}

def lang_code(lang):
    return LANG_CODES.get(lang.lower(), "en-US")

Now TTS uses the target language (Spanish text → es-US pronunciation) and STT uses the speaker's language (Spanish speech → es-US recognition).

The Translation Call

Uses Telnyx AI Inference — the OpenAI-compatible /v2/ai/chat/completions endpoint:

def translate(text, from_lang, to_lang):
    resp = requests.post(INFERENCE_URL,
        headers={"Authorization": f"Bearer {TELNYX_API_KEY}"},
        json={
            "model": AI_MODEL,
            "messages": [
                {"role": "system",
                 "content": f"Translate from {from_lang} to {to_lang}. "
                           "Return ONLY the translation, nothing else."},
                {"role": "user", "content": text}
            ],
            "max_tokens": 200, "temperature": 0.1
        })
    return resp.json()["choices"][0]["message"]["content"]

Temperature 0.1 for deterministic translations. 200 max tokens for conversational phrases. The model defaults to moonshotai/Kimi-K2.6 — swap to any model in the Telnyx catalog via the AI_MODEL env var.

No separate translation API account. The same Telnyx API key that makes the phone call does the translation.

Webhook Signature Verification

Telnyx signs every webhook with an Ed25519 key. The app verifies before trusting:

@app.route("/webhooks/voice", methods=["POST"])
def handle_voice():
    try:
        client.webhooks.unwrap(
            request.get_data(as_text=True),
            headers=dict(request.headers)
        )
    except Exception:
        return jsonify({"error": "invalid signature"}), 401

Without this, anyone who knows your webhook URL can inject fake call events. The telnyx SDK's webhooks.unwrap() handles the Ed25519 verification.

The Call Loop

Each Telnyx event triggers the next action:

# Caller finished speaking → we have a transcript
elif event_type == "call.gather.ended" and bridge:
    speech = p.get("speech", {}).get("result", "")
    if speech and bridge.get("state") == "active":
        from_lang = bridge[f"lang_{side}"]
        to_lang = bridge["lang_b"] if side == "a" else bridge["lang_a"]
        other_ccid = bridge["ccids"].get("b" if side == "a" else "a")
        translated = translate(speech, from_lang, to_lang)
        if other_ccid:
            client.calls.actions.speak(other_ccid, payload=translated,
                voice="female", language_code=lang_code(to_lang))
        client.calls.actions.gather(ccid, input_type="speech",
            language_code=lang_code(bridge[f"lang_{side}"]))

When either caller hangs up, the other is hung up too — a translation bridge with one caller is pointless:

elif event_type == "call.hangup" and bridge:
    other_side = "b" if side == "a" else "a"
    other_ccid = bridge.get("ccids", {}).get(other_side)
    if other_ccid:
        requests.post(f".../calls/{other_ccid}/actions/hangup", ...)
    bridge["state"] = "ended"

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-real-time-translation-bridge-python
cp .env.example .env   # TELNYX_API_KEY, TELNYX_PUBLIC_KEY, BRIDGE_NUMBER, CONNECTION_ID
pip install -r requirements.txt
python app.py

Then:

ngrok http 5000

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

Trigger a bridge:

curl -X POST http://localhost:5000/bridge \
  -H "Content-Type: application/json" \
  -d '{"number_a": "+13125550001", "lang_a": "English",
       "number_b": "+5215550002", "lang_b": "Spanish"}'

Call from two phones. Speak in different languages. Hear real-time translation.

Key links:

Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/ai-real-time-translation-bridge-python
Telnyx Portal: https://portal.telnyx.com
Call Control docs: https://developers.telnyx.com/docs/voice/call-control
AI Inference docs: https://developers.telnyx.com/docs/inference

Route Phone Calls by Geography at the Carrier Edge — One Number, Three Regions, Three Compliant Experiences

Harpreet Singh Seehra — Mon, 27 Jul 2026 13:51:05 +0000

Route Phone Calls by Geography at the Carrier Edge

Edge Geo Smart Router — a 256-line Flask webhook that detects caller region from the E.164 number prefix and applies different language, voice, recording, and GDPR consent policy per region. US callers get English AI, LATAM gets Spanish, EU gets a consent prompt before any recording starts.

The Problem With Global Phone Numbers

A single phone number can ring from anywhere. A US toll-free number might receive calls from New York, São Paulo, London, and Berlin in the same hour. Traditional IVRs handle this with a single language menu — "Press 1 for English, 2 for Spanish" — which puts the burden on the caller and guarantees nothing about compliance.

The compliance part is the hard one. If you record a call from an EU caller without explicit consent, you've violated GDPR. If you don't record a US call, you lose your quality assurance trail. If you greet a LATAM caller in English, you've already lost them.

The Edge Geo Smart Router solves this by making the routing decision at the carrier edge — before the first word of greeting is spoken. The caller's number prefix (+1 for US, +55 for Brazil, +44 for UK) is the geo signal. No geoip lookup, no external database, no round-trip to a backend. The webhook receives call.initiated, reads the from field, classifies the region, and applies the right config for the rest of the call.

What It Does

One phone number. Three regions. Three compliant experiences.

Region	Language	Voice	Recording	Consent
US	en-US	AWS.Polly.Joanna-Neural	✅ auto-start	❌ not required
LATAM	es-MX	AWS.Polly.Lupe-Neural	✅ auto-start	❌ not required
EU	en-GB	AWS.Polly.Amy-Neural	⏸️ after consent	✅ DTMF prompt
DEFAULT	en-US	AWS.Polly.Joanna-Neural	✅ auto-start	❌ not required

US and LATAM callers get straight into a conversation with AI in their language. EU callers hear: "This call will be recorded for quality purposes. Press 1 to consent and continue, or press 2 to proceed without recording." If they press 1, recording starts. If they press 2, the call proceeds unrecorded. Either way, GDPR is satisfied.

The Architecture

Everything lives in one 256-line Flask file. No database, no Redis, no message queue. Call state travels in Telnyx's client_state field — a base64-encoded JSON blob that Telnyx passes back to you on every webhook event for the same call.

Caller dials your number
        ↓
Telnyx sends call.initiated webhook → /webhooks/voice
        ↓
app detects region from `from` number prefix
        ↓
app calls /actions/answer with client_state = {region, step: "greeting"}
        ↓
Telnyx sends call.answered → app reads client_state
        ↓
  ┌─────────────┬─────────────┬─────────────┐
  │ US / LATAM  │ EU          │ DEFAULT     │
  │             │             │             │
  │ record_start│ gather DTMF │ record_start│
  │ gather_speak│ (consent)   │ gather_speak│
  │ (greeting)  │             │ (greeting)  │
  └─────┬───────┴──────┬──────┴──────┬──────┘
        ↓              ↓             ↓
  call.gather.ended → app reads step from client_state
        ↓
  "consent" step → did they press 1?
        ↓            → record_start + gather (greeting)
  "conversation" step → call_inference(speech, language)
        ↓                        → gather_speak (AI response)
  loop until hangup
        ↓
  call.hangup → clean up session

Region Detection: The Prefix Match

The routing logic is 12 lines of Python. No geoip, no database, no external API call.

EU_PREFIXES = ["+33", "+34", "+39", "+44", "+49", "+31", "+32", "+43",
               "+45", "+46", "+47", "+48", "+351", "+353", "+358",
               "+352", "+356", "+370", "+371", "+372", "+420", "+421"]
LATAM_PREFIXES = ["+52", "+55", "+54", "+56", "+57", "+51", "+58",
                  "+53", "+506", "+507", "+502", "+503", "+504", "+505"]

def detect_region(phone):
    if not phone:
        return "DEFAULT"
    for prefix in EU_PREFIXES:
        if phone.startswith(prefix):
            return "EU"
    for prefix in LATAM_PREFIXES:
        if phone.startswith(prefix):
            return "LATAM"
    if phone.startswith("+1"):
        return "US"
    return "DEFAULT"

EU is checked first because some European country codes share the +1 prefix with NANP. By checking EU first, we avoid misrouting through the US path.

Per-Region Config

Each region has its own voice, language, greeting, recording policy, and consent requirement.

REGION_CONFIG = {
    "US":     {"language": "en-US", "voice": "AWS.Polly.Joanna-Neural",
               "greeting": "Welcome. How can I help you today?",
               "requires_consent": False, "record": True},
    "LATAM":  {"language": "es-MX", "voice": "AWS.Polly.Lupe-Neural",
               "greeting": "Bienvenido. Como puedo ayudarle hoy?",
               "requires_consent": False, "record": True},
    "EU":     {"language": "en-GB", "voice": "AWS.Polly.Amy-Neural",
               "greeting": "This call will be recorded for quality purposes. "
                           "Press 1 to consent and continue, or press 2 to "
                           "proceed without recording.",
               "requires_consent": True, "record": False},
    "DEFAULT":{"language": "en-US", "voice": "AWS.Polly.Joanna-Neural",
               "greeting": "Welcome. How can I help you?",
               "requires_consent": False, "record": True}
}

The voice ID gotcha

Bare "female" and "male" voice IDs only work with service_level: "basic", which is en-US only. If you try "voice": "female" with "language": "es-MX", the call silently fails. The fix is to use full neural voice IDs — AWS.Polly.Lupe-Neural for es-MX, AWS.Polly.Amy-Neural for en-GB. Each voice supports its respective language.

The GDPR Consent Flow

EU callers get a DTMF (touch-tone) consent prompt before any recording starts. The flow is:

call.answered fires for an EU caller
App sees requires_consent: True in the region config
App calls gather_using_speak with the consent greeting, minimum_digits: 1, maximum_digits: 1
Caller presses 1 (consent) or 2 (no consent)
call.gather.ended fires with step: "consent" in client_state
If digit is "1": app calls record_start, then gather_using_speak with the actual greeting
If digit is "2": app calls gather_using_speak with the greeting, no recording

if step == "consent":
    digits = ep.get("digits", "")
    if digits == "1":
        requests.post(f".../actions/record_start",
                      headers=HEADERS, json={"format": "mp3", "channels": "dual"})
        call_sessions.get(cc_id, {})["consented"] = True
    requests.post(f".../actions/gather_using_speak",
                  json={"payload": "Thank you. How can I help you today?",
                        "voice": config["voice"], "language": config["language"],
                        "input_type": "speech", "timeout_secs": 30,
                        "client_state": encode_state({"region": region, "step": "conversation"})})

The key insight: recording only starts AFTER the caller presses 1. If they press 2, the call proceeds with AI conversation but no recording is captured. GDPR satisfied by design, not by policy.

State Management Without a Database

Every call carries its own state in Telnyx's client_state field. This is a base64-encoded JSON blob that Telnyx passes back to you on every subsequent webhook event for the same call. You set it when you call an action (answer, gather, record), and you read it when the next event fires.

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: return {}

The state contains just two fields: region (so every event handler knows which region config to apply) and step (so call.gather.ended knows whether it's handling the consent prompt or the actual conversation). No session store, no Redis, no database. The call is self-describing.

Webhook Signature Verification

Telnyx signs every webhook with an Ed25519 key. The signature is over <timestamp>|<raw body>. Verifying it before you trust the payload is non-optional — without it, anyone can POST to your webhook URL and inject fake call events.

def verify_telnyx_signature(raw_body, headers):
    if _TELNYX_VERIFY_KEY is None:
        return False
    signature = headers.get("telnyx-signature-ed25519", "")
    timestamp = headers.get("telnyx-timestamp", "")
    if not signature or not timestamp:
        return False
    try:
        if abs(time.time() - int(timestamp)) > MAX_SKEW_SECONDS:
            return False
        signed = f"{timestamp}|".encode() + raw_body
        _TELNYX_VERIFY_KEY.verify(base64.b64decode(signature), signed)
        return True
    except (InvalidSignature, ValueError, TypeError):
        return False

The 300-second skew window rejects replay attacks. The raw body is verified, not the parsed JSON — because JSON parsing is not canonical (key order, whitespace), and the signature would fail. Get the public key from Portal → Keys & Credentials.

Recording Archival to Telnyx Storage

When call.recording.saved fires, the app downloads the recording and uploads it to Telnyx Storage (S3-compatible). The key includes the region and call session ID for later retrieval.

elif event_type == "call.recording.saved":
    recording_url = ep.get("recording_urls", {}).get("mp3")
    if not (recording_url and is_telnyx_url(recording_url)):
        return jsonify({"status": "ok"})
    if s3 is None:
        app.logger.info("Recording ready but Storage credentials not configured; skipping.")
        return jsonify({"status": "ok"})
    audio = requests.get(recording_url, headers=HEADERS, timeout=30).content
    key = f"recordings/{region}/{call_session_id}.mp3"
    s3.put_object(Bucket=STORAGE_BUCKET, Key=key, Body=audio, ContentType="audio/mpeg")

Two guards here:

SSRF protection: is_telnyx_url() only fetches URLs on telnyx.com or *.telnyx.com hosts. The recording URL comes from the webhook payload, which is attacker-controllable. Without this check, a forged webhook could make your server download from any URL — including internal services.
Storage credentials ≠ API key: Telnyx Storage is S3-compatible but uses its own access/secret key pair, created under Portal → Storage → Credentials. The original version of this sample reused the Telnyx API key as both the S3 access key AND the secret key — which silently failed. The fix: separate STORAGE_ACCESS_KEY and STORAGE_SECRET_KEY environment variables. If they're not set, archival is gracefully skipped (the call still works, just no recording saved).

AI Inference: Multilingual Replies

When the caller speaks, the app sends the transcript to Telnyx AI Inference (OpenAI-compatible /v2/ai/chat/completions endpoint) with a language instruction derived from the region config.

def call_inference(prompt, language="en-US"):
    lang_instruction = "Respond in Spanish." if "es" in language else "Respond in English."
    resp = requests.post(INFERENCE_URL, headers=HEADERS, timeout=15, json={
        "model": AI_MODEL,
        "messages": [{"role": "system",
                      "content": f"You are a helpful business assistant. Be concise. {lang_instruction}"},
                     {"role": "user", "content": prompt}]
    })
    if resp.ok:
        return resp.json()["choices"][0]["message"]["content"]
    return "I apologize, I'm having trouble right now. Please try again."

The AI response is then spoken back to the caller via gather_using_speak with the region's voice and language, and the loop continues until the caller hangs up.

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/edge-geo-smart-router-python
cp .env.example .env   # add TELNYX_API_KEY, TELNYX_PUBLIC_KEY, optional Storage creds
pip install -r requirements.txt
python app.py

Then:

ngrok http 5000

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

Call your Telnyx number from:

A US number (+1) → English AI, auto-recording
A Brazil number (+55) → Spanish AI, auto-recording
A UK number (+44) → English (en-GB) AI, consent prompt first

Check the routing config:

curl http://localhost:5000/regions

Check health:

curl http://localhost:5000/health

Key links:

Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/edge-geo-smart-router-python
Telnyx Portal: https://portal.telnyx.com
Call Control docs: https://developers.telnyx.com/docs/voice/call-control
AI Inference docs: https://developers.telnyx.com/docs/inference

Build an AI Video Dubbing Pipeline with STT, LLM, and TTS on One Network

Harpreet Singh Seehra — Mon, 27 Jul 2026 12:32:56 +0000

Imagine uploading a podcast clip in English and getting back the same conversation dubbed in Spanish — same speakers, same pacing, same tone, but in a different language. No stitching together Deepgram for transcription, OpenAI for translation, and ElevenLabs for synthesis. No juggling three API keys, three billing accounts, three SDKs. One pipeline, three API calls, one network.

This is the AI Video Dubbing Pipeline — a ~280-line Python app built on Telnyx that runs the full Speech-to-Text → AI Inference → Text-to-Speech chain. Upload any audio file, pick from 15 target languages, and download the dubbed mp3 a few seconds later. The pipeline labels speakers, translates the dialogue, and synthesizes each speaker with a matched voice from a KokoroTTS voice pool.

In this walkthrough, you'll build it from scratch. Clone the repo, add your API key, and dub your first clip in minutes.

What You'll Build

An HTTP API that dubs audio files into another language:

Upload audio — POST /dub with a multipart audio file and a target language
STT transcribes — Whisper-large-v3-turbo turns the audio into timestamped segments
LLM labels + translates — a single chat-completions call assigns speaker labels (SPEAKER_0, SPEAKER_1, ...) and translates each segment to the target language
TTS synthesizes — each translated segment is rendered with a voice assigned per speaker from a KokoroTTS voice pool
Download the dub — GET /dub/<job_id>/audio streams the concatenated mp3
Side-by-side transcript — GET /dub/<job_id>/transcript returns the original and translated text per segment

The whole pipeline runs asynchronously: POST /dub returns 202 with a job_id, and the client polls GET /dub/<job_id> until status == "complete". No webhooks, no phone numbers, no ngrok — this is a pure HTTP API.

Why This Is Interesting

Most dubbing pipelines are stitched together from three separate vendors. That means three API keys, three billing relationships, three sets of rate limits, and three network round trips — the STT transcription, the LLM translation, and the TTS synthesis each cross the internet to a different provider. Latency adds up. Authentication adds up. So does the operational overhead of keeping three SDKs current.

This pipeline runs all three primitives on Telnyx. The STT, chat, and TTS endpoints share one API key, one billing account, and one network path. The STT and chat endpoints are OpenAI-compatible — drop-in for the OpenAI Python or JS SDK with the base URL changed to https://api.telnyx.com/v2. The TTS endpoint takes a Telnyx voice id (Telnyx.KokoroTTS.am_onyx, af_nova, etc.) and renders in the target language you pass in.

The interesting technical move: Telnyx STT does not perform speaker diarization. The transcription API returns segments with start, end, and text — no speaker field. Most pipelines handle this by adding a fourth vendor (a diarization service) or by accepting that all segments come back as one speaker. This app takes a different path: it sends all the transcribed segments to the LLM in a single chat call and asks the model to assign speaker labels AND translate in the same response. One model, two jobs, no extra vendor.

Prerequisites

Python 3.8+
A Telnyx account with funded balance
A Telnyx API key

That's it. No phone number, no Call Control Application, no ngrok, no webhook configuration. This is a pure HTTP API — you upload audio and poll for results.

The Architecture

Upload audio (POST /dub, multipart)
        │
        ▼
  ┌──────────────────┐
  │ STT (Whisper)     │ ── POST /v2/ai/audio/transcriptions
  │                   │    returns segments: [{start, end, text}, ...]
  └────────┬──────────┘
           │ segments
           ▼
  ┌──────────────────┐
  │ AI Inference      │ ── POST /v2/ai/chat/completions
  │ (single call)     │    returns JSON: [{id, speaker, translated}, ...]
  │ label + translate │    (LLM assigns SPEAKER_0/1/... AND translates)
  └────────┬──────────┘
           │ translated segments with speaker labels
           ▼
  ┌──────────────────┐
  │ TTS (Kokoro)      │ ── POST /v2/text-to-speech/speech  (per segment)
  │ per-segment render│    voice assigned per speaker from a 5-voice pool
  └────────┬──────────┘
           │
           ├──► GET /dub/<job_id>         (status + transcript)
           ├──► GET /dub/<job_id>/audio    (download mp3)
           └──► GET /dub/<job_id>/transcript (side-by-side text)

The pipeline runs in a background thread. POST /dub validates the request, stores the job, kicks off the thread, and returns 202 immediately. The thread runs STT → Inference → TTS, updates the job status at each step, and stores the concatenated audio on the job when done.

Step 1: Clone and Configure

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-video-dubbing-pipeline-python
cp .env.example .env
pip install -r requirements.txt

Edit .env with your Telnyx API key:

TELNYX_API_KEY=your_telnyx_api_key_here
AI_MODEL=moonshotai/Kimi-K2.6
STT_MODEL=openai/whisper-large-v3-turbo
HOST=127.0.0.1
PORT=5000

The defaults are solid. whisper-large-v3-turbo is multilingual — it transcribes any of the 16 supported source languages. If you only ever dub English source audio, you can switch to distil-whisper/distil-large-v2 for a faster, cheaper, English-only pass.

Step 2: Understand the Code

Everything lives in app.py (~280 lines). Here's what each piece does.

The Voice Pool

The app cycles through five Telnyx KokoroTTS voices, one per speaker:

VOICE_MAP = {
    "male_low": "Telnyx.KokoroTTS.am_onyx",
    "male_mid": "Telnyx.KokoroTTS.am_echo",
    "female_mid": "Telnyx.KokoroTTS.af_nova",
    "female_high": "Telnyx.KokoroTTS.af_heart",
    "neutral": "Telnyx.KokoroTTS.af_alloy",
}

When the pipeline hits a new speaker label, it assigns the next voice from the pool. A two-speaker conversation gets am_onyx and am_echo. A five-speaker panel gets all five. A sixth speaker wraps back to am_onyx. The voice ids are real — you can list every available voice with GET /v2/text-to-speech/voices on the Telnyx API.

The STT Helper

def transcribe_audio(audio_bytes, source_language="en"):
    bcp47 = STT_LANGUAGE_MAP.get(source_language, "en-US")
    resp = requests.post(f"{API}/ai/audio/transcriptions", headers={
        "Authorization": f"Bearer {TELNYX_API_KEY}"
    }, files={
        "file": ("audio.mp3", audio_bytes, "audio/mpeg")
    }, data={
        "model": STT_MODEL,
        "language": bcp47,
        "response_format": "verbose_json",
        "timestamp_granularities[]": "segment"
    }, timeout=120)
    resp.raise_for_status()
    return resp.json()

OpenAI-compatible transcription. The verbose_json response format with timestamp_granularities[]=segment returns segments with start, end, and text — exactly what the next stage needs. Note: no diarize parameter. Telnyx STT doesn't diarize. The LLM does that downstream.

The LLM Labels Speakers AND Translates in One Call

This is the move that replaces a fourth vendor:

def label_and_translate_segments(segments, source_lang, target_lang_name):
    segment_texts = [
        {"id": i, "text": s.get("text", "").strip(),
         "start": s.get("start", 0), "end": s.get("end", 0)}
        for i, s in enumerate(segments)
    ]
    prompt = (
        f"You are a dubbing assistant. Below are {len(segment_texts)} transcribed audio segments "
        f"from a conversation in {source_lang}. For EACH segment:\n"
        f"1. Assign a speaker label (SPEAKER_0, SPEAKER_1, etc.) based on conversational context.\n"
        f"2. Translate the text to {target_lang_name}, preserving tone and meaning.\n\n"
        f"Return ONLY a JSON array (no markdown fences, no explanation) where each element is:\n"
        f'{{"id": <int>, "speaker": "SPEAKER_N", "translated": "<text in {target_lang_name}>"}}\n\n'
        f"Segments:\n{json.dumps(segment_texts, ensure_ascii=False)}"
    )
    raw = inference([
        {"role": "system", "content": "You are a translation and speaker-diarization engine. You always return valid JSON arrays, never prose."},
        {"role": "user", "content": prompt}
    ], max_tokens=4000)
    # ... parse JSON, fall back to per-segment translation if the model returns non-JSON ...

One chat call. The LLM reads every segment, figures out who's speaking based on conversational context (turn-taking, topic shifts, register), assigns SPEAKER_0, SPEAKER_1, ..., and translates each segment to the target language. Returns a JSON array. If the model wraps the JSON in markdown fences or returns prose, the parser strips the fences; if JSON parsing still fails, the function falls back to translating each segment individually and labeling everything as SPEAKER_0.

The TTS Helper

def tts_generate(text, voice, language):
    resp = requests.post(f"{API}/text-to-speech/speech", headers=HEADERS, json={
        "text": text,
        "voice": voice,
        "language": language,
        "output_type": "binary_output",
        "provider": "telnyx"
    }, timeout=30)
    resp.raise_for_status()
    return resp.content

One TTS call per segment. output_type: "binary_output" returns the raw mp3 bytes. provider: "telnyx" selects the Telnyx KokoroTTS engine. The language parameter controls pronunciation for the target language — you pass es for Spanish, ja for Japanese, etc.

The Async Pipeline

@app.route("/dub", methods=["POST"])
def start_dubbing():
    # ... validate, build job record ...
    thread = threading.Thread(
        target=_run_pipeline,
        args=(job_id, audio_bytes, source_lang, target_lang),
        daemon=True
    )
    thread.start()
    return jsonify({"job_id": job_id, "status": "queued", ...}), 202

The handler returns immediately. The background thread runs STT → Inference → TTS, updates the job status at each step (transcribing → translating → synthesizing → complete), and stores the concatenated audio bytes on the job. On any exception, the job moves to failed with the error message.

The Audio Download Endpoint

@app.route("/dub/<job_id>/audio", methods=["GET"])
def get_audio(job_id):
    # ... fetch job, check status == "complete" ...
    return Response(
        io.BytesIO(audio),
        mimetype="audio/mpeg",
        headers={"Content-Disposition": f"attachment; filename={job_id}.mp3"}
    )

The dubbed audio is the concatenated mp3 chunks from each TTS call. Byte-level concatenation is fine for a demo — for production you'd mux with ffmpeg's concat demuxer so segment boundaries are sample-accurate.

Step 3: Run the App

python app.py

Server starts on http://127.0.0.1:5000. No ngrok, no webhook configuration — this is a pure HTTP API.

Step 4: Dub a Clip

Health check first:

curl http://localhost:5000/health
# {"status":"ok","total_jobs":0,"active":0,"supported_languages":15,"version":"1.0.0"}

List the supported target languages:

curl http://localhost:5000/languages
# {"languages":{"es":"Spanish","fr":"French","de":"German","pt":"Portuguese",
#  "it":"Italian","ja":"Japanese","ko":"Korean","zh":"Chinese",
#  "ar":"Arabic","hi":"Hindi","ru":"Russian","nl":"Dutch",
#  "sv":"Swedish","pl":"Polish","tr":"Turkish"}}

Start a dubbing job — upload a short mp3 and pick a target language:

curl -X POST http://localhost:5000/dub \
  -F audio=@episode.mp3 \
  -F target_language=es \
  -F source_language=en
# {"job_id":"dub-a1b2c3d4","status":"queued","source_language":"en",
#  "target_language":"es (Spanish)",
#  "message":"Pipeline started. Poll GET /dub/<job_id> for status."}

Poll for status:

curl http://localhost:5000/dub/dub-a1b2c3d4 | python3 -m json.tool
# status: queued → transcribing → translating → synthesizing → complete

Once status == "complete", download the dubbed mp3:

curl http://localhost:5000/dub/dub-a1b2c3d4/audio --output dubbed.mp3

And get the side-by-side transcript:

curl http://localhost:5000/dub/dub-a1b2c3d4/transcript | python3 -m json.tool
# {
#   "job_id": "dub-a1b2c3d4",
#   "source": "en",
#   "target": "es",
#   "segments": [
#     {
#       "speaker": "SPEAKER_0",
#       "original": "Hello and welcome to the show.",
#       "translated": "Hola y bienvenidos al programa.",
#       "timestamp": "0.0s - 2.5s"
#     }
#   ]
# }

Customizing the Pipeline

The VOICE_MAP and SUPPORTED_LANGUAGES dicts are the control surfaces. Small changes produce very different dubs.

Use different voices per language — pick voices that match the target language's accent:

LANGUAGE_VOICE_OVERRIDE = {
    "es": {"male": "Telnyx.KokoroTTS.em_alex", "female": "Telnyx.KokoroTTS.ef_dora"},
    "fr": {"male": "Telnyx.KokoroTTS.bm_daniel", "female": "Telnyx.KokoroTTS.ff_siwis"},
    "pt": {"male": "Telnyx.KokoroTTS.pm_alex", "female": "Telnyx.KokoroTTS.pf_dora"},
}

Add more languages — extend SUPPORTED_LANGUAGES and STT_LANGUAGE_MAP:

SUPPORTED_LANGUAGES["vi"] = "Vietnamese"
STT_LANGUAGE_MAP["vi"] = "vi-VN"

Use a stronger diarization prompt — for panel discussions with 5+ speakers, give the LLM more context:

prompt = (
    f"You are a dubbing assistant. Below are {len(segment_texts)} transcribed audio segments "
    f"from a panel discussion in {source_lang}. There are likely 3-5 speakers. "
    f"For EACH segment: assign a speaker label and translate to {target_lang_name}. ..."
)

Going to Production

This example uses in-memory storage and byte-level MP3 concatenation for simplicity. For production:

Database — replace the in-memory dict with PostgreSQL or Redis so jobs survive restarts
Audio muxing — replace byte concat with ffmpeg's concat demuxer for sample-accurate boundaries
Object storage — write finished audio to Telnyx Storage and return a signed URL instead of streaming from memory
Authentication — add API key validation on your endpoints
Error recovery — retry transient STT/TTS failures with exponential backoff
Prompt engineering — tune the speaker-labeling prompt for your domain (2-person interview vs. podcast with N speakers)
Rate limiting — protect /dub from abuse
WebSockets — for real-time dubbing, swap the polling model for the Telnyx streaming STT and TTS WebSocket endpoints

Frequently Asked Questions

Can I use a different LLM for the labeling + translation call?
Yes. Telnyx AI Inference is OpenAI-compatible. Set AI_MODEL in .env to any model available on the platform — moonshotai/Kimi-K2.6 is the default, but meta-llama/Llama-3.3-70B-Instruct or openai/gpt-4o work too. The JSON-output prompt is model-agnostic.

Why does the LLM do speaker labeling instead of the STT API?
Telnyx STT (like OpenAI's Whisper transcription API) does not return speaker labels. Rather than adding a separate diarization vendor, the app sends all segments to the LLM in one chat call and asks it to assign speaker labels based on conversational context — turn-taking, topic shifts, register. The LLM is already in the pipeline for translation, so the marginal cost is a slightly longer prompt.

How much does a dub cost?
A 1-minute clip with ~120 words of dialogue costs roughly: STT ~$0.006 (Whisper per minute), Inference ~$0.001 (one short chat call), TTS ~$0.015 (Kokoro per 1K characters). Total: ~2 cents per minute of dubbed audio. The 15-language voice pool is the same price.

What audio formats can I upload?
The Telnyx STT endpoint accepts mp3, wav, m4a, ogg, and webm. The example uploads as audio/mpeg (mp3) — change the files= tuple's mime type if you upload a different format.

Can I dub a 30-minute file?
Yes. The async pipeline means the HTTP request returns immediately. A 30-minute file takes ~30-60 seconds end-to-end (STT is the slow part). The in-memory dict will hold the full audio — for files longer than an hour, swap to object storage.

What happens if the LLM returns non-JSON?
The parser strips markdown fences if present. If JSON parsing still fails, the function falls back to translating each segment individually with a per-segment chat call and labeling everything as SPEAKER_0. The job completes with a single-speaker dub — degraded but not broken.

Can I stream the dubbed audio back instead of downloading?
Yes — Telnyx has streaming STT (GET /v2/speech-to-text/transcription, WebSocket) and streaming TTS (GET /v2/text-to-speech/speech, WebSocket) endpoints. For real-time dubbing, swap the batch endpoints for the streaming ones and pipe STT segments to the LLM as they arrive.

Resources

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

Harpreet Singh Seehra — Thu, 23 Jul 2026 11:01:26 +0000

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, or BLOCK.
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:app as 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

Turn a Text Chatbot Into a Voice Bot With Telnyx Conversation Relay

Harpreet Singh Seehra — Tue, 21 Jul 2026 13:08:40 +0000

Turn a Text Chatbot Into a Voice Bot With Telnyx Conversation Relay

Call a phone number and talk to your existing AI chatbot — the same one that already answers your Telegram, Slack, or web messages. Telnyx Conversation Relay handles all the telephony audio (speech-to-text, text-to-speech, call control). Your application only exchanges text over a WebSocket, so the chatbot needs zero changes: it doesn't even know the input came from a phone call instead of a chat message.

What You'll Build

A bridge app that connects a phone call to an existing text-in/text-out AI chatbot:

Caller dials a Telnyx number — the call hits your TeXML application
TeXML returns <Connect><ConversationRelay> — Telnyx opens a WebSocket to your app
Caller speaks — Telnyx transcribes the speech to text and sends it as a prompt frame
Bridge forwards to your chatbot — calls your existing OpenAI-compatible /chat/completions endpoint
Chatbot replies — the bridge streams the reply back as text frames
Telnyx speaks the reply — TTS converts the text to speech for the caller

The whole flow is visible on a live dashboard: the WebSocket frames, the conversation history, and the chatbot's streaming tokens — all updating in real time via Server-Sent Events.

Why This Matters

Every team that builds an AI chatbot eventually faces the same question: "Can we make this a voice bot too?" The traditional answer involves rebuilding the bot for voice — new prompts, new latency constraints, new telephony integration, new audio handling.

Conversation Relay skips all of that. If your bot already takes text in and produces text out, it's already a voice bot. Telnyx converts caller speech to text (your bot's input), and your bot's text to speech (back to the caller). Your bot's logic, personality, tools, and knowledge base stay exactly the same.

This is the carrier-grade solution: no custom STT/TTS pipeline, no audio streaming code, no SIP integration. Just a WebSocket that moves text.

Prerequisites

Python 3.10+
A Telnyx account with a phone number
An existing AI chatbot with an OpenAI-compatible /chat/completions endpoint (any framework, any language)
ngrok for exposing your local server

The Architecture

  Caller speaks into phone
        │
        ▼
  ┌─────────────────────────────┐
  │ Telnyx (Conversation Relay) │  STT + TTS + call control
  │  transcribes speech → text   │
  └──────────────┬──────────────┘
                 │  WebSocket text frame: {"type":"prompt","text":"...","last":true}
                 ▼
  ┌─────────────────────────────┐
  │  This bridge app (app.py)    │  ~140 lines of glue
  │  forwards text to your bot   │
  └──────────────┬──────────────┘
                 │  POST /v1/chat/completions (OpenAI-compatible)
                 ▼
  ┌─────────────────────────────┐
  │  Your existing AI chatbot    │  ← NO CHANGES
  │  returns a text reply        │
  └──────────────┬──────────────┘
                 │  text reply
                 ▼
  ┌─────────────────────────────┐
  │  Bridge sends text frame     │  {"type":"text","token":"...","last":true}
  └──────────────┬──────────────┘
                 │
                 ▼
  Telnyx TTS speaks the reply to the caller

The key insight: if your bot already takes text in and produces text out, Conversation Relay makes it a voice bot. Telnyx converts caller speech to text (your bot's input), and your bot's text to speech (back to the caller). Your bot's logic, personality, tools, and knowledge base stay exactly the same.

Step 1: Clone and Configure

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/conversation-relay-voice-bot-python
cp .env.example .env
pip install -r requirements.txt

Edit .env with your credentials:

TELNYX_API_KEY=your_telnyx_api_key_here
CHATBOT_BASE_URL=http://localhost:18789    # your existing chatbot's URL
CHATBOT_TOKEN=your_gateway_auth_token       # bearer token for your chatbot
CHATBOT_MODEL=openclaw                      # model name your endpoint expects
TELNYX_PUBLIC_BASE_URL=https://abc.ngrok.app  # your ngrok URL (set in Step 2)

The CHATBOT_BASE_URL, CHATBOT_TOKEN, and CHATBOT_MODEL point to your existing AI chatbot — any endpoint that accepts OpenAI-compatible /chat/completions requests works.

Step 2: Start the Bridge and Expose It

Conversation Relay needs a public WebSocket URL. Start the app, then start ngrok:

python app.py           # starts on http://localhost:8000

In another terminal:

ngrok http 8000

Copy the HTTPS URL (e.g. https://abc.ngrok.app) and set it as TELNYX_PUBLIC_BASE_URL in .env, then restart the app.

Step 3: Configure a TeXML Application

Go to the Telnyx Portal → TeXML Applications → create a new app
Set the Voice URL (inbound) to https://abc.ngrok.app/texml/inbound
Assign a phone number to this TeXML application

When a caller dials that number, Telnyx fetches the TeXML, which returns <Connect><ConversationRelay> — telling Telnyx to open a WebSocket to your app and start the text bridge.

Step 4: Understand the Code

Everything lives in app.py. Here's what each piece does.

The TeXML Response

When Telnyx fetches the voice URL, the app returns TeXML that tells Telnyx to use Conversation Relay:

def texml_response() -> str:
    ws_url = escape(conversation_relay_ws_url(), quote=True)
    return f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
    <Connect action="{action_url}">
        <ConversationRelay
            url="{ws_url}"
            welcomeGreeting="{greeting}"
            voice="{voice}"
            language="{language}"
            transcriptionProvider="{provider}"
            dtmfDetection="true"
        />
    </Connect>
</Response>"""

The url attribute is the WebSocket endpoint Telnyx connects to. The welcomeGreeting is spoken immediately when the call connects.

The WebSocket Bridge

The core of the app is the WebSocket handler. It receives frames from Telnyx and forwards text to your chatbot:

@sock.route("/ws/conversation-relay")
def conversation_relay_socket(ws) -> None:
    while True:
        raw = ws.receive()
        frame = json.loads(raw)
        ftype = str(frame.get("type") or "unknown")

        if ftype == "setup":
            session_id = str(frame.get("sessionId"))
            sessions[session_id] = {"history": [], "started": time.time()}

        elif ftype == "prompt" and frame.get("last") is True:
            caller_text = frame.get("voicePrompt") or frame.get("text")
            session["history"].append({"role": "user", "content": caller_text})
            reply = ask_chatbot_streamed(session["history"], ws)
            session["history"].append({"role": "assistant", "content": reply})

The Streaming Chatbot Call

The bridge streams the chatbot's reply token-by-token so TTS starts speaking the first words while the LLM is still generating the rest:

def ask_chatbot_streamed(history, ws) -> str:
    payload = {"model": CHATBOT_MODEL, "messages": messages, "stream": True}
    resp = requests.post(url, json=payload, stream=True)
    for line in resp.iter_lines(decode_unicode=True):
        chunk = json.loads(line.removeprefix("data: "))
        token = chunk.get("choices", [{}])[0].get("delta", {}).get("content")
        if token:
            ws.send(text_frame(token, last=False))  # partial frame
    ws.send(text_frame("", last=True))  # final frame

The VOICE_SYSTEM_PROMPT prepends a phone-specific instruction: "You are on a phone call. Keep responses short and conversational — 2-3 sentences max. No markdown." This adapts the chatbot's output for voice without changing the chatbot itself.

Step 5: Call and Talk

Dial your Telnyx number. You'll hear the welcome greeting, then speak — Telnyx transcribes your speech, the bridge forwards it to your chatbot, and your chatbot's reply is spoken back to you.

Conversation Relay Frame Types

Telnyx → your app (received):

Frame	Meaning
`setup`	First frame — session ID and call info
`prompt`	Transcribed caller speech. `last:false` = partial, `last:true` = final
`dtmf`	Caller pressed a digit
`interrupt`	Caller barged in over TTS playback
`error`	Error frame

Your app → Telnyx (sent):

Frame	Meaning
`text`	Text to speak via TTS. `last:false` = stream tokens, `last:true` = finalize

Frequently Asked Questions

What is Telnyx Conversation Relay? Conversation Relay is a Telnyx product that bridges phone calls to your application over a WebSocket. Telnyx handles all the telephony audio (speech-to-text, text-to-speech, call control), and your app only exchanges text — no audio handling required.

Does my chatbot need to know it's on a phone call? No. Your chatbot receives text and returns text, exactly as it does for chat messages. The bridge prepends a voice-specific system prompt but your chatbot's logic stays unchanged.

Can I use this with any chatbot framework? Yes, as long as your chatbot exposes an OpenAI-compatible /chat/completions endpoint. This includes LangChain, LlamaIndex, custom OpenAI wrappers, Clawdbot, and any endpoint that accepts the standard chat completions payload.

What transcription providers are supported? Conversation Relay supports Deepgram (default), Google, and Telnyx's own transcription provider. Set TRANSCRIPTION_PROVIDER in .env to switch.

How does streaming work? The bridge sends partial text frames (last: false) as tokens arrive from the LLM, so TTS starts speaking the first words while the LLM is still generating. The final frame (last: true) tells Telnyx the reply is complete.

What happens when the caller interrupts? Telnyx sends an interrupt frame when the caller talks over TTS playback. You can use this signal to stop the in-flight LLM request.

Can I handle DTMF input? Yes. Telnyx sends a dtmf frame when the caller presses a digit. You can extend the handler to use DTMF for menu navigation or input collection.

Resources

Eliminate Dead Air in AI Voice Assistants with Filler Messages — Telnyx AI Assistant Webhook Demo

Harpreet Singh Seehra — Tue, 21 Jul 2026 12:41:00 +0000

When an AI voice assistant calls a sync webhook tool — to look up an order, query a database, check inventory — the caller hears silence. Five seconds of dead air on a phone call feels like thirty. Hang up. Call again. Complaint filed.

Filler messages solve this. Configured per-tool in Telnyx Mission Control, they let the AI Assistant speak scripted phrases while waiting for the webhook to respond: "Let me look that up for you." at 0 seconds, "Still working on this, one moment please." at 5 seconds, "Almost there, thanks for your patience." at 15 seconds. The caller stays engaged. The webhook takes the time it needs. No dead air.

This walkthrough builds a complete demo: a Flask webhook server with a live split-screen dashboard that shows the webhook call, the filler messages firing, and the countdown in real time. 124 lines of Python, one AI Assistant, one phone number, one dashboard — and zero silence.

What You'll Build

A webhook server that an AI Assistant calls when a caller asks about an order:

Caller dials the AI Assistant — a voice AI agent configured in Mission Control
Caller asks a question — "What's the status of my order 12345?"
AI Assistant calls the webhook — a sync tool call to /webhook/order-status
Webhook delays — intentionally sleeps for 12 seconds (configurable) to simulate a slow backend
Filler messages fire — the AI Assistant speaks scripted phrases at 0s, 5s, and 15s while the webhook is processing
Webhook responds — returns mock order status as JSON
AI Assistant reads the result — "Your order 12345 has shipped via FedEx, tracking number FX-98765, estimated delivery July 20th."
Dashboard shows everything — a split-screen browser UI with the call timeline on the left and server logs on the right, updated in real time via Server-Sent Events

Why Filler Messages Matter

Every voice AI developer hits the same problem: sync webhook tool calls take time. A database query takes 2 seconds. A third-party API takes 5 seconds. A complex lookup takes 10 seconds. On a phone call, that's dead air — and dead air is the enemy of conversation.

Without filler messages, developers build workarounds:

Background music — doesn't tell the caller what's happening
"Please hold" TTS — fires once, then silence resumes
Async tools — more complex to build, don't work for all use cases
Faster webhooks — not always possible when you depend on external systems

Filler messages are the carrier-grade solution: the AI Assistant speaks configurable phrases at configurable intervals while the webhook processes. The caller knows the system is working. The developer doesn't have to build a workaround.

Prerequisites

Python 3.8+
A Telnyx account with funded balance
A Telnyx API key
A Telnyx phone number assigned to an AI Assistant
An AI Assistant with telephony enabled
ngrok for exposing your local server

The Architecture

  Caller dials AI Assistant
        │
        ▼
  ┌──────────────────────┐
  │ Telnyx AI Assistant   │
  │ (Mission Control)     │
  └────────┬─────────────┘
           │ sync tool call
           ▼
  ┌──────────────────────┐    SSE     ┌──────────────────┐
  │ Flask Webhook Server  │──────────►│ Browser Dashboard  │
  │ POST /webhook/        │           │ Split-screen UI    │
  │   order-status        │           │ (timeline + logs)  │
  └──────────────────────┘           └──────────────────┘
           │
    delays N seconds,
    then returns mock
    order status

Step 1: Clone and Configure

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-assistant-filler-messages-demo-python
cp .env.example .env
pip install -r requirements.txt

Edit .env with your credentials:

TELNYX_API_KEY=your_telnyx_api_key_here
WEBHOOK_DELAY_SECONDS=12
PORT=5000

Step 2: Set Up the AI Assistant

The repo includes a setup.py script that creates the AI Assistant, adds the webhook tool with filler messages, and assigns your phone number — all via the Telnyx API.

ngrok http 5000
python setup.py https://abc123.ngrok.io

The script will:

List your active phone numbers — pick one
Create an AI Assistant named "Filler Messages Demo" with Claude Haiku 4.5
Add a sync webhook tool (check_order_status) pointing to your ngrok URL
Configure the three filler messages on the tool
Assign your phone number to the assistant's backing TeXML app

Important: The API sets the filler message values, but playback on calls requires configuring them through the Mission Control UI. After running setup.py, open the tool's Filler Messages tab in Mission Control and verify the three messages are present.

Step 3: Understand the Code

Everything lives in app.py (124 lines). Here's what each piece does.

The Filler Message Configuration

FILLER_CONFIG = [
    {"type": "request_start", "content": "Let me look that up for you.", "timing_ms": 0},
    {"type": "request_response_delayed", "content": "Still working on this, one moment please.", "timing_ms": 5000},
    {"type": "request_response_delayed", "content": "Almost there, thanks for your patience.", "timing_ms": 15000},
]

Two types:

request_start — fires immediately when the webhook tool is called
request_response_delayed — fires after timing_ms milliseconds if the webhook hasn't responded yet

The Webhook Endpoint

@app.route("/webhook/order-status", methods=["POST"])
def webhook_order_status():
    body = request.get_json(silent=True) or {}
    order_id = body.get("order_id", "unknown")

    broadcast("tool_call_received", {"order_id": order_id, "request_body": body, "delay_seconds": WEBHOOK_DELAY_SECONDS})

    for filler in FILLER_CONFIG:
        if filler.get("timing_ms", 0) / 1000 < WEBHOOK_DELAY_SECONDS:
            broadcast("filler_message", {"content": filler["content"], "filler_type": filler["type"], "timing_ms": filler.get("timing_ms", 0)})

    for elapsed in range(1, WEBHOOK_DELAY_SECONDS + 1):
        time.sleep(1)
        broadcast("countdown", {"elapsed": elapsed, "total": WEBHOOK_DELAY_SECONDS, "remaining": WEBHOOK_DELAY_SECONDS - elapsed})

    order = MOCK_ORDERS.get(order_id, {"order_id": order_id, "status": "not_found", "error": "Order not found"})
    return jsonify({"result": order}), 200

The webhook broadcasts four event types to the dashboard: tool_call_received, filler_message, countdown, and response_sent.

The SSE Dashboard

Real-time updates via Server-Sent Events. Each connected browser gets a queue, and the broadcast() function pushes events to all clients:

def broadcast(event_type, data):
    payload = json.dumps({"type": event_type, "timestamp": time.time(), **data})
    msg = f"event: {event_type}\ndata: {payload}\n\n"
    dead = []
    with clients_lock:
        for q in clients:
            try:
                q.put_nowait(msg)
            except queue.Full:
                dead.append(q)
        for q in dead:
            clients.remove(q)

Step 4: Run the Demo

python app.py

Open http://localhost:5000 to see the split-screen dashboard. Call your AI Assistant's phone number. Ask:

"What's the status of my order 12345?"

Watch the dashboard and listen to the call:

0s — the AI Assistant says "Let me look that up for you." and calls the webhook
5s — the AI Assistant says "Still working on this, one moment please."
12s — the webhook responds with the order status
After response — the AI Assistant reads back the order details

Customizing the Filler Messages

Different timings and content produce very different caller experiences:

Add a 10-second filler:

FILLER_CONFIG = [
    {"type": "request_start", "content": "Let me look that up for you.", "timing_ms": 0},
    {"type": "request_response_delayed", "content": "Still working on this, one moment please.", "timing_ms": 5000},
    {"type": "request_response_delayed", "content": "I'm checking with the shipping carrier now.", "timing_ms": 10000},
    {"type": "request_response_delayed", "content": "Almost there, thanks for your patience.", "timing_ms": 15000},
]

Use context-aware fillers for different tools:

For a payment processing tool:

{"type": "request_start", "content": "Let me process that payment for you.", "timing_ms": 0}
{"type": "request_response_delayed", "content": "I'm confirming the transaction with your bank.", "timing_ms": 5000}

Going to Production

Real data source — replace MOCK_ORDERS with actual database or API queries
Remove artificial delay — the delay exists only for demo purposes; real webhooks should respond as fast as possible
Webhook verification — validate Telnyx webhook signatures
Timeout alignment — set the tool timeout_ms (default 5000ms) to at least 30000ms

Frequently Asked Questions

What are AI Assistant filler messages?
Filler messages are scripted phrases the AI Assistant speaks while waiting for a sync webhook tool to respond. They eliminate dead air on the call and keep the caller informed. They're configured per-tool in Mission Control.

How are filler messages different from TTS hold music?
Filler messages are spoken by the AI Assistant in its own voice, as part of the conversation. Hold music is audio played while the call is on hold. Filler messages feel like the assistant is still there and working on your request.

What happens if the webhook responds before the first filler message fires?
If the webhook responds before the first request_response_delayed message's timing_ms, that message doesn't fire. The request_start message always fires immediately when the tool is called.

How do I configure filler messages?
Filler messages are configured per-tool in Mission Control. Open the AI Assistant, edit the webhook tool, and go to the Filler Messages tab. You can also set them via the API (setup.py does this), but playback on calls requires the Mission Control UI configuration.

Resources

Practice Negotiations with an AI Phone Agent — Real-Time Roleplay with Telnyx Voice AI

Harpreet Singh Seehra — Wed, 15 Jul 2026 15:11:00 +0000

Imagine picking up your phone, dialing a number, and practicing a salary negotiation against an AI that plays the hiring manager — one that pushes back on your first offer, counters with budget constraints, and then scores your technique after you hang up. No scheduling a mock interview, no paying a coach, no awkward roleplay with a colleague.

This is the AI Negotiation Practice Phone — a 110-line Python app built with Telnyx Call Control and AI Inference. Three scenarios (salary, sales deal, vendor contract), voice-driven conversation, and a structured performance score delivered after every call. The AI stays in character, adapts to your approach, and gives you actionable feedback.

In this walkthrough, you'll build it from scratch. Clone the repo, configure a phone number, and start practicing in minutes.

What You'll Build

A phone number anyone can call to practice negotiations:

Caller dials in — Telnyx answers and offers a menu
Scenario selection — press 1 for salary negotiation, 2 for sales deal, 3 for vendor contract
AI opens — the AI plays the opposing role (hiring manager, enterprise buyer, or vendor account manager) and makes an opening position
Live negotiation — caller speaks naturally, AI responds in character, pushes back, counters, and adapts
Post-call scoring — on hangup, the AI scores the negotiation across 5 dimensions and returns structured JSON
Session history — every practice session is stored and accessible via GET /sessions

The whole interaction is voice-driven: Text-to-Speech reads the AI's lines, and the caller responds with natural speech. The model (Llama 3.3 70B via Telnyx AI Inference) handles both the real-time roleplay and the post-call evaluation.

Why This Is Interesting

Most negotiation training tools are text-based chatbots or static video courses. Neither captures the pressure of a live conversation — the pauses, the pushback, the moment you have to think on your feet. This one puts you on a real phone call with an AI that has a budget, a role, and a hidden constraint (like "max 15% discount" or "budget is $155K with flexibility to $165K").

The scoring system makes it more than a conversation. After you hang up, the AI evaluates your performance across five dimensions — anchoring, concession strategy, active listening, creativity, and confidence — plus an overall score and specific strengths/improvements. You can practice the same scenario five times and track whether your technique improves.

It also demonstrates the DTMF + speech pattern: the app uses DTMF (keypad input) for scenario selection and speech recognition for the negotiation itself. This two-input pattern shows up in many real IVR workflows.

Prerequisites

Python 3.8+
A Telnyx account with funded balance
A Telnyx API key
A Telnyx phone number with voice enabled
A Call Control Application configured with your webhook URL
ngrok for exposing your local server to Telnyx webhooks

The Architecture

Phone Call
    │
    ▼
Telnyx Call Control (webhook events)
    │
    ▼
Flask app (app.py, 110 lines)
    │
    ├──► call.initiated → answer the call
    ├──► call.answered → TTS menu (press 1, 2, or 3)
    ├──► call.speak.ended → gather DTMF (scenario selection)
    ├──► call.gather.ended → DTMF → pick scenario → AI opening line
    │                    └──► speech → AI inference → TTS response (negotiation loop)
    └──► call.hangup → score negotiation as JSON → store
    │
    ▼
Telnyx AI Inference (Llama 3.3 70B)
    │
    ▼
Structured score (JSON with 5 dimensions + feedback)

The app is a state machine with two phases: selection (DTMF menu) and negotiating (speech conversation). Each Telnyx webhook event drives the next action. After hangup, the full conversation is sent to the AI with a scoring prompt that returns structured JSON.

Step 1: Clone and Configure

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-negotiation-practice-phone-python
cp .env.example .env
pip install -r requirements.txt

Edit .env with your credentials:

TELNYX_API_KEY=your_telnyx_api_key_here
TELNYX_PUBLIC_KEY=...
AI_MODEL=meta-llama/Llama-3.3-70B-Instruct
PRACTICE_NUMBER=+13105551234

Step 2: Understand the Code

Everything lives in app.py (110 lines). Here's what each piece does.

The Scenarios

The app ships with three built-in negotiation scenarios, each with a role and a hidden context:

SCENARIOS = {
    "1": {"role": "hiring manager", "context": "The candidate wants $180K. Your budget is $155K with flexibility to $165K. Push back on experience level. You can offer equity or signing bonus as alternatives."},
    "2": {"role": "enterprise buyer", "context": "You're evaluating their SaaS product at $50K/year. You have a competing offer at $35K. Your budget is $45K. Ask for volume discounts and longer payment terms."},
    "3": {"role": "vendor account manager", "context": "The client wants to reduce their contract by 40%. They're a top-10 account. You can offer 15% discount max, or restructure the deal with different terms."}
}

The context is injected into the system prompt so the AI knows its constraints — but the caller doesn't know them. You discover the other side's budget and flexibility through the conversation, just like a real negotiation.

The State Machine: Selection → Negotiating

The webhook handler has two states. In select state, it gathers DTMF digits to pick a scenario:

if call["state"] == "select":
    client.calls.actions.gather(ccid, input_type="dtmf", timeout_secs=10, min_digits=1, max_digits=1)

When the caller presses 1, 2, or 3, the app loads the scenario, builds the system prompt, asks the AI for an opening line, and switches to negotiating state:

if call["state"] == "select":
    scenario = SCENARIOS.get(digits, SCENARIOS["1"])
    call["state"] = "negotiating"
    call["scenario"] = scenario
    call["conversation"] = [{"role": "system", "content": f"You are a {scenario['role']} in a negotiation. {scenario['context']} Stay in character. Be firm but fair. Push back on their first offer. Keep responses under 2 sentences. After 6 exchanges, start wrapping up."}]
    opening = call_inference(call["conversation"] + [{"role": "user", "content": "The negotiation begins. Make your opening position."}])
    call["conversation"].append({"role": "assistant", "content": opening})
    client.calls.actions.speak(ccid, payload=opening, voice="female", language_code="en-US")

In negotiating state, it gathers speech and runs the conversation loop:

else:
    client.calls.actions.gather(ccid, input_type="speech", end_silence_timeout_secs=2, timeout_secs=20, language_code="en-US")

The conversation loop: speak → gather → infer → speak. Each turn, the AI sees the full conversation history, so it maintains context and remembers what you offered.

The Inference Helper

def call_inference(messages, max_tokens=200):
    resp = requests.post(INFERENCE_URL, headers={"Authorization": f"Bearer {TELNYX_API_KEY}", "Content-Type": "application/json"},
        json={"model": AI_MODEL, "messages": messages, "max_tokens": max_tokens, "temperature": 0.7}, timeout=15)
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

OpenAI-compatible chat completions call against Telnyx AI Inference. Temperature 0.7 — higher than the price quote agent (0.5) because you want the negotiator to be a bit more creative and unpredictable, not perfectly consistent every time.

Post-Call Scoring

When the caller hangs up, the app sends the full conversation to the AI with a scoring prompt:

elif event_type == "call.hangup":
    call = active_calls.pop(ccid, None)
    if call and len(call.get("conversation", [])) > 3:
        score_prompt = [{"role": "system", "content": "Score this negotiation practice. Return JSON: anchoring (1-10), concession_strategy (1-10), active_listening (1-10), creativity (1-10), confidence (1-10), overall (1-10), strengths (list), improvements (list), deal_outcome (string)."},
            {"role": "user", "content": chr(10).join(f"{m['role']}: {m['content']}" for m in call["conversation"] if m["role"] != "system")}]
        try:
            score = json.loads(call_inference(score_prompt, max_tokens=400))
            sessions.append({"scenario": call.get("scenario", {}).get("role"), "score": score, "duration": int(time.time() - call["start"])})
        except Exception:
            pass

The AI returns a structured score with five dimensions, a list of strengths, a list of improvements, and the deal outcome. The app stores it alongside the scenario and call duration.

Accessing Sessions

@app.route("/sessions", methods=["GET"])
def list_sessions():
    return jsonify({"sessions": sessions[-20:]}), 200

Hit GET /sessions to see the last 20 practice sessions as JSON — track your progress over time, compare scenarios, or pipe into an analytics dashboard.

Memory Cleanup

The app includes a background thread that cleans up stale call state:

def _start_ttl_cleanup(*stores, ttl_seconds=3600, interval=300):
    def _cleanup():
        while True:
            _ttl_time.sleep(interval)
            cutoff = _ttl_time.time() - ttl_seconds
            for store in stores:
                expired = [k for k, v in store.items()
                           if isinstance(v, dict) and v.get("_ts", _ttl_time.time()) < cutoff]
                for k in expired:
                    store.pop(k, None)
    threading.Thread(target=_cleanup, daemon=True).start()

Calls that crash or get abandoned are automatically cleaned up after an hour — no memory leaks in long-running deployments.

Step 3: Run the App

Start the Flask server:

python app.py

In a separate terminal, expose your local server with ngrok:

ngrok http 5000

Copy the HTTPS URL and configure it in the Telnyx Portal:

Go to Call Control Applications
Create or edit your application
Set the Webhook URL to https://<your-ngrok-url>.ngrok.app/webhooks/voice

Assign your Telnyx phone number to this Call Control Application if you haven't already.

Step 4: Call and Practice

Call your Telnyx number from any phone. You'll hear:

"Negotiation Practice! Press 1 for salary negotiation, 2 for sales deal, 3 for vendor contract."

Press 1. The AI (playing a hiring manager) opens with something like:

"Thanks for calling in. I've reviewed your experience, and I have to be honest — we typically bring in candidates at $155K for this level. I know you mentioned $180K, but given the experience gap, I'm not sure I can justify that range. What would make you worth the premium?"

Negotiate. Push back. Make your case. The AI will counter, bring up constraints, and eventually either reach a deal or hold firm.

After you hang up, check your score:

curl http://localhost:5000/sessions | python3 -m json.tool

You'll see a structured score object like:

{
  "scenario": "hiring manager",
  "score": {
    "anchoring": 7,
    "concession_strategy": 5,
    "active_listening": 8,
    "creativity": 6,
    "confidence": 7,
    "overall": 6.6,
    "strengths": ["Strong opening anchor at $180K", "Asked about equity as alternative to base salary"],
    "improvements": ["Conceded too quickly on base salary", "Didn't explore signing bonus option"],
    "deal_outcome": "Settled at $162K base + $10K signing bonus"
  },
  "duration": 184
}

Customizing the Scenarios

The SCENARIOS dict is the control surface. Small changes produce very different practice sessions:

Add a real estate negotiation:

"4": {"role": "seller's agent", "context": "The house is listed at $650K. The seller will accept $620K minimum. The buyer seems interested but price-sensitive. Push for close to asking but signal flexibility on closing dates."}

Make the AI more aggressive:

call["conversation"] = [{"role": "system", "content": f"You are a {scenario['role']} in a negotiation. {scenario['context']} Stay in character. Be aggressive and push hard on the first offer. Never accept the first counter. Keep responses under 2 sentences. After 6 exchanges, start wrapping up."}]

Add multi-round scoring with trend tracking:

# After each session, compare to previous sessions in the same scenario
previous = [s for s in sessions if s.get("scenario") == call.get("scenario", {}).get("role")]
if previous:
    last_score = previous[-1]["score"]["overall"]
    new_score = score["overall"]
    score["trend"] = "improving" if new_score > last_score else "declining" if new_score < last_score else "stable"

Send the score via SMS after the call:

from telnyx import Message
score_text = f"Negotiation score: {score['overall']}/10. Strengths: {', '.join(score['strengths'][:2])}. Improvements: {', '.join(score['improvements'][:2])}."
Message.create(to=call["caller"], from_=PRACTICE_NUMBER, text=score_text)

Going to Production

This example uses in-memory storage for simplicity. For a production deployment:

Database — replace the in-memory sessions list with PostgreSQL or Redis so scores survive restarts
Authentication — add API key validation on the /sessions endpoint
User accounts — track scores per-user with phone number as the key
Scenario library — store scenarios in a database so coaches can add new ones without redeploying
Score history dashboard — build a simple web UI that shows score trends over time
Webhook verification — the app already validates Telnyx Ed25519 signatures; ensure your public key is current
Concurrency — run the app behind gunicorn with multiple workers
Error recovery — handle inference timeouts and call failures gracefully
Rate limiting — protect your webhook endpoint from abuse
Prompt tuning — test different system prompts and temperature settings for different difficulty levels

Build an AI Price Quote Phone Agent — Real-Time Custom Quotes with Telnyx Voice AI

Harpreet Singh Seehra — Wed, 15 Jul 2026 11:47:02 +0000

Imagine a customer calls your business and asks for a quote. Instead of putting them on hold, transferring to sales, or promising a callback, an AI agent picks up — asks 3–4 qualifying questions, builds a line-item quote with quantities and unit prices, and delivers the total on the call. No forms, no waiting, no friction.

This is the AI Price Quote Phone Agent — a 102-line Python app built with Telnyx Call Control and AI Inference. No call center, no CRM integration, no pre-built pricing engine. The AI has your product catalog and pricing, asks the right questions, and generates a structured quote in real time.

In this walkthrough, you'll build it from scratch. Clone the repo, configure a phone number, and deploy in minutes.

What You'll Build

A phone number that anyone can call to get a custom price quote:

Caller dials in — Telnyx answers and greets them
AI asks what they need — "What kind of communication services are you looking for?"
Conversation — caller describes their needs, AI asks 3–4 follow-up questions to estimate quantities
Quote delivered — AI summarizes with line items and monthly total
Structured extraction — on hangup, AI extracts the full quote as JSON (line items, quantities, unit prices, subtotals, monthly total, notes)
Quote stored — accessible via GET /quotes endpoint

The whole interaction is voice-driven: Text-to-Speech reads each question and the final quote, and the caller responds with natural speech. The AI model (Llama 3.3 70B via Telnyx AI Inference) handles the conversation and quote generation, and Call Control handles the telephony.

Why This Is Interesting

Most price quote flows are static: fill out a web form, wait for a sales email. This one is dynamic — a real-time conversation where the AI adapts its questions based on what the caller says. Need voice minutes and SMS? The AI asks about expected volume. Just want a toll-free number? The AI skips irrelevant questions and gets straight to the quote.

It also demonstrates a pattern that goes beyond chat: the AI doesn't just converse — it extracts structured data from the conversation. After the call ends, the app sends the full transcript back to the model with a "extract the quote as JSON" prompt. You get a clean, machine-readable quote object with line items and a monthly total — ready to store, email, or pipe into your billing system.

Prerequisites

Python 3.8+
A Telnyx account with funded balance
A Telnyx API key
A Telnyx phone number with voice enabled
A Call Control Application configured with your webhook URL
ngrok for exposing your local server to Telnyx webhooks

The Architecture

  Phone Call
      │
      ▼
  Telnyx Call Control (webhook events)
      │
      ▼
  Flask app (app.py, 102 lines)
      │
      ├──► call.initiated  → answer the call
      ├──► call.answered   → TTS greeting
      ├──► call.gather.ended → speech → AI Inference → TTS response
      ├──► call.speak.ended → gather next input
      └──► call.hangup     → extract quote as JSON → store
      │
      ▼
  Telnyx AI Inference (Llama 3.3 70B)
      │
      ▼
  Structured quote (JSON with line items)

The app is a state machine driven by Telnyx webhook events. Each event triggers the next action — answer, speak, gather, or hang up. The AI Inference call handles two jobs: conversing with the caller to build the quote, and extracting a structured JSON quote after the call ends.

Step 1: Clone and Configure

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-price-quote-phone-agent-python
cp .env.example .env
pip install -r requirements.txt

Edit .env with your credentials:

TELNYX_API_KEY=KEY019C8F8C3D268774F8F560946B984D9E_...
TELNYX_PUBLIC_KEY=...
AI_MODEL=meta-llama/Llama-3.3-70B-Instruct
QUOTE_NUMBER=+13105551234

Step 2: Understand the Code

Everything lives in app.py (102 lines). Here's what each piece does.

The Pricing Catalog

The AI has a built-in product catalog with per-unit pricing:

PRICING = {"voice_minutes": 0.005, "sms_messages": 0.004, "local_numbers": 1.00, "toll_free_numbers": 2.00,
    "sip_trunking_channel": 2.50, "ai_inference_1k_tokens": 0.01, "cloud_storage_gb": 0.05, "fax_pages": 0.07}

This gets injected into the system prompt so the AI knows exactly what to charge per unit:

SYSTEM_PROMPT = f"You are a pricing specialist. Available products and per-unit pricing: {json.dumps(PRICING)}. Ask what the caller needs, estimate quantities, build a quote. Keep responses under 2 sentences. After gathering requirements (usually 3-4 questions), summarize the quote with line items and monthly total."

The prompt is the control surface. You can swap the product catalog, change the questioning style, or adjust how many questions the AI asks before summarizing.

Handling Webhooks

The webhook handler is the core state machine. Each Telnyx event triggers the next action:

if event_type == "call.initiated" and p.get("direction") == "incoming":
    active_calls[ccid] = {"caller": p.get("from"), "conversation": [{"role": "system", "content": SYSTEM_PROMPT}], "start": time.time()}
    client.calls.actions.answer(ccid)
elif event_type == "call.answered":
    client.calls.actions.speak(ccid, payload="Hi! I can put together a custom quote for you right now. What kind of communication services are you looking for?", voice="female", language_code="en-US")
elif event_type == "call.speak.ended" and call:
    client.calls.actions.gather(ccid, input_type="speech", end_silence_timeout_secs=2, timeout_secs=20, language_code="en-US")
elif event_type == "call.gather.ended" and call:
    speech = p.get("speech", {}).get("result", "")
    call["conversation"].append({"role": "user", "content": speech})
    response = call_inference(call["conversation"])
    call["conversation"].append({"role": "assistant", "content": response})
    client.calls.actions.speak(ccid, payload=response, voice="female", language_code="en-US")

The conversation loop: speak → gather → infer → speak. Each turn, the AI sees the full conversation history, so it maintains context across all exchanges.

The Inference Helper

def call_inference(messages, max_tokens=250):
    resp = requests.post(INFERENCE_URL, headers={"Authorization": f"Bearer {TELNYX_API_KEY}", "Content-Type": "application/json"},
        json={"model": AI_MODEL, "messages": messages, "max_tokens": max_tokens, "temperature": 0.5}, timeout=15)
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

Straightforward OpenAI-compatible chat completions call against Telnyx AI Inference. Temperature 0.5 for consistency — you want quotes to be predictable, not creative.

Post-Call Quote Extraction

This is the interesting part. When the caller hangs up, the app sends the full conversation back to the AI with a different prompt — extract the structured quote:

elif event_type == "call.hangup":
    call = active_calls.pop(ccid, None)
    if call and len(call.get("conversation", [])) > 3:
        extract = [{"role": "system", "content": "Extract the price quote. Return JSON: line_items (list of {product, quantity, unit_price, subtotal}), monthly_total (number), notes (string)."},
            {"role": "user", "content": chr(10).join(f"{m['role']}: {m['content']}" for m in call["conversation"] if m["role"] != "system")}]
        quote = json.loads(call_inference(extract, max_tokens=400))
        quote["caller"] = call["caller"]
        quote["timestamp"] = time.strftime("%Y-%m-%dT%H:%M:%SZ")
        quotes.append(quote)

The AI returns clean JSON with line items, quantities, unit prices, subtotals, and a monthly total. The app attaches the caller's number and timestamp, then stores it.

Accessing Quotes

@app.route("/quotes", methods=["GET"])
def list_quotes():
    return jsonify({"quotes": quotes[-20:]}), 200

Hit GET /quotes to see the last 20 quotes as JSON — ready to pipe into a CRM, billing system, or analytics dashboard.

Step 3: Run the App

Start the Flask server:

python app.py

In a separate terminal, expose your local server with ngrok:

ngrok http 5000

Copy the HTTPS URL and configure it in the Telnyx Portal:

Go to Call Control Applications
Create or edit your application
Set the Webhook URL to https://<your-ngrok-url>.ngrok.app/webhooks/voice

Assign your Telnyx phone number to this Call Control Application if you haven't already.

Step 4: Call and Get a Quote

Call your Telnyx number from any phone. You'll hear:

"Hi! I can put together a custom quote for you right now. What kind of communication services are you looking for?"

Say something like: "I need about 50 phone numbers, 10,000 voice minutes a month, and a few thousand SMS messages."

The AI asks follow-up questions — how many SMS, any toll-free numbers, do you need SIP trunking — then summarizes the quote with line items and a monthly total.

After you hang up, check the extracted quote:

curl http://localhost:5000/quotes | python3 -m json.tool

You'll see a structured JSON object like:

{
  "line_items": [
    {"product": "local_numbers", "quantity": 50, "unit_price": 1.00, "subtotal": 50.00},
    {"product": "voice_minutes", "quantity": 10000, "unit_price": 0.005, "subtotal": 50.00},
    {"product": "sms_messages", "quantity": 5000, "unit_price": 0.004, "subtotal": 20.00}
  ],
  "monthly_total": 120.00,
  "notes": "Customer indicated potential for higher SMS volume in Q3.",
  "caller": "+12125551234",
  "timestamp": "2026-07-15T14:30:00Z"
}

Customizing the Agent

The pricing catalog and system prompt are the control surface. Small changes produce very different agents:

Swap the product catalog:

PRICING = {"consulting_hour": 200.00, "audit_report": 2500.00, "retainer_monthly": 8000.00, "training_session": 1500.00}

Change the questioning style:

SYSTEM_PROMPT = f"You are a pricing specialist. Available products: {json.dumps(PRICING)}. Ask one question at a time. Be conversational but efficient. After exactly 3 questions, summarize the quote with line items and monthly total. If the caller asks for a discount, explain standard pricing tiers."

Add discount logic:

SYSTEM_PROMPT = f"You are a pricing specialist. Products: {json.dumps(PRICING)}. For volumes over 10,000 units, apply a 15% discount. For over 50,000 units, apply 25%. Mention the discount in your summary."

Send the quote via SMS after the call:

# After extracting the quote, send it as SMS
from telnyx import Message
quote_text = f"Your quote: {quote['monthly_total']}/month. {len(quote['line_items'])} line items."
Message.create(to=call["caller"], from_=QUOTE_NUMBER, text=quote_text)

The pattern — conversational AI with a product catalog + post-call structured extraction — works for any quoting or ordering flow.

Going to Production

This example uses in-memory storage for simplicity. For a production deployment:

Database — replace the in-memory quotes list with PostgreSQL or Redis so quotes survive restarts
Concurrency — run the app behind gunicorn with multiple workers
Error recovery — handle inference timeouts and call failures gracefully with retry or SMS fallback
Authentication — add API key validation on the /quotes endpoint
Webhook verification — the app already validates Telnyx Ed25519 signatures; ensure your public key is current
Prompt tuning — test different system prompts and temperature settings for your product catalog
Rate limiting — protect your webhook endpoint from abuse
Monitoring — add structured logging and alerting on call success/failure rates
CRM integration — pipe extracted quotes into Salesforce, HubSpot, or your billing system

Frequently Asked Questions

Can I use a different AI model? Yes. Telnyx AI Inference is OpenAI-compatible. Set AI_MODEL in .env to any model available on the platform. Llama 3.3 70B Instruct is a good default for structured quoting.

How much does a call cost? Telnyx Call Control is priced per minute. AI Inference is priced per 1K tokens. A typical 4-question quote call uses ~1,500 tokens and ~2 minutes of call time — a few cents per call.

Can multiple people call at the same time? Yes. Each call gets its own entry in active_calls keyed by call_control_id. The app handles concurrent calls without interference.

What happens if the caller doesn't speak? The gather step times out after 20 seconds. The app re-prompts: "Could you tell me more about what you need?"

Is the quote accurate? The AI uses the exact pricing from the PRICING dict in its system prompt. Quantities are estimated from the conversation. For production, you may want to add a confirmation step or validate the extracted JSON against the catalog.

Resources

Build an AI Phone Story Hotline — Interactive Storytelling with Telnyx Voice AI

Harpreet Singh Seehra — Wed, 15 Jul 2026 10:32:27 +0000

Build an AI Phone Story Hotline — Interactive Storytelling with Telnyx Voice AI

Imagine calling a phone number, picking a genre — mystery, sci-fi, fantasy, horror, or romance — and listening to an AI narrate a story that adapts to your choices in real time. Press 1 to open the creaking door. Press 2 to check the window. The story branches, the AI continues, and every call is a new adventure.

This is the AI Phone Story Hotline — a 104-line Python app built with Telnyx Call Control and AI Inference. No game engine, no branching script, no pre-written dialogue trees. The AI generates the story as you go, and your phone keypad shapes what happens next.

In this walkthrough, you'll build it from scratch. Clone the repo, configure a phone number, and deploy in minutes.

What You'll Build

A phone number that anyone can call to start an interactive story:

Caller dials in — Telnyx answers and greets them with a genre menu
Caller picks a genre — press 1–5 for mystery, sci-fi, fantasy, horror, or romance
AI generates chapter one — a 3–4 sentence story segment ending with two choices
Caller chooses — press 1 or 2, or speak the choice aloud
Story continues — AI generates the next chapter based on the choice, keeping full conversation context
After 5 chapters — the AI brings the story to a satisfying ending

The whole interaction is voice-driven: Text-to-Speech reads each chapter, and the caller responds with DTMF keypresses or speech. The AI model (Llama 3.3 70B via Telnyx AI Inference) handles the storytelling, and Call Control handles the telephony.

Why This Is Interesting

Most AI phone demos are business use cases — book a table, qualify a lead, reset a password. This one is different. It's creative AI — the model is writing fiction in real time, branching based on human input, and delivering it over a phone call. The storytelling format (short chapters, two choices each) keeps the AI responses tight and the call engaging.

It also demonstrates a pattern that works for any conversational AI app: webhook-driven state machine + LLM with conversation memory + voice I/O. Once you see how the story hotline works, you can swap the storytelling prompt for any domain — a choose-your-own-adventure onboarding flow, an interactive quiz, a branching training simulation.

Prerequisites

Python 3.8+
A Telnyx account with funded balance
A Telnyx API key
A Telnyx phone number with voice enabled
A Call Control Application configured with your webhook URL
ngrok for exposing your local server to Telnyx webhooks

The Architecture

  Phone Call
      │
      ▼
  Telnyx Call Control (webhook events)
      │
      ▼
  Flask app (app.py, 104 lines)
      │
      ├──► call.initiated  → answer the call
      ├──► call.answered   → TTS greeting + genre menu
      ├──► call.gather.ended → DTMF/speech input → AI Inference
      ├──► call.speak.ended → gather next choice → AI Inference
      └──► call.hangup     → cleanup session
      │
      ▼
  Telnyx AI Inference (Llama 3.3 70B)
      │
      ▼
  Story chapter (TTS back to caller)

The app is a state machine driven by Telnyx webhook events. Each event triggers the next action — answer, speak, gather, or hang up. The AI Inference call sits in the middle, generating story chapters from the running conversation history.

Step 1: Clone and Configure

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-phone-story-hotline-python
cp .env.example .env
pip install -r requirements.txt

Edit .env with your credentials:

TELNYX_API_KEY=KEY0123456789ABCDEF    # from portal.telnyx.com/api-keys
TELNYX_PUBLIC_KEY=                    # from portal.telnyx.com/api-keys (public key)
STORY_NUMBER=+13105551234              # your Telnyx phone number
AI_MODEL=meta-llama/Llama-3.3-70B-Instruct

Step 2: Understand the Code

The entire app is 104 lines in a single file: app.py. Here are the key pieces.

Webhook Signature Verification

Every Telnyx webhook is signed with an Ed25519 key. The app verifies the signature before trusting any event:

@app.route("/webhooks/voice", methods=["POST"])
def handle_voice():
    try:
        client.webhooks.unwrap(request.get_data(as_text=True), headers=dict(request.headers))
    except Exception:
        return jsonify({"error": "invalid signature"}), 401

This prevents spoofed webhook calls from triggering call actions.

The State Machine

Each call is tracked in an in-memory dict keyed by call_control_id:

active_calls = {}

The state machine handles five events:

Call initiated — store the call state and answer:

if event_type == "call.initiated" and p.get("direction") == "incoming":
    active_calls[ccid] = {"state": "genre_select", "conversation": [], "chapters": 0}
    client.calls.actions.answer(ccid)

Call answered — greet the caller with the genre menu using Text-to-Speech:

elif event_type == "call.answered":
    client.calls.actions.speak(ccid,
        payload="Welcome to Story Hotline! Choose your adventure. Press 1 for Mystery, 2 for Sci-Fi, 3 for Fantasy, 4 for Horror, 5 for Romance.",
        voice="female", language_code="en-US")

Speak ended — after TTS finishes, gather the caller's input. During genre selection, gather DTMF digits. During the story, gather speech or DTMF:

elif event_type == "call.speak.ended" and call:
    if call["state"] == "genre_select":
        client.calls.actions.gather(ccid, input_type="dtmf", timeout_secs=10, min_digits=1, max_digits=1)
    else:
        client.calls.actions.gather(ccid, input_type="speech dtmf", end_silence_timeout_secs=3, timeout_secs=20, language_code="en-US")

The Storytelling Prompt

When the caller picks a genre, the app builds the system prompt that guides the AI for the rest of the call:

GENRES = {"1": "mystery", "2": "sci-fi", "3": "fantasy", "4": "horror", "5": "romance"}

if call["state"] == "genre_select":
    genre = GENRES.get(digits, "mystery")
    call["state"] = "story"
    call["conversation"] = [{"role": "system", "content":
        f"You are an interactive {genre} storyteller on a phone hotline. "
        f"Tell a gripping story in short chapters (3-4 sentences each). "
        f"End each chapter with exactly TWO choices: 'Press 1 to...' or 'Press 2 to...'. "
        f"Make it vivid and cinematic. After 5 chapters, bring the story to a satisfying ending."
    }]
    story_start = call_inference(call["conversation"] + [{"role": "user", "content": "Begin the story."}])
    call["conversation"].append({"role": "assistant", "content": story_start})
    client.calls.actions.speak(ccid, payload=story_start, voice="female", language_code="en-US")

The prompt does three things:

Format constraint — short chapters (3–4 sentences) that fit naturally in a phone call
Choice structure — exactly two options ending with "Press 1" or "Press 2" so the gather step can capture DTMF
Ending condition — after 5 chapters, wrap up the story

AI Inference

The call_inference helper sends the full conversation history to Telnyx AI Inference and returns the model's response:

def call_inference(messages, max_tokens=250):
    resp = requests.post(INFERENCE_URL,
        headers={"Authorization": f"Bearer {TELNYX_API_KEY}", "Content-Type": "application/json"},
        json={"model": AI_MODEL, "messages": messages, "max_tokens": max_tokens, "temperature": 0.9, },
        timeout=20)
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

Temperature is set to 0.9 — high enough for creative storytelling, low enough to keep the narrative coherent. The model is Llama 3.3 70B Instruct, available on Telnyx AI Inference with an OpenAI-compatible API.

The Conversation Loop

Each time the caller makes a choice, the app appends it to the conversation and asks the AI for the next chapter:

elif call["state"] == "story":
    choice = digits or speech
    call["conversation"].append({"role": "user", "content": f"I choose option {choice}"})
    call["chapters"] += 1
    if call["chapters"] >= 5:
        call["conversation"][-1]["content"] += ". Bring the story to a dramatic conclusion."
    continuation = call_inference(call["conversation"])
    call["conversation"].append({"role": "assistant", "content": continuation})
    client.calls.actions.speak(ccid, payload=continuation, voice="female", language_code="en-US")

After 5 chapters, the app injects a final instruction to bring the story to an end. The AI sees the full conversation history on every call, so it maintains narrative continuity across all chapters.

Step 3: Run the App

Start the Flask server:

python app.py

In a separate terminal, expose your local server with ngrok:

ngrok http 5000

Copy the HTTPS URL and configure it in the Telnx Portal:

Go to Call Control Applications
Create or edit your application
Set the Webhook URL to https://<your-ngrok-url>.ngrok.app/webhooks/voice

Assign your Telnyx phone number to this Call Control Application if you haven't already.

Step 4: Call and Play

Call your Telnyx number from any phone. You'll hear:

"Welcome to Story Hotline! Choose your adventure. Press 1 for Mystery, 2 for Sci-Fi, 3 for Fantasy, 4 for Horror, 5 for Romance."

Press a key. The AI generates the first chapter and reads it aloud. At the end, you'll hear two choices. Press 1 or 2 — or speak your choice — and the story continues.

After 5 chapters, the AI wraps up the story and the call ends.

Customizing the Experience

The storytelling system prompt is the control surface. Small changes produce very different experiences:

Change the genres:

GENRES = {"1": "noir detective", "2": "space opera", "3": "post-apocalyptic", "4": "gothic horror", "5": "cozy romance"}

Add more choices per chapter:

f"End each chapter with exactly THREE choices: 'Press 1 to...', 'Press 2 to...', or 'Press 3 to...'."

Make it a different format — a quiz, a therapy session, a training scenario:

f"You are an interactive compliance quiz host on a phone hotline. Ask one multiple-choice question per chapter (3-4 sentences). End with 'Press 1 for A, 2 for B, 3 for C.' After 10 questions, score the caller and give feedback."

The pattern — webhook state machine + LLM with conversation history + voice I/O — works for any branching conversational experience.

Going to Production

This example uses in-memory storage for simplicity. For a production deployment:

Database — replace the in-memory active_calls dict with Redis or PostgreSQL so call state survives restarts
Concurrency — run the app behind gunicorn with multiple workers
Error recovery — handle inference timeouts and call failures gracefully with retry or SMS fallback
Prompt tuning — test different system prompts and temperature settings for your genre
Rate limiting — protect your webhook endpoint from abuse
Monitoring — add structured logging and alerting on call success/failure rates

Frequently Asked Questions

How much does a call cost? Telnyx Call Control is priced per minute. AI Inference is priced per 1K tokens. A 5-chapter story typically uses ~2,000 tokens and ~3 minutes of call time — a few cents per call.

Can multiple people call at the same time? Yes. Each call gets its own entry in active_calls keyed by call_control_id. The app handles concurrent calls without interference.

What happens if the caller doesn't choose? The gather step times out after 10 seconds (genre selection) or 20 seconds (story choices). The app re-prompts the caller to press 1 or 2.

Can the caller speak instead of pressing keys? Yes. During story choices, the gather accepts both DTMF and speech. The caller can say "one" or "two" instead of pressing keys.

DEV Community: Harpreet Singh Seehra

AI Voice Agent with Function Calling — Calling Real APIs Mid-Conversation in Python

The Problem: Voice Agents That Lie

What It Does

The Architecture

The Tools: OpenAI-Style Function Calling

The Conversation Loop

The Inference Function: Tool-Calling Loop

What the Original Sample Got Wrong

One API Key for Voice, AI, and TTS

The gather_using_ai vs gather Distinction

Environment Variables

Try It Yourself

Browser Phone Calls with Real-Time AI Coaching

The Problem: Sales Calls Without Feedback

What It Does

The Architecture

The Backend: Flask in 75 Lines

WebRTC Telephony Credentials

AI Coaching Endpoint

The Frontend: Browser WebRTC + SpeechRecognition

WebRTC Connection with TelnyxRTC

Browser SpeechRecognition for Live Transcript

The AI Coaching Loop

Call Timer and Status Badges

One API Key for WebRTC and AI

Environment Variables

Try It Yourself

From Phone Call to Formatted Email in 80 Lines of Python — AI Voice Memo Cleanup with Telnyx

The Voice Memo Problem

What It Does

The Architecture

The Call Flow State Machine

AI-Powered Memo Cleanup

Email Delivery via Telnyx Messaging

Webhook Signature Verification

Try It Yourself

Build a Real-Time AI Translation Bridge for Phone Calls in 141 Lines of Python

What It Does

The Architecture

The Language Code Bug

The Translation Call

Webhook Signature Verification

The Call Loop

Try It Yourself

Route Phone Calls by Geography at the Carrier Edge — One Number, Three Regions, Three Compliant Experiences

Route Phone Calls by Geography at the Carrier Edge

The Problem With Global Phone Numbers

What It Does

The Architecture

Region Detection: The Prefix Match

Per-Region Config

The voice ID gotcha

The GDPR Consent Flow

State Management Without a Database

Webhook Signature Verification

Recording Archival to Telnyx Storage

AI Inference: Multilingual Replies

Try It Yourself

Build an AI Video Dubbing Pipeline with STT, LLM, and TTS on One Network

What You'll Build

Why This Is Interesting

Prerequisites

The Architecture

Step 1: Clone and Configure

Step 2: Understand the Code

The Voice Pool

The STT Helper

The LLM Labels Speakers AND Translates in One Call

The TTS Helper

The Async Pipeline

The Audio Download Endpoint

Step 3: Run the App

Step 4: Dub a Clip

Customizing the Pipeline

Going to Production

Frequently Asked Questions

Resources

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

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

The `gather_using_ai` vs `gather` Distinction