Outbound Call with Python and Flask

Overview

Build a production-ready Flask application that initiates outbound phone calls using the Telnyx Voice API (Call Control). This tutorial covers the command-event model at the heart of Call Control: you send a dial command via the API, then receive webhook events as the call progresses through its lifecycle. By the end, you'll have an endpoint that places calls and a webhook handler that reacts to call events in real time.

Prerequisites

• Python 3.8 or higher.
• A Telnyx account with an active API key from the Telnyx Portal.
• A Telnyx phone number assigned to a Call Control Application.
• A Call Control Application configured in the Telnyx Portal with a webhook URL pointing to your server.
• pip (Python package manager).
• A publicly accessible URL for webhooks (use ngrok for local development).

Step 1: Setup

Install the required dependencies:

pip install telnyx flask python-dotenv

Create a project directory and navigate into it:

mkdir telnyx-outbound-call
cd telnyx-outbound-call

If you're developing locally, start ngrok in a separate terminal to expose your Flask server:

ngrok http 5000

Copy the HTTPS forwarding URL (e.g., https://abc123.ngrok.io) and set it as the webhook URL in your Call Control Application settings in the Telnyx Portal.

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_CALL_CONTROL_APP_ID_HERE

Replace the placeholder values:

• YOUR_API_KEY_HERE — your API key from the Telnyx Portal.
• +15551234567 — your Telnyx phone number in E.164 format.
• YOUR_CALL_CONTROL_APP_ID_HERE — the Connection ID (Call Control Application ID) from the Telnyx Portal. This is a static configuration value, not a per-call identifier.

Step 3: Implementation

Create app.py and start by initializing the Telnyx client and defining a helper function that places the outbound call:

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 place_outbound_call(to_number: str) -> dict:
    """Dial an outbound call 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)")

    # Initiate the call — connection_id is your Call Control App ID (static config).
    # call_control_id is RETURNED in the response, never passed as input to dial().
    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,
        "from": from_number,
        "to": to_number,
        "status": "dialing",
    }

Next, add the webhook handler. Telnyx delivers events as the call progresses — you respond with commands to control the call flow:

def handle_call_event(event_type: str, payload: dict) -> str:
    """Process a Call Control webhook event and issue follow-up commands."""
    call_control_id = payload.get("call_control_id")

    if event_type == "call.initiated":
        # Outbound call is ringing — no action needed, wait for answer
        return f"Call {call_control_id} is ringing"

    elif event_type == "call.answered":
        # Call connected — speak a greeting using TTS
        client.calls.actions.speak(
            call_control_id,
            payload_="Hello! This is an automated call from Telnyx. Goodbye.",
            language="en-US",
            voice="female",
        )
        return f"Greeting spoken on call {call_control_id}"

    elif event_type == "call.speak.ended":
        # TTS finished — hang up the call cleanly
        client.calls.actions.hangup(call_control_id)
        return f"Hanging up call {call_control_id}"

    elif event_type == "call.hangup":
        # Call ended — clean up any resources
        return f"Call {call_control_id} ended"

    else:
        return f"Unhandled event: {event_type}"

Step 4: Testing

Add the Flask routes with comprehensive error handling, then start the server:

from flask import Flask, jsonify, request

app = Flask(__name__)


@app.route("/calls/dial", methods=["POST"])
def dial_endpoint():
    """HTTP endpoint to initiate an 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 = place_outbound_call(to_number)
        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("/webhooks/call-control", methods=["POST"])
def call_control_webhook():
    """Receive and process Call Control webhook events from Telnyx."""
    event = request.get_json()

    if not event:
        return jsonify({"error": "Empty webhook payload"}), 400

    # Telnyx webhook payloads nest data under "data" → "event_type" and "data" → "payload"
    event_data = event.get("data", {})
    event_type = event_data.get("event_type", "unknown")
    payload = event_data.get("payload", {})

    try:
        message = handle_call_event(event_type, payload)
        return jsonify({"status": "ok", "message": message}), 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


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

Start the server:

python app.py

Test the dial endpoint using curl:

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

Expected response:

{
  "call_control_id": "v3:abc123def456-7890-abcd-ef01-234567890abc",
  "from": "+15551234567",
  "to": "+15559876543",
  "status": "dialing"
}

Once the call connects, Telnyx delivers webhook events to /webhooks/call-control. You'll see the call progress through call.initiated → call.answered → call.speak.ended → call.hangup in your Flask logs.

Complete Code

#!/usr/bin/env python3
"""Production-ready Flask app for outbound calls via Telnyx Call Control."""

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"))


def place_outbound_call(to_number: str) -> dict:
    """Dial an outbound call 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)")

    # Initiate the call — connection_id is your Call Control App ID (static config).
    # call_control_id is RETURNED in the response, never passed as input to dial().
    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,
        "from": from_number,
        "to": to_number,
        "status": "dialing",
    }


def handle_call_event(event_type: str, payload: dict) -> str:
    """Process a Call Control webhook event and issue follow-up commands."""
    call_control_id = payload.get("call_control_id")

    if event_type == "call.initiated":
        # Outbound call is ringing — no action needed, wait for answer
        return f"Call {call_control_id} is ringing"

    elif event_type == "call.answered":
        # Call connected — speak a greeting using TTS
        client.calls.actions.speak(
            call_control_id,
            payload_="Hello! This is an automated call from Telnyx. Goodbye.",
            language="en-US",
            voice="female",
        )
        return f"Greeting spoken on call {call_control_id}"

    elif event_type == "call.speak.ended":
        # TTS finished — hang up the call cleanly
        client.calls.actions.hangup(call_control_id)
        return f"Hanging up call {call_control_id}"

    elif event_type == "call.hangup":
        # Call ended — clean up any resources
        return f"Call {call_control_id} ended"

    else:
        return f"Unhandled event: {event_type}"


@app.route("/calls/dial", methods=["POST"])
def dial_endpoint():
    """HTTP endpoint to initiate an 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 = place_outbound_call(to_number)
        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("/webhooks/call-control", methods=["POST"])
def call_control_webhook():
    """Receive and process Call Control webhook events from Telnyx."""
    event = request.get_json()

    if not event:
        return jsonify({"error": "Empty webhook payload"}), 400

    # Telnyx webhook payloads nest data under "data" → "event_type" and "data" → "payload"
    event_data = event.get("data", {})
    event_type = event_data.get("event_type", "unknown")
    payload = event_data.get("payload", {})

    try:
        message = handle_call_event(event_type, payload)
        return jsonify({"status": "ok", "message": message}), 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


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

Troubleshooting

Next Steps

• Handle Inbound Calls with Webhooks.
• Record Calls with Telnyx Call Control.
• Build an IVR Menu with DTMF Input.