DEV Community

Harpreet Singh Seehra
Harpreet Singh Seehra

Posted on

Build an AI Agent That Emails, Texts, and Calls — One API, Shared Context

This tutorial builds an AI agent that communicates with customers across email, SMS, and voice — choosing the right channel automatically and maintaining shared context across all of them.

The agent uses Claude API for tool-calling (the AI brain), Telnyx APIs for email/SMS/voice delivery (the communication layer), and SQLite for persistent cross-channel context.

I work with Telnyx. The code is available here: https://github.com/team-telnyx/telnyx-code-examples/tree/main/omnichannel-ai-agent-python

What you will build

An AI agent that:

  1. Receives a customer scenario (e.g., billing dispute)
  2. Retrieves the full cross-channel conversation history from SQLite
  3. Uses Claude tool-calling to decide which channel to use next
  4. Sends the message via Telnyx (Email, SMS, or Voice)
  5. Stores the interaction and loops until the issue is resolved

Each message references previous interactions across channels — the SMS mentions the email, the voice call mentions both.

Prerequisites

  • Python 3.11+
  • A Telnyx account with API key, phone number, and verified email sender (or use the demo — no credentials needed)
  • An Anthropic API key for Claude (or use the demo)

Project structure 

omnichannel-ai-agent-python/
├── app.py                 # Main Flask app (~300 lines)
├── demo/
│   └── demo_server.py     # Full demo without credentials
├── smoke_test.py          # 11 smoke tests
├── requirements.txt
├── .env.example
└── README.md
Enter fullscreen mode Exit fullscreen mode

Step 1: Install dependencies 

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/omnichannel-ai-agent-python

pip install -r requirements.txt
Enter fullscreen mode Exit fullscreen mode

Dependencies: flask, telnyx, anthropic, python-dotenv, requests.

Step 2: Understand the architecture 

                    Claude API (AI Brain)
                         │
                    Tool-calling decides
                    which channel to use
                         │
                    ┌────┴────┐
                    │ Flask   │
                    │ app.py  │
                    │         │
              ┌─────┤ Context ├─────┐
              │     │ (SQLite)│     │
              │     └────┬────┘     │
              │          │          │
         Email API   Messaging  Call Control
        POST /v2/    POST /v2/   POST /v2/
     email_messages  messages     calls
Enter fullscreen mode Exit fullscreen mode

The AI brain (Claude) receives four tool definitions and the conversation history. It decides which tool to call, and the Flask app executes it via the corresponding Telnyx API.

Step 3: Define the tools

Claude uses tool-calling to decide channel routing. Each tool maps to a Telnyx API:

TOOLS = [
    {
        "name": "send_email",
        "description": (
            "Send a detailed email to the customer. Use for formal "
            "acknowledgments, detailed explanations, or when a written "
            "record is needed."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "subject": {"type": "string", "description": "Email subject line"},
                "body": {"type": "string", "description": "Email body text"},
            },
            "required": ["subject", "body"],
        },
    },
    {
        "name": "send_sms",
        "description": (
            "Send a short SMS text message to the customer. Use for quick "
            "status updates, confirmations, or time-sensitive notifications."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "text": {"type": "string", "description": "SMS message text"},
            },
            "required": ["text"],
        },
    },
    {
        "name": "make_call",
        "description": (
            "Call the customer and speak a message. Use for complex "
            "resolution, urgent matters, or when a personal touch is needed."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "speak_text": {"type": "string", "description": "Text to speak"},
            },
            "required": ["speak_text"],
        },
    },
    {
        "name": "resolve_issue",
        "description": "Mark the customer issue as resolved.",
        "input_schema": {
            "type": "object",
            "properties": {
                "summary": {"type": "string", "description": "Resolution summary"},
            },
            "required": ["summary"],
        },
    },
]
Enter fullscreen mode Exit fullscreen mode

The descriptions are the routing rules. Claude reads "formal acknowledgments" → email. "Quick status updates" → SMS. "Complex resolution" → voice call.

Step 4: Build the context store

SQLite stores every interaction across all channels:

def init_db():
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()
    cur.execute("""
        CREATE TABLE IF NOT EXISTS conversations (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            customer_id TEXT    NOT NULL,
            channel     TEXT    NOT NULL,
            role        TEXT    NOT NULL,
            content     TEXT    NOT NULL,
            timestamp   TEXT    NOT NULL
        )
    """)
    cur.execute(
        "CREATE INDEX IF NOT EXISTS idx_conv_customer ON conversations (customer_id)"
    )
    conn.commit()
    conn.close()
Enter fullscreen mode Exit fullscreen mode

