Outbound Call with Python and Flask

Overview

Build a production-ready Flask endpoint that initiates outbound calls using the Telnyx Voice API. This tutorial demonstrates the Call Control command-event model, proper webhook handling, and secure credential management for voice applications.

Prerequisites

Python 3.8 or higher.
A Telnyx account with an active API key from the Telnyx Portal.
A Telnyx phone number enabled for outbound calling.
A Call Control Application configured in the Telnyx Portal.
pip (Python package manager).
ngrok or similar tool for webhook testing (optional but recommended).

Step 1: Setup

Install the required dependencies:

pip install telnyx flask python-dotenv

Create a project directory and navigate into it:

mkdir telnyx-outbound-calls
cd telnyx-outbound-calls

Step 2: Configuration

Create a .env file in your project root to store credentials securely:

TELNYX_API_KEY=YOUR_API_KEY_HERE
TELNYX_PHONE_NUMBER=+15551234567
TELNYX_CONNECTION_ID=YOUR_CONNECTION_ID_HERE

Replace YOUR_API_KEY_HERE with your actual API key, +15551234567 with your Telnyx phone number in E.164 format, and YOUR_CONNECTION_ID_HERE with your Call Control Application ID from the Telnyx Portal.

Step 3: Implementation

Create app.py and initialize the Telnyx client. Define helper functions for call management with proper validation:

import os
import telnyx
from dotenv import load_dotenv

load_dotenv()

# Initialize client with the new SDK pattern
client = telnyx.Telnyx(api_key=os.getenv("TELNYX_API_KEY"))


def initiate_call(to_number: str) -> dict:
    """Initiate outbound call via Telnyx and return JSON-serializable response data."""
    from_number = os.getenv("TELNYX_PHONE_NUMBER")
    connection_id = os.getenv("TELNYX_CONNECTION_ID")

    if not from_number:
        raise ValueError("TELNYX_PHONE_NUMBER environment variable not set")
    if not connection_id:
        raise ValueError("TELNYX_CONNECTION_ID environment variable not set")

    # Validate E.164 format to prevent API errors
    if not to_number.startswith("+"):
        raise ValueError("Phone number must be in E.164 format (e.g., +15551234567)")

    # Use client.calls.dial() with connection_id (NOT call_control_id)
    response = client.calls.dial(
        from_=from_number,
        to=to_number,
        connection_id=connection_id,
    )

    # Extract serializable data — SDK objects are NOT JSON-serializable
    return {
        "call_control_id": response.data.call_control_id,
        "call_session_id": response.data.call_session_id,
        "from": from_number,
        "to": to_number,
        "status": "initiated",
    }


def get_call_status(call_control_id: str) -> dict:
    """Retrieve call status and return JSON-serializable response data."""
    response = client.calls.retrieve_status(call_control_id)

    return {
        "call_control_id": response.data.call_control_id,
        "is_alive": response.data.is_alive,
        "call_session_id": response.data.call_session_id,
    }


def hangup_call(call_control_id: str) -> dict:
    """Hangup active call and return confirmation."""
    response = client.calls.actions.hangup(call_control_id)

    return {
        "call_control_id": response.data.call_control_id,
        "status": "hangup_requested",
    }

Step 4: Testing

Add Flask routes with comprehensive error handling and webhook support:

from flask import Flask, jsonify, request

app = Flask(__name__)

# Store active calls for demo purposes (use database in production)
active_calls = {}


@app.route("/calls/initiate", methods=["POST"])
def initiate_call_endpoint():
    """HTTP endpoint to initiate outbound call."""
    data = request.get_json()

    if not data:
        return jsonify({"error": "Request body required"}), 400

    to_number = data.get("to")

    if not to_number:
        return jsonify({"error": "Missing required field: 'to'"}), 400

    try:
        result = initiate_call(to_number)
        # Store call for tracking (use database in production)
        active_calls[result["call_control_id"]] = result
        return jsonify(result), 200

    except telnyx.AuthenticationError:
        return jsonify({"error": "Invalid API key"}), 401
    except telnyx.RateLimitError:
        return jsonify({"error": "Rate limit exceeded. Please slow down."}), 429
    except telnyx.APIStatusError as e:
        return jsonify({"error": str(e), "status_code": e.status_code}), e.status_code
    except telnyx.APIConnectionError:
        return jsonify({"error": "Network error connecting to Telnyx"}), 503
    except ValueError as e:
        return jsonify({"error": str(e)}), 400


