Voice AI

From Phone Call to Formatted Email in 80 Lines of Python — AI Voice Memo Cleanup with Telnyx

AI Voice Memo to Email — an 80-line Flask webhook that answers a phone call, gathers a spoken memo, runs it through AI Inference to clean up grammar and extract structure, and delivers a formatted email. One API key for voice, AI, and messaging. No third-party services.

The Voice Memo Problem

Voice memos are the fastest way to capture a thought — you speak, you're done. But what you get is a rambling audio blob that nobody (including you) wants to read later. The raw transcript is worse: no punctuation, false starts, filler words, and no structure. You still have to manually clean it up before it's useful as an email, a status update, or a meeting summary.

The existing solutions split the problem across multiple services. A transcription service converts audio to text. An LLM API cleans up the text. An email service sends the result. Three vendors, three API keys, three bills, three points of failure.

The AI Voice Memo to Email example does all of it on one network — Telnyx Call Control handles the phone call, Telnyx AI Inference cleans up the transcript, and Telnyx Messaging delivers the email. One API key. One Flask file. About 80 lines of Python.

What It Does

You call a Telnyx number. The app answers, speaks a greeting, and starts listening. You dictate your memo — a status update, a meeting summary, a bug report, whatever — and press # when you're done. The app sends the transcript to AI Inference with a prompt that returns structured JSON: a subject line, a formatted body, and a list of action items. The app sends that as an email to your default address and confirms back on the call: "Memo saved and emailed. Subject: [inferred subject]. Goodbye!"

