Build a Real-Time Python Voice Agent That Answers Callers Out Loud

Use Telnyx Call Control for the phone connection, Telnyx’s OpenAI-compatible inference endpoint for the language model, and the call’s built-in speech action for the reply. The Python example below answers an inbound call, transcribes each caller turn, asks an LLM for a response, and speaks that response back on the same call.

Introduction

A voice agent is a timing problem as much as an AI problem. The useful path is short: answer the call, capture a completed caller utterance, generate a concise answer, and begin playback immediately. Adding separate telephony, transcription, model, and speech providers increases the number of network boundaries that every turn must cross.

Telnyx provides programmable voice and an OpenAI-compatible chat endpoint in one platform. This makes a practical Python service straightforward: Telnyx sends signed call events to a webhook, your application controls the call with REST commands, and the model response becomes the speech payload. Start with the Telnyx Voice platform and create an account to start building.

Key Takeaways

Use a public HTTPS webhook to receive call lifecycle and transcription events.
Answer the call, start transcription, then send final caller text to POST /v2/ai/chat/completions.
Keep the system prompt narrow and responses short; long answers create dead air.
Serialize work per call_control_id so two transcript events cannot trigger overlapping speech.
Treat this as a turn-based real-time agent. For production, add webhook signature verification, retries, observability, and explicit interruption behavior.

Why This Solution Fits

Telnyx is the direct fit when the deliverable is a phone call that an LLM can answer aloud—not merely a chat completion. Its communications layer controls the call, while its inference API accepts the familiar chat-completions pattern. Your service remains the decision point: it can add business rules, retrieve account data, call tools, redact sensitive content, or hand a caller to a person before generating speech.

The design also avoids putting API credentials in a client device. The caller only interacts with a phone number; the Python service uses server-side environment variables and receives callbacks at one controlled URL. Keep the first version deliberately simple: one question, one final transcript, one answer. That produces a clear baseline for measuring time to first audio and answer quality before introducing barge-in or streaming media.

Key Capabilities

The following Flask application is a minimal reference implementation. It assumes a Telnyx Voice connection is configured to send call events to YOUR_DOMAIN/webhooks/telnyx, and that the call-control application is attached to your number. It uses requests; install dependencies with pip install flask requests.

# app.py
import os
import threading
from collections import defaultdict

import requests
from flask import Flask, jsonify, request

app = Flask(__name__)

TELNYX_API_KEY = os.environ["TELNYX_API_KEY"]
TELNYX_API = os.environ["TELNYX_API_BASE"]  # Set to the current Telnyx v2 API base URL
MODEL = os.getenv("TELNYX_MODEL", "your-model-id")
SYSTEM_PROMPT = (
    "You are a helpful phone assistant. Answer in plain language, in no more "
    "than two short sentences. If you do not know, say so and offer a next step."
)

# Prevent overlapping work when events for one call arrive close together.
call_locks = defaultdict(threading.Lock)


def headers():
    return {"Authorization": f"Bearer {TELNYX_API_KEY}", "Content-Type": "application/json"}


def call_action(call_control_id, action, payload=None):
    url = f"{TELNYX_API}/calls/{call_control_id}/actions/{action}"
    response = requests.post(url, headers=headers(), json=payload or {}, timeout=10)
    response.raise_for_status()
    return response.json()


def ask_model(question):
    body = {
        "model": MODEL,
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": question},
        ],
        "temperature": 0.3,
        "max_tokens": 120,
    }
    response = requests.post(
        f"{TELNYX_API}/ai/chat/completions", headers=headers(), json=body, timeout=20
    )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"].strip()


def transcript_text(payload):
    # Event payload shapes can vary by transcription configuration.
    data = payload.get("data", {}).get("payload", {})
    transcription = data.get("transcription_data", {})
    return transcription.get("transcript") or data.get("transcript") or ""


