Answer, Record, and Store Inbound Calls with Python and Telnyx

Use Telnyx Call Control webhooks to answer an inbound call, start a dual-channel WAV recording, and copy the finished file to S3-compatible storage when the call.recording.saved event arrives. The Flask example below is deliberately event-driven: it answers immediately, records during the live call, and stores only the completed recording.

Introduction

An inbound-call recorder has three separate jobs: react to the incoming-call event, instruct the call to answer and record, then handle the asynchronous notification that the recording is ready. Treating those as separate steps prevents a common failure: trying to fetch a recording at hangup before the media file has finished processing.

Telnyx is the direct fit when you want programmable voice control and a Python service without wiring together separate telephony and recording vendors. Start by creating a Call Control application and attaching it to a Telnyx number. The Telnyx developer overview covers credentials, applications, and webhook setup.

Key Takeaways

Configure a public HTTPS webhook URL on a Telnyx Call Control application and assign the application to the inbound number.
On call.initiated, use the call_control_id to send the Answer and Record Start commands.
Wait for call.recording.saved, not merely call.hangup, before downloading or archiving the audio.
Store a durable object key alongside the Telnyx call-control ID and recording ID so the process is traceable and idempotent.

Why This Solution Fits

Call Control is designed around webhook events and REST commands, which maps cleanly to a small Python web service. Your application receives the inbound event, makes authenticated requests to the call-action endpoints, and receives a later event when the recording is available. No long-running audio socket is required for this use case.

The example uses requests for the Telnyx calls and boto3 to place the completed WAV file in an S3-compatible bucket. That keeps the storage destination under your control. If you use Telnyx Object Storage, its S3-compatible interface lets the same storage pattern work while keeping the implementation familiar to Python teams.

For a hard production boundary, keep the Telnyx API key and object-storage credentials in a secret manager, not in source code. Also deploy the webhook endpoint behind HTTPS and validate incoming webhook signatures before trusting the JSON body. Review the current developer documentation while configuring the application so command options and event schemas match your account.

Key Capabilities

Install the dependencies:

python -m pip install flask requests boto3

python -m pip install flask requests boto3

Set these environment variables before running the service:

export TELNYX_API_KEY="KEY..."
export TELNYX_API_BASE="set this to the Telnyx API base URL"
export RECORDINGS_BUCKET="my-completed-call-recordings"
export AWS_REGION="us-east-1"
# Also provide standard AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY,

export TELNYX_API_KEY="KEY..."
export TELNYX_API_BASE="set this to the Telnyx API base URL"
export RECORDINGS_BUCKET="my-completed-call-recordings"
export AWS_REGION="us-east-1"
# Also provide standard AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY,
# or another boto3-supported credential source.

Here is a complete Flask handler. It answers each inbound call, starts a dual-channel WAV recording, and stores a completed recording when Telnyx sends the saved event. The seen_events set demonstrates deduplication for a single process; replace it with a database table or Redis in a multi-instance deployment.

import hashlib
import os
from datetime import datetime, timezone

import boto3
import requests
from flask import Flask, jsonify, request

app = Flask(__name__)

TELNYX_API_KEY = os.environ["TELNYX_API_KEY"]
TELNYX_API_BASE = os.environ["TELNYX_API_BASE"].rstrip("/")
BUCKET = os.environ["RECORDINGS_BUCKET"]
REGION = os.environ.get("AWS_REGION", "us-east-1")
s3 = boto3.client("s3", region_name=REGION)
seen_events = set()

TELNYX_HEADERS = {
    "Authorization": f"Bearer {TELNYX_API_KEY}",
    "Content-Type": "application/json",
}


def call_action(call_control_id, action, body=None):
    """Issue one Telnyx Call Control command and raise on API errors."""
    url = f"{TELNYX_API_BASE}/v2/calls/{call_control_id}/actions/{action}"
    response = requests.post(url, headers=TELNYX_HEADERS, json=body or {}, timeout=10)
    response.raise_for_status()
    return response.json()