StepEventAction
1call.initiated (incoming)Answer the call, create session
2call.answeredTTS: "Voice memo. Speak your memo after the tone. Press pound when finished."
3call.speak.endedStart speech gather (120s timeout, # terminates)
4call.gather.endedSend transcript to AI Inference → get structured JSON → send email → TTS confirmation
5call.hangupClean up session

The memo is also stored in memory and accessible via GET /memos — so even if email delivery isn't configured, the formatted memo is still retrievable.

The Architecture

Everything lives in one Flask file. No database, no Redis, no Celery. Call state is tracked in an in-memory dict keyed by call_control_id. Memos are stored in a list. A background thread cleans up expired sessions every 5 minutes (1-hour TTL).

Caller dials your Telnyx number
        ↓
Telnyx sends call.initiated webhook → /webhooks/voice
        ↓
app calls answer() → creates session in active_calls[ccid]
        ↓
Telnyx sends call.answered → app calls speak() with greeting
        ↓
Telnyx sends call.speak.ended → app calls gather(input_type="speech", terminating_digit="#")
        ↓
Caller dictates memo, presses #
        ↓
Telnyx sends call.gather.ended with speech transcript
        ↓
app sends transcript to AI Inference → gets JSON {subject, body, action_items}
        ↓
app sends email via Telnyx Messaging API
        ↓
app calls speak() with confirmation: "Memo saved and emailed. Subject: X. Goodbye!"
        ↓
Telnyx sends call.hangup → app removes session

The Call Flow State Machine

The webhook handler is a state machine driven by Telnyx events. Each event triggers the next action:

@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()
    event_type = payload.get("data", {}).get("event_type")
    data = payload.get("data", {})
    p = data.get("payload", {})
    ccid = p.get("call_control_id")

    if event_type == "call.initiated" and p.get("direction") == "incoming":
        active_calls[ccid] = {"caller": p.get("from"), "raw_text": [], "start": time.time()}
        client.calls.actions.answer(ccid)
        return jsonify({"status": "answering"}), 200

    elif event_type == "call.answered":
        client.calls.actions.speak(ccid,
            payload="Voice memo. Speak your memo after the tone. Press pound when finished.",
            voice="female", language_code="en-US")
        return jsonify({"status": "greeting"}), 200

    elif event_type == "call.speak.ended":
        client.calls.actions.gather(ccid,
            input_type="speech", end_silence_timeout_secs=5, timeout_secs=120,
            language_code="en-US", terminating_digit="#")
        return jsonify({"status": "recording"}), 200

    elif event_type == "call.gather.ended":
        call = active_calls.get(ccid)
        speech = p.get("speech", {}).get("result", "")
        if call and speech:
            call["raw_text"].append(speech)
            # ... AI cleanup + email + confirmation (see next section)
        return jsonify({"status": "processed"}), 200

    elif event_type == "call.hangup":
        active_calls.pop(ccid, None)
        return jsonify({"status": "ended"}), 200

The state machine has five transitions, one per event. The call.initiated handler checks direction == "incoming" to avoid processing outbound call legs. The call.speak.ended handler is what advances from greeting to gathering — Telnyx fires this event when TTS playback finishes, so you know the caller has heard the greeting before the gather starts.

The gather uses end_silence_timeout_secs=5 — if the caller stops speaking for 5 seconds, the gather ends automatically. The timeout_secs=120 caps the total gather at 2 minutes. The terminating_digit="#" lets the caller explicitly signal "I'm done" by pressing pound.

AI-Powered Memo Cleanup

The core of the app is a single inference call that turns rambling speech into structured JSON. The prompt is specific about the output format:

def call_inference(messages, max_tokens=400):
    resp = requests.post(INFERENCE_URL,
        headers={"Authorization": f"Bearer {TELNYX_API_KEY}", "Content-Type": "application/json"},
        json={"model": AI_MODEL, "messages": messages,
              "max_tokens": max_tokens, "temperature": 0.3},
        timeout=15)
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

The system prompt asks for three fields — subject, body, and action_items:

formatted = call_inference([
    {"role": "system", "content":
        "Clean up this voice memo into a well-formatted email. "
        "Fix grammar, add structure (paragraphs, bullets if needed). "
        "Return JSON: subject (string, inferred from content), "
        "body (string, the formatted memo), "
        "action_items (list of strings)."},
    {"role": "user", "content": speech}
])
memo = json.loads(formatted)

Temperature is 0.3 — low enough that the same memo produces roughly the same output every time, but high enough that the AI can infer a reasonable subject line from the content. The max_tokens=400 cap is sufficient for a typical voice memo (a few paragraphs of cleaned-up text).

The AI returns a JSON string. The app parses it into a memo dict, attaches the caller's number and the raw transcript for reference, and appends it to the in-memory memos list:

memo = json.loads(formatted)
memo["caller"] = call["caller"]
memo["raw"] = speech
memo["timestamp"] = time.strftime("%Y-%m-%dT%H:%M:%SZ")
memos.append(memo)

If the AI response isn't valid JSON (model returned text instead of JSON), the except block saves the raw speech and speaks a simpler confirmation — the caller still gets their memo saved, just without the email:

except Exception:
    memos.append({"raw": speech, "caller": call["caller"],
                  "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ")})
    client.calls.actions.speak(ccid, payload="Memo saved. Goodbye!",
        voice="female", language_code="en-US")

This is graceful degradation — the call is never wasted. If AI fails, the raw transcript is preserved. If email fails, the formatted memo is preserved. The caller always gets a confirmation.

Email Delivery via Telnyx Messaging

After the memo is formatted, the app sends it as an email through the Telnyx Messaging API:

def send_email(to, subject, body):
    try:
        requests.post("https://api.telnyx.com/v2/messages",
            headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
                     "Content-Type": "application/json"},
            json={"from": {"email_address": f"memo@{MEMO_NUMBER.replace('+','')}.telnyx.com"},
                  "to": [{"email_address": to}],
                  "subject": subject, "body": body, "type": "email"},
            timeout=15)
    except Exception as e:
        app.logger.error("Email send failed (expected - may need Telnyx email setup): %s", e)