Before each agent run, the full history is retrieved:

def get_conversation_history(customer_id):
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    cur.execute(
        "SELECT channel, role, content, timestamp FROM conversations "
        "WHERE customer_id = ? ORDER BY timestamp",
        (customer_id,),
    )
    rows = cur.fetchall()
    conn.close()
    return [dict(r) for r in rows]
Enter fullscreen mode Exit fullscreen mode

Step 5: Implement the channel functions

Each channel function hits a Telnyx API and stores the message:

Email (Telnyx Email API) 

def send_email(to_email, subject, body):
    resp = requests.post(
        "https://api.telnyx.com/v2/email_messages",
        headers={
            "Authorization": f"Bearer {TELNYX_API_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "from": {"email": TELNYX_EMAIL_FROM},
            "to": [{"email": to_email}],
            "subject": subject,
            "body": body,
        },
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json()
Enter fullscreen mode Exit fullscreen mode

SMS (Telnyx Messaging API) 

def send_sms(to_number, text):
    payload = {"from": TELNYX_FROM_NUMBER, "to": to_number, "text": text}
    if MESSAGING_PROFILE_ID:
        payload["messaging_profile_id"] = MESSAGING_PROFILE_ID
    resp = requests.post(
        "https://api.telnyx.com/v2/messages",
        headers={
            "Authorization": f"Bearer {TELNYX_API_KEY}",
            "Content-Type": "application/json",
        },
        json=payload,
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()
Enter fullscreen mode Exit fullscreen mode

Voice (Telnyx Call Control API) 

def make_call(to_number, speak_text):
    resp = requests.post(
        "https://api.telnyx.com/v2/calls",
        headers={
            "Authorization": f"Bearer {TELNYX_API_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "connection_id": CONNECTION_ID,
            "to": to_number,
            "from": TELNYX_FROM_NUMBER,
        },
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()
Enter fullscreen mode Exit fullscreen mode

All three channels use the same API key and base URL. One platform, one authentication.

Step 6: Build the agentic loop

The agent runs a standard Claude tool-calling loop:

def run_agent(customer, scenario):
    history = get_conversation_history(customer["id"])
    messages = [{"role": "user", "content": f"Customer: {customer['name']}...\nIssue: {scenario}"}]

    actions = []

    while True:
        response = claude_client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            system=SYSTEM_PROMPT,
            tools=TOOLS,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            break

        tool_use_blocks = [b for b in response.content if b.type == "tool_use"]
        messages.append({"role": "assistant", "content": response.content})

        tool_results = []
        for tool in tool_use_blocks:
            result = execute_tool(tool.name, tool.input, customer)
            actions.append({"type": "tool_call", "tool": tool.name, "result": result})
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": tool.id,
                "content": result,
            })

        messages.append({"role": "user", "content": tool_results})

    return actions
Enter fullscreen mode Exit fullscreen mode

Step 7: Run the demo

No credentials needed:

python demo/demo_server.py
Enter fullscreen mode Exit fullscreen mode

The demo:

  1. Starts the Flask server on localhost:5555
  2. Automatically triggers the agent with a billing dispute scenario
  3. Prints each channel firing in sequence: email → SMS → voice
  4. Stores everything in SQLite

Check the conversation history:

curl http://localhost:5555/conversations | python -m json.tool
Enter fullscreen mode Exit fullscreen mode

Step 8: Run with real credentials 

cp .env.example .env
# Edit .env with your Telnyx and Claude API keys
python app.py
Enter fullscreen mode Exit fullscreen mode

Trigger the agent:

curl -X POST http://localhost:5000/agent/run \
  -H "Content-Type: application/json" \
  -d '{
    "customer": {
      "id": "cust_001",
      "name": "Sarah Chen",
      "email": "sarah@example.com",
      "phone": "+15551234567"
    },
    "scenario": "Customer is disputing a $147.50 charge."
  }'
Enter fullscreen mode Exit fullscreen mode

Step 9: Run the tests 

python -m pytest smoke_test.py -v
Enter fullscreen mode Exit fullscreen mode

11 tests covering module imports, route registration, database operations, tool definitions, and API endpoints. All pass.

What makes this omnichannel

The difference between multichannel and omnichannel is shared context. This agent:

  • Stores every interaction in one SQLite table regardless of channel
  • Passes the full cross-channel history to Claude before every decision
  • References previous channel interactions in each new message
  • Uses AI judgment (not hardcoded rules) for channel routing

The SMS mentions the email. The voice call mentions both. One conversation. Three channels.

Resources

Related Examples

Top comments (0)