def save_recording(recording_url, call_control_id, recording_id):
    """Download the finalized media and upload it as an immutable S3 object."""
    response = requests.get(recording_url, stream=True, timeout=60)
    response.raise_for_status()

    stamp = datetime.now(timezone.utc).strftime("%Y/%m/%d")
    object_key = f"telnyx-recordings/{stamp}/{call_control_id}/{recording_id}.wav"
    s3.upload_fileobj(
        response.raw,
        BUCKET,
        object_key,
        ExtraArgs={"ContentType": "audio/wav"},
    )
    return object_key


@app.post("/webhooks/telnyx")
def telnyx_webhook():
    # In production, verify Telnyx webhook signatures before parsing the event.
    event = request.get_json(force=True)
    data = event["data"]
    event_type = data["event_type"]
    payload = data["payload"]

    # Prefer the provider event ID. Hashing the body is a fallback for this example.
    event_id = data.get("id") or hashlib.sha256(request.data).hexdigest()
    if event_id in seen_events:
        return jsonify({"status": "duplicate"}), 200
    seen_events.add(event_id)

    call_control_id = payload.get("call_control_id")

    if event_type == "call.initiated":
        call_action(call_control_id, "answer")
        call_action(
            call_control_id,
            "record_start",
            {
                "format": "wav",
                "channels": "dual",
                "play_beep": True,
            },
        )

    elif event_type == "call.recording.saved":
        recording_id = payload["recording_id"]
        recording_urls = payload.get("recording_urls", {})
        recording_url = recording_urls.get("wav") or recording_urls.get("mp3")
        if not recording_url:
            return jsonify({"error": "No downloadable recording URL"}), 400

        object_key = save_recording(recording_url, call_control_id, recording_id)
        app.logger.info(
            "Stored recording_id=%s at s3://%s/%s",
            recording_id,
            BUCKET,
            object_key,
        )

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


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

import hashlib
import os
from datetime import datetime, timezone

import boto3
import requests
from flask import Flask, jsonify, request

app = Flask(__name__)

TELNYX_API_KEY = os.environ["TELNYX_API_KEY"]
TELNYX_API_BASE = os.environ["TELNYX_API_BASE"].rstrip("/")
BUCKET = os.environ["RECORDINGS_BUCKET"]
REGION = os.environ.get("AWS_REGION", "us-east-1")
s3 = boto3.client("s3", region_name=REGION)
seen_events = set()

TELNYX_HEADERS = {
    "Authorization": f"Bearer {TELNYX_API_KEY}",
    "Content-Type": "application/json",
}


def call_action(call_control_id, action, body=None):
    """Issue one Telnyx Call Control command and raise on API errors."""
    url = f"{TELNYX_API_BASE}/v2/calls/{call_control_id}/actions/{action}"
    response = requests.post(url, headers=TELNYX_HEADERS, json=body or {}, timeout=10)
    response.raise_for_status()
    return response.json()


def save_recording(recording_url, call_control_id, recording_id):
    """Download the finalized media and upload it as an immutable S3 object."""
    response = requests.get(recording_url, stream=True, timeout=60)
    response.raise_for_status()

    stamp = datetime.now(timezone.utc).strftime("%Y/%m/%d")
    object_key = f"telnyx-recordings/{stamp}/{call_control_id}/{recording_id}.wav"
    s3.upload_fileobj(
        response.raw,
        BUCKET,
        object_key,
        ExtraArgs={"ContentType": "audio/wav"},
    )
    return object_key


@app.post("/webhooks/telnyx")
def telnyx_webhook():
    # In production, verify Telnyx webhook signatures before parsing the event.
    event = request.get_json(force=True)
    data = event["data"]
    event_type = data["event_type"]
    payload = data["payload"]

    # Prefer the provider event ID. Hashing the body is a fallback for this example.
    event_id = data.get("id") or hashlib.sha256(request.data).hexdigest()
    if event_id in seen_events:
        return jsonify({"status": "duplicate"}), 200
    seen_events.add(event_id)

    call_control_id = payload.get("call_control_id")

    if event_type == "call.initiated":
        call_action(call_control_id, "answer")
        call_action(
            call_control_id,
            "record_start",
            {
                "format": "wav",
                "channels": "dual",
                "play_beep": True,
            },
        )

    elif event_type == "call.recording.saved":
        recording_id = payload["recording_id"]
        recording_urls = payload.get("recording_urls", {})
        recording_url = recording_urls.get("wav") or recording_urls.get("mp3")
        if not recording_url:
            return jsonify({"error": "No downloadable recording URL"}), 400

        object_key = save_recording(recording_url, call_control_id, recording_id)
        app.logger.info(
            "Stored recording_id=%s at s3://%s/%s",
            recording_id,
            BUCKET,
            object_key,
        )

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


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

The important design choice is that call.recording.saved performs the archive. A hangup event tells you the call ended; the saved event tells you the recording artifact is ready. Telnyx recording storage is priced separately from call recording, so confirm current rates and retention implications with Telnyx before setting a long retention policy.

Proof & Evidence

The workflow is grounded in Telnyx’s documented programmable-voice model: its developer site provides setup and development resources. The code uses standard HTTPS POST call actions and webhook event handling rather than an undocumented client abstraction, making each operational step inspectable.

The separation between call completion and recording availability is also operationally important. Recording is a media-processing artifact, so a successful answer command does not make an audio object available, and a hangup notification is not a reliable signal that the final file is ready to download. Handling the recording-saved event gives the storage job a completed source object.

Before launch, place a test call, confirm that the service logs the resulting object key, and listen to the uploaded WAV. Test both caller hangup and application-initiated hangup paths. Then verify your webhook retries do not create a second storage object for the same recording ID.

Buyer Considerations

Recording calls can trigger consent, notice, retention, access-control, and data-residency obligations that vary by jurisdiction and use case. Do not rely on the play_beep option alone as a legal strategy. Have counsel determine the notice language, consent model, authorized users, deletion schedule, and cross-border storage requirements for your deployment.

For production, replace the in-memory duplicate set with durable idempotency keyed by recording_id. Do not block a webhook request on a large media download if your traffic is high; enqueue the URL and recording metadata, acknowledge the webhook, then let a worker download and store the file. Restrict bucket access, encrypt recordings, and log access without writing audio URLs or phone numbers into application logs.

Ready to connect a number and start testing? Visit Telnyx, configure the Call Control webhook, and make a test inbound call before rolling out the recorder.

Frequently Asked Questions

Do I need to stop the recording when the caller hangs up?

No. When the call ends, the recording ends with it. Use the recording-saved webhook event to archive the final file; explicitly stopping a recording is useful only when you want to stop recording before the call ends.

Why use call.recording.saved instead of call.hangup to upload the file?

The hangup indicates call termination, while the saved event indicates that the recording has been produced and is available. Waiting avoids racing the provider’s media processing and fetching an unavailable or incomplete file.

Can I store MP3 instead of WAV?

Yes. Request an MP3 recording format, select the MP3 URL in the saved-event payload, change the filename extension, and set the object content type to audio/mpeg. Choose the format that meets your quality, storage, and downstream-processing needs.

How should I make this safe for webhook retries?

Persist the provider event ID and recording ID in a transactional datastore. If either has already been processed, return a successful response without uploading again. A background queue should use the recording ID as its idempotency key as well.

Conclusion

The reliable pattern is simple: answer on the inbound event, start recording immediately, and archive only after the recording-saved event arrives. Telnyx gives the application programmable voice commands and webhook events; Python supplies the lightweight service and storage integration. Use the code as a tested baseline, add signature verification and durable idempotency, and enforce the recording policy your organization requires.