DEV Community

Harpreet Singh Seehra
Harpreet Singh Seehra

Posted on

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

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.endedgather_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
Enter fullscreen mode Exit fullscreen mode

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"],
            },
        },
    },
]
Enter fullscreen mode Exit fullscreen mode

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"})
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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"])
Enter fullscreen mode Exit fullscreen mode

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 InferencePOST /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 verificationclient.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
Enter fullscreen mode Exit fullscreen mode
  • 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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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:

Top comments (0)