@app.route("/calls/<call_control_id>/status", methods=["GET"])
def get_call_status_endpoint(call_control_id):
    """HTTP endpoint to retrieve call status."""
    try:
        result = get_call_status(call_control_id)
        return jsonify(result), 200

    except telnyx.AuthenticationError:
        return jsonify({"error": "Invalid API key"}), 401
    except telnyx.APIStatusError as e:
        return jsonify({"error": str(e), "status_code": e.status_code}), e.status_code
    except telnyx.APIConnectionError:
        return jsonify({"error": "Network error connecting to Telnyx"}), 503


@app.route("/calls/<call_control_id>/hangup", methods=["POST"])
def hangup_call_endpoint(call_control_id):
    """HTTP endpoint to hangup active call."""
    try:
        result = hangup_call(call_control_id)
        # Remove from active calls tracking
        active_calls.pop(call_control_id, None)
        return jsonify(result), 200

    except telnyx.AuthenticationError:
        return jsonify({"error": "Invalid API key"}), 401
    except telnyx.APIStatusError as e:
        return jsonify({"error": str(e), "status_code": e.status_code}), e.status_code
    except telnyx.APIConnectionError:
        return jsonify({"error": "Network error connecting to Telnyx"}), 503


@app.route("/webhooks/voice", methods=["POST"])
def voice_webhook():
    """Handle Telnyx voice webhooks for call events."""
    data = request.get_json()

    if not data:
        return jsonify({"error": "Invalid webhook payload"}), 400

    event_type = data.get("data", {}).get("event_type")
    call_control_id = data.get("data", {}).get("payload", {}).get("call_control_id")

    print(f"Received webhook: {event_type} for call {call_control_id}")

    # Handle different call events
    if event_type == "call.answered":
        print(f"Call {call_control_id} was answered")
    elif event_type == "call.hangup":
        print(f"Call {call_control_id} ended")
        # Clean up call tracking
        active_calls.pop(call_control_id, None)
    elif event_type == "call.initiated":
        print(f"Call {call_control_id} was initiated")

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


if __name__ == "__main__":
    app.run(debug=True, port=5000)

Start the server:

python app.py

Test the endpoint using curl:

curl -X POST http://localhost:5000/calls/initiate \
  -H "Content-Type: application/json" \
  -d '{"to": "+15559876543"}'

Expected response:

{
  "call_control_id": "v3:T02llQxIyaRkhfRKxgAP8nY511EhFLizdvdUKJKHjsUBrQ",
  "call_session_id": "428c31b6-7a59-4cf0-8b2b-8b2b8b2b8b2b",
  "from": "+15551234567",
  "to": "+15559876543",
  "status": "initiated"
}

Complete Code

#!/usr/bin/env python3
"""Production-ready Flask endpoint for outbound calls via Telnyx Voice API."""

import os
import telnyx
from dotenv import load_dotenv
from flask import Flask, jsonify, request

load_dotenv()

app = Flask(__name__)

# Initialize client with the new SDK pattern
client = telnyx.Telnyx(api_key=os.getenv("TELNYX_API_KEY"))

# Store active calls for demo purposes (use database in production)
active_calls = {}


def initiate_call(to_number: str) -> dict:
    """Initiate outbound call via Telnyx and return JSON-serializable response data."""
    from_number = os.getenv("TELNYX_PHONE_NUMBER")
    connection_id = os.getenv("TELNYX_CONNECTION_ID")

    if not from_number:
        raise ValueError("TELNYX_PHONE_NUMBER environment variable not set")
    if not connection_id:
        raise ValueError("TELNYX_CONNECTION_ID environment variable not set")

    # Validate E.164 format to prevent API errors
    if not to_number.startswith("+"):
        raise ValueError("Phone number must be in E.164 format (e.g., +15551234567)")

    # Use client.calls.dial() with connection_id (NOT call_control_id)
    response = client.calls.dial(
        from_=from_number,
        to=to_number,
        connection_id=connection_id,
    )

    # Extract serializable data — SDK objects are NOT JSON-serializable
    return {
        "call_control_id": response.data.call_control_id,
        "call_session_id": response.data.call_session_id,
        "from": from_number,
        "to": to_number,
        "status": "initiated",
    }


def get_call_status(call_control_id: str) -> dict:
    """Retrieve call status and return JSON-serializable response data."""
    response = client.calls.retrieve_status(call_control_id)

    return {
        "call_control_id": response.data.call_control_id,
        "is_alive": response.data.is_alive,
        "call_session_id": response.data.call_session_id,
    }


