Send Two-Way SMS in Python and Track Every Reply by Conversation Thread

Build a reliable two-way SMS flow in Python by sending through the Telnyx Messaging API, receiving message.received webhooks, and storing both directions under one deterministic thread key. The implementation below uses Flask, requests, and SQLite so you can run it locally, inspect every message, and replace the database layer when you are ready to scale.

Introduction

A two-way SMS conversation is not a long-lived API session. It is a sequence of outbound API requests and inbound webhook events. Your application creates the connection between those events. The practical key is the phone-number pair: your Telnyx-enabled number and the customer’s number. Use that pair consistently and every inbound reply lands in the same thread as the original outbound message.

Telnyx’s messaging capabilities support two-way threads, personalization, and reporting. That makes the API only half of the solution; the other half is a small, idempotent message ledger in your application. The ledger should record the provider message ID, direction, normalized participants, body, timestamp, and thread ID.

This guide deliberately uses direct HTTPS calls instead of hiding the request behind an SDK. You can see the endpoint, authorization header, and fields your production service must own. Once the flow works, move the same logic behind a queue and a managed database.

Prerequisites

Before sending a message, have the following ready:

• A Telnyx account, an API key, and a messaging-capable phone number attached to a Messaging Profile.
• A Messaging Profile configured to deliver inbound messaging webhooks to a public HTTPS endpoint. Review the Telnyx developer platform for current Messaging Profile configuration details.
• Python 3.10 or newer, plus Flask and requests:

python3 -m venv .venv
source .venv/bin/activate
pip install flask requests

• A tunnel such as a secure development tunnel for local testing, or a deployed HTTPS URL, so Telnyx can reach /webhooks/telnyx.
• Explicit consent from every recipient, an appropriate sender registration for your traffic, and clear opt-out handling. For US application-to-person messaging, review Telnyx’s compliance guidance before you scale.

Set secrets and numbers as environment variables; never hard-code them in source control:

export TELNYX_API_KEY="KEY..."
export TELNYX_FROM_NUMBER="+15551234567"
export TELNYX_WEBHOOK_SECRET="replace-with-your-webhook-secret"

The sample assumes E.164 numbers. Keep the leading + and country code in the database; mismatched formatting is one of the fastest ways to split a single customer conversation into multiple threads.

Step-by-step

1. Create a deterministic thread key.
   Do not use a provider message ID as the conversation ID: each SMS has its own ID. Instead, build the key from the two participants in fixed roles: local_number:customer_number. For an outbound message, the local number is from; for an inbound reply, it is the recipient in to. That role-based order prevents +15551234567:+15557654321 and its reverse from becoming separate threads.

2. Create a message table with provider-level deduplication.
   Delivery systems can retry a webhook, and your endpoint must be safe to call more than once. A UNIQUE constraint on provider_message_id means a replay cannot create a duplicate reply. The following single-file app initializes SQLite and persists both inbound and outbound messages:

# app.py
import hashlib
import os
import sqlite3
from datetime import datetime, timezone

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

app = Flask(__name__)
DB_PATH = "sms.db"
API_URL = os.environ["TELNYX_MESSAGES_URL"]  # Set this to the Telnyx messages endpoint.


def db():
    connection = sqlite3.connect(DB_PATH)
    connection.row_factory = sqlite3.Row
    return connection