The from address is derived from the memo number — memo@{number}.telnyx.com. The type: "email" field tells the Messaging API to deliver this as an email rather than an SMS. The same TELNYX_API_KEY that answers the call and runs the AI inference also sends the email — one key, one bill, one network.

The email send is wrapped in a try/except because email delivery may require additional Telnyx setup (domain verification, email profile configuration). If it fails, the memo is still saved in the in-memory store and retrievable via GET /memos. The error is logged, not raised — the call continues to the confirmation step.

Webhook Signature Verification

Every Telnyx webhook is signed with an Ed25519 key. The app verifies the signature before processing the event — without this, anyone can POST to your webhook URL and inject fake call events:

try:
    client.webhooks.unwrap(request.get_data(as_text=True), headers=dict(request.headers))
except Exception:
    return jsonify({"error": "invalid signature"}), 401

The webhooks.unwrap() method from the Telnyx Python SDK handles the Ed25519 verification internally — it reads the telnyx-signature-ed25519 and telnyx-timestamp headers, reconstructs the signed payload (<timestamp>|<raw body>), and verifies the signature against the public key configured in the Telnyx client. If verification fails, the request is rejected with 401 before any event processing happens.

The raw body is verified, not the parsed JSON — because JSON parsing is not canonical (key order, whitespace vary), and the signature would fail. request.get_data(as_text=True) returns the raw request body as a string, which is what the SDK needs.

In-Memory State with TTL Cleanup

The app uses two in-memory stores: active_calls (call sessions, keyed by call_control_id) and memos (completed memos, a list). A background thread cleans up expired sessions every 5 minutes:

def _start_ttl_cleanup(*stores, ttl_seconds=3600, interval=300):
    def _cleanup():
        while True:
            _ttl_time.sleep(interval)
            cutoff = _ttl_time.time() - ttl_seconds
            for store in stores:
                expired = [k for k, v in store.items()
                           if isinstance(v, dict) and v.get("_ts", _ttl_time.time()) < cutoff]
                for k in expired:
                    store.pop(k, None)
    threading.Thread(target=_cleanup, daemon=True).start()

_start_ttl_cleanup(active_calls)

Sessions older than 1 hour are evicted. This prevents memory leaks from calls that hang up without firing the call.hangup event (rare, but it happens). For production, replace the in-memory dict with PostgreSQL or Redis — the rest of the code stays the same.

One API Key for Voice, AI, and Messaging

The entire app uses a single TELNYX_API_KEY:

  • Call Control — answer, speak, gather, hangup (via the Telnyx Python SDK)
  • AI Inference — chat completions at POST /v2/ai/chat/completions (via requests.post with the API key as Bearer token)
  • Messaging — send email at POST /v2/messages (via requests.post with the API key as Bearer token)

The SDK handles Call Control. The two requests.post calls handle AI and messaging directly — they're simple HTTP calls to Telnyx API endpoints, authenticated with the same key. No third-party transcription service, no separate LLM provider, no email API key. One network, one key, one bill.

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-voice-memo-to-email-python
cp .env.example .env   # add TELNYX_API_KEY, MEMO_NUMBER, DEFAULT_EMAIL
pip install -r requirements.txt
python app.py           # starts on http://localhost:5000

Then expose your server for webhooks:

ngrok http 5000

Configure your Call Control Application webhook URL to https://<id>.ngrok.io/webhooks/voice in the Telnyx Portal.

Call your Telnyx number. Speak your memo. Press #. Check your email.

Check saved memos:

curl http://localhost:5000/memos | python3 -m json.tool

Check health:

curl http://localhost:5000/health

Key links:

  • Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/ai-voice-memo-to-email-python
  • Telnyx Portal: https://portal.telnyx.com
  • Call Control docs: https://developers.telnyx.com/docs/voice/call-control
  • AI Inference docs: https://developers.telnyx.com/docs/inference
  • Messaging docs: https://developers.telnyx.com/docs/messaging

Ready to build with low-latency voice AI?

Join developers building the future of real-time conversations