def hangup_call(call_control_id: str) -> dict:
    """Hangup active call and return confirmation."""
    response = client.calls.actions.hangup(call_control_id)

    return {
        "call_control_id": response.data.call_control_id,
        "status": "hangup_requested",
    }


@app.route("/calls/initiate", methods=["POST"])
def initiate_call_endpoint():
    """HTTP endpoint to initiate outbound call."""
    data = request.get_json()

    if not data:
        return jsonify({"error": "Request body required"}), 400

    to_number = data.get("to")

    if not to_number:
        return jsonify({"error": "Missing required field: 'to'"}), 400

    try:
        result = initiate_call(to_number)
        # Store call for tracking (use database in production)
        active_calls[result["call_control_id"]] = result
        return jsonify(result), 200

    except telnyx.AuthenticationError:
        return jsonify({"error": "Invalid API key"}), 401
    except telnyx.RateLimitError:
        return jsonify({"error": "Rate limit exceeded. Please slow down."}), 429
    except telnyx.APIStatusError as e:
        return jsonify({"error": str(e), "status_code": e.status_code}), e.status_code
    except telnyx.APIConnectionError:
        return jsonify({"error": "Network error connecting to Telnyx"}), 503
    except ValueError as e:
        return jsonify({"error": str(e)}), 400


@app.route("/calls/<call_control_id>/status", methods=["GET"])
def get_call_status_endpoint(call_control_id):
    """HTTP endpoint to retrieve call status."""
    try:
        result = get_call_status(call_control_id)
        return jsonify(result), 200

    except telnyx.AuthenticationError:
        return jsonify({"error": "Invalid API key"}), 401
    except telnyx.APIStatusError as e:
        return jsonify({"error": str(e), "status_code": e.status_code}), e.status_code
    except telnyx.APIConnectionError:
        return jsonify({"error": "Network error connecting to Telnyx"}), 503


@app.route("/calls/<call_control_id>/hangup", methods=["POST"])
def hangup_call_endpoint(call_control_id):
    """HTTP endpoint to hangup active call."""
    try:
        result = hangup_call(call_control_id)
        # Remove from active calls tracking
        active_calls.pop(call_control_id, None)
        return jsonify(result), 200

    except telnyx.AuthenticationError:
        return jsonify({"error": "Invalid API key"}), 401
    except telnyx.APIStatusError as e:
        return jsonify({"error": str(e), "status_code": e.status_code}), e.status_code
    except telnyx.APIConnectionError:
        return jsonify({"error": "Network error connecting to Telnyx"}), 503


@app.route("/webhooks/voice", methods=["POST"])
def voice_webhook():
    """Handle Telnyx voice webhooks for call events."""
    data = request.get_json()

    if not data:
        return jsonify({"error": "Invalid webhook payload"}), 400

    event_type = data.get("data", {}).get("event_type")
    call_control_id = data.get("data", {}).get("payload", {}).get("call_control_id")

    print(f"Received webhook: {event_type} for call {call_control_id}")

    # Handle different call events
    if event_type == "call.answered":
        print(f"Call {call_control_id} was answered")
    elif event_type == "call.hangup":
        print(f"Call {call_control_id} ended")
        # Clean up call tracking
        active_calls.pop(call_control_id, None)
    elif event_type == "call.initiated":
        print(f"Call {call_control_id} was initiated")

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


if __name__ == "__main__":
    app.run(debug=True, port=5000)

Troubleshooting

Issue	Problem	Solution
Authentication Error (401)	The endpoint returns `{"error": "Invalid API key"}` with HTTP 401.	Verify your `TELNYX_API_KEY` in the `.env` file matches the key shown in the Telnyx Portal. Ensure there are no trailing spaces or quotes. If the key was regenerated recently, update your environment file and restart the Flask server.
Missing Connection ID Error	You receive `ValueError: TELNYX_CONNECTION_ID environment variable not set` when initiating calls.	Ensure you have created a Call Control Application in the Telnyx Portal and copied its Connection ID to your `.env` file as `TELNYX_CONNECTION_ID`. This is different from your API key and phone number—it links your number to the Call Control application.
Invalid Phone Number Format	You receive a 400 error stating "Phone number must be in E.164 format" or a Telnyx API error about invalid destination.	Ensure all phone numbers use E.164 format: start with `+`, followed by country code and number without spaces or dashes. Example: `+15551234567` (US) or `+447700900123` (UK). Update your test curl command to use properly formatted numbers.