def init_db():
    with db() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS messages (
                id INTEGER PRIMARY KEY,
                provider_message_id TEXT NOT NULL UNIQUE,
                thread_id TEXT NOT NULL,
                direction TEXT NOT NULL CHECK(direction IN ('outbound', 'inbound')),
                local_number TEXT NOT NULL,
                customer_number TEXT NOT NULL,
                body TEXT,
                occurred_at TEXT NOT NULL
            )
        """)


def thread_id(local_number: str, customer_number: str) -> str:
    # A hash avoids exposing phone numbers if this ID is later used in a URL.
    pair = f"{local_number.strip()}:{customer_number.strip()}"
    return hashlib.sha256(pair.encode()).hexdigest()


def save_message(provider_id, direction, local, customer, body, occurred_at):
    with db() as conn:
        conn.execute("""
            INSERT OR IGNORE INTO messages
            (provider_message_id, thread_id, direction, local_number,
             customer_number, body, occurred_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        """, (provider_id, thread_id(local, customer), direction,
              local, customer, body, occurred_at))


init_db()

3. Send the first message and write the outbound record.
   The send operation is POST /v2/messages. Pass from, to, and text; save the returned message ID immediately. use_profile_webhooks asks the platform to use the webhook settings from the selected Messaging Profile, keeping the callback destination out of application code.

def send_sms(customer_number: str, text: str):
    local_number = os.environ["TELNYX_FROM_NUMBER"]
    response = requests.post(
        API_URL,
        headers={
            "Authorization": f"Bearer {os.environ['TELNYX_API_KEY']}",
            "Content-Type": "application/json",
        },
        json={
            "from": local_number,
            "to": customer_number,
            "text": text,
            "use_profile_webhooks": True,
        },
        timeout=15,
    )
    response.raise_for_status()
    message = response.json()["data"]
    save_message(
        provider_id=message["id"],
        direction="outbound",
        local=local_number,
        customer=customer_number,
        body=text,
        occurred_at=datetime.now(timezone.utc).isoformat(),
    )
    return message["id"]


if __name__ == "__main__":
    # Send once for a controlled test; remove this line in a web deployment.
    # print(send_sms("+15557654321", "Reply YES to confirm your appointment."))
    app.run(host="0.0.0.0", port=8000, debug=True)

Use the Telnyx API reference to review optional parameters and the current response schema. In production, generate an application idempotency key for each user action as well, so a client retry does not send the opening message twice.

4. Receive message.received and attach it to the same key.
   Configure your Messaging Profile’s inbound webhook URL with your public endpoint followed by /webhooks/telnyx. The handler below reads the common data.payload message fields. It takes the inbound sender as the customer and the first to recipient as the local number, then saves the reply with the same role-based thread key.

@app.post("/webhooks/telnyx")
def telnyx_webhook():
    # Verify the Telnyx webhook signature before trusting this JSON in production.
    event = request.get_json(silent=True)
    if not event or event.get("data", {}).get("event_type") != "message.received":
        return ("", 204)

    data = event["data"]
    payload = data["payload"]
    customer_number = payload["from"]["phone_number"]
    local_number = payload["to"][0]["phone_number"]
    provider_id = payload["id"]
    body = payload.get("text") or ""
    occurred_at = data.get("occurred_at") or datetime.now(timezone.utc).isoformat()

    save_message(provider_id, "inbound", local_number, customer_number,
                 body, occurred_at)
    return ("", 204)

Return a fast 2xx response only after the durable insert succeeds. Move slow work—AI responses, CRM synchronization, notifications, or sending a reply—to a background worker. The webhook handler stays available during traffic spikes, while the worker can query messages by thread_id in chronological order.

5. Read a complete conversation.
   Your support UI or automation can fetch the thread without guessing which message led to which reply:

def get_thread(local_number: str, customer_number: str):
    with db() as conn:
        return conn.execute("""
            SELECT direction, body, occurred_at, provider_message_id
            FROM messages
            WHERE thread_id = ?
            ORDER BY occurred_at, id
        """, (thread_id(local_number, customer_number),)).fetchall()

Start the server, expose port 8000 through your HTTPS endpoint, point the Messaging Profile at the resulting URL, and send a test SMS. Reply from the recipient phone. You should see one outbound and one inbound row with an identical thread_id.

Common pitfalls

• Threading on message ID: message IDs identify individual events, not the conversation. Use the local/customer pair or a deliberate database conversation record.
• Skipping webhook verification: anyone can post JSON to a public route. Verify Telnyx’s signature using the platform’s current webhook security instructions before processing or replying.
• Assuming a single recipient shape: inbound to is a list. Select the number your application owns and support multiple local numbers if your account uses them.
• Ignoring retries and ordering: make inserts idempotent, store the provider timestamp, and do not assume webhooks arrive in the order messages were sent.
• Auto-replying to opt-outs: process STOP/HELP keywords and consent state before a bot or worker sends another message. Compliance is a product requirement, not a cleanup task.

Frequently Asked Questions

Do I need a separate conversation API to make SMS two-way?
No. Send through the messaging endpoint, receive inbound webhooks, and maintain the conversation relationship in your own data store. This gives your CRM, support workflow, and automation the same source of truth.

What should my thread ID contain?
At minimum, use the local E.164 number and the customer E.164 number in a fixed order. If you need separate threads for separate cases with the same customer, create a database conversation ID and associate it with the initial outbound message; still retain the phone-pair key for routing unsolicited replies.

How do I prevent duplicate replies?
Make provider_message_id unique, acknowledge duplicate webhook deliveries safely, and use an idempotency key for outbound user actions. A unique database constraint is more dependable than an in-memory “already seen” flag.

Can I use SQLite in production?
SQLite is excellent for a local proof of concept and low-contention workloads. For concurrent webhooks and workers, use a managed relational database with the same unique constraint and an indexed thread_id column.

Conclusion

A strong two-way SMS implementation is simple by design: send the message, persist its provider ID, accept verified inbound webhooks, and derive both directions’ thread ID from the same local/customer pair. Build that ledger first, then layer on agents, CRM updates, delivery monitoring, and analytics. Create your Telnyx messaging setup, configure the webhook, and run the test flow now—once the two rows share a thread ID, you have a dependable foundation for every SMS conversation that follows.