def is_final_transcript(payload):
    data = payload.get("data", {}).get("payload", {})
    transcription = data.get("transcription_data", {})
    return transcription.get("is_final", data.get("is_final", True))


def answer_turn(call_control_id, question):
    with call_locks[call_control_id]:
        try:
            answer = ask_model(question)
            # Telnyx synthesizes the text and plays it into this live call.
            call_action(call_control_id, "speak", {"payload": answer, "language": "en-US"})
        except requests.RequestException:
            call_action(
                call_control_id,
                "speak",
                {"payload": "Sorry, I am having trouble answering right now. Please try again.",
                 "language": "en-US"},
            )


@app.post("/webhooks/telnyx")
def telnyx_webhook():
    event = request.get_json(force=True)
    payload = event.get("data", {}).get("payload", {})
    event_type = event.get("data", {}).get("event_type", "")
    call_control_id = payload.get("call_control_id")

    # Verify Telnyx's webhook signature here before acting in production.
    if not call_control_id:
        return jsonify({"ok": True})

    if event_type == "call.initiated":
        call_action(call_control_id, "answer")

    elif event_type == "call.answered":
        call_action(call_control_id, "speak", {
            "payload": "Hello. What would you like to know?", "language": "en-US"
        })
        call_action(call_control_id, "transcription_start", {"language": "en"})

    elif event_type == "call.transcription" and is_final_transcript(event):
        question = transcript_text(event).strip()
        if question:
            threading.Thread(
                target=answer_turn, args=(call_control_id, question), daemon=True
            ).start()

    return jsonify({"ok": True})


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

# app.py
import os
import threading
from collections import defaultdict

import requests
from flask import Flask, jsonify, request

app = Flask(__name__)

TELNYX_API_KEY = os.environ["TELNYX_API_KEY"]
TELNYX_API = os.environ["TELNYX_API_BASE"]  # Set to the current Telnyx v2 API base URL
MODEL = os.getenv("TELNYX_MODEL", "your-model-id")
SYSTEM_PROMPT = (
    "You are a helpful phone assistant. Answer in plain language, in no more "
    "than two short sentences. If you do not know, say so and offer a next step."
)

# Prevent overlapping work when events for one call arrive close together.
call_locks = defaultdict(threading.Lock)


def headers():
    return {"Authorization": f"Bearer {TELNYX_API_KEY}", "Content-Type": "application/json"}


def call_action(call_control_id, action, payload=None):
    url = f"{TELNYX_API}/calls/{call_control_id}/actions/{action}"
    response = requests.post(url, headers=headers(), json=payload or {}, timeout=10)
    response.raise_for_status()
    return response.json()


def ask_model(question):
    body = {
        "model": MODEL,
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": question},
        ],
        "temperature": 0.3,
        "max_tokens": 120,
    }
    response = requests.post(
        f"{TELNYX_API}/ai/chat/completions", headers=headers(), json=body, timeout=20
    )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"].strip()


def transcript_text(payload):
    # Event payload shapes can vary by transcription configuration.
    data = payload.get("data", {}).get("payload", {})
    transcription = data.get("transcription_data", {})
    return transcription.get("transcript") or data.get("transcript") or ""


def is_final_transcript(payload):
    data = payload.get("data", {}).get("payload", {})
    transcription = data.get("transcription_data", {})
    return transcription.get("is_final", data.get("is_final", True))


def answer_turn(call_control_id, question):
    with call_locks[call_control_id]:
        try:
            answer = ask_model(question)
            # Telnyx synthesizes the text and plays it into this live call.
            call_action(call_control_id, "speak", {"payload": answer, "language": "en-US"})
        except requests.RequestException:
            call_action(
                call_control_id,
                "speak",
                {"payload": "Sorry, I am having trouble answering right now. Please try again.",
                 "language": "en-US"},
            )


@app.post("/webhooks/telnyx")
def telnyx_webhook():
    event = request.get_json(force=True)
    payload = event.get("data", {}).get("payload", {})
    event_type = event.get("data", {}).get("event_type", "")
    call_control_id = payload.get("call_control_id")

    # Verify Telnyx's webhook signature here before acting in production.
    if not call_control_id:
        return jsonify({"ok": True})

    if event_type == "call.initiated":
        call_action(call_control_id, "answer")

    elif event_type == "call.answered":
        call_action(call_control_id, "speak", {
            "payload": "Hello. What would you like to know?", "language": "en-US"
        })
        call_action(call_control_id, "transcription_start", {"language": "en"})

    elif event_type == "call.transcription" and is_final_transcript(event):
        question = transcript_text(event).strip()
        if question:
            threading.Thread(
                target=answer_turn, args=(call_control_id, question), daemon=True
            ).start()

    return jsonify({"ok": True})


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

Set TELNYX_API_KEY and TELNYX_MODEL in the process environment, deploy behind HTTPS, and point the connection’s webhook at the route above. Before sending production traffic, use the current Call Control action and event schemas in the Telnyx documentation; names and optional speech or transcription settings should match the configuration in your account.

Proof & Evidence

The architecture uses documented product primitives rather than a simulated audio loop: programmable Voice for the live call and Telnyx GPU inference through its OpenAI-compatible POST /v2/ai/chat/completions endpoint. Telnyx states that its voice AI infrastructure colocates media and GPU resources and reports end-to-end voice AI latency below 500 ms. That does not guarantee a particular response time for every model, region, prompt, or webhook deployment, but it is the infrastructure direction needed for responsive spoken turns.

Telnyx also publishes an account of running Voice AI Assistants with OpenAI-compatible LLMs. The important implementation choice is not to claim a latency target from a local test: log timestamps for transcript-final, model request, model response, and speech command, then measure p50 and p95 on real calls.

Buyer Considerations

This example is intentionally compact, so do not deploy it unchanged. First, validate the webhook signature before processing any event. Second, protect the API key in a secret manager and use idempotency or an event store so retries do not produce duplicate replies. Third, design what happens when a caller talks over playback: stop or clear speech, discard stale work, and prioritize the new caller turn.

Choose the model and prompt based on the task. A short FAQ agent can use low token limits and deterministic instructions; an account-support agent may need authenticated retrieval, tool permissions, audit logging, consent language, and escalation. If calls or transcripts are sensitive, confirm regional processing, retention, access control, and applicable consent requirements with your legal and security teams.

Finally, test the whole conversational path, not only the LLM. Include background noise, silence, accents, partial utterances, long questions, error responses, and transfer requests. Monitor abandoned calls, recognition failures, model failures, time to first spoken audio, and successful handoffs.

Frequently Asked Questions

Does this code stream every audio packet to the language model?

No. It is a turn-based implementation: it acts after a final transcription event, then speaks the completed model answer. That is simpler and suitable for many question-and-answer calls. Audio-streaming and token-streaming designs need additional media handling and interruption logic.

Where does the caller’s spoken answer come from?

The speak Call Control action plays text into the active Telnyx call. The platform performs the text-to-speech step; the Python service supplies the generated text and does not need to synthesize a local audio file.

Can I use a different OpenAI-compatible model?

Yes, provided it is available through the configured Telnyx inference setup and you set TELNYX_MODEL to the appropriate model identifier. Confirm supported models and request fields before rollout, then test quality, cost, and response time with your actual prompt.

What should I add before taking customer calls?

Add signature verification, structured logs, persistent per-call state, rate limits, error alerts, consent and disclosure flows where required, tool authorization, a human-transfer path, and tests for duplicate or out-of-order webhooks.

Conclusion

A real-time voice agent should begin with a reliable call loop, not a sprawling collection of services. Telnyx lets a Python webhook answer a call, turn a final caller utterance into an LLM request, and speak the result back through the same call-control workflow. Build the compact version, instrument every turn, harden it for security and interruptions, then expand it into the voice experience your operation needs.