personal

Pharmacy Voice Agent

Twilio telephony with Deepgram speech-to-speech

Overview

This project is a pharmacy voice agent built around Twilio telephony, a local Python WebSocket bridge, and Deepgram’s Voice Agent API for speech-to-speech. A caller dials a Twilio number; Media Streams tunnel through ngrok into an asyncio server on localhost:5000. The server forwards inbound mulaw chunks to Deepgram (nova-3 STT, gpt-4o-mini reasoning, aura-2-thalia-en TTS) and streams synthesized audio back to the phone. Mid-conversation, the agent can call local Python tools against in-memory drug and order stores, and barge-in is handled by clearing Twilio playback as soon as the caller starts speaking.

Architecture

flowchart TB
    Caller["Caller phone"]

    subgraph Cloud["Cloud services"]
        Twilio["Twilio phone number + Media Streams"]
        subgraph DG["Deepgram Voice Agent API"]
            STT["Listen / STT Deepgram nova-3"]
            LLM["Think / LLM OpenAI gpt-4o-mini"]
            TTS["Speak / TTS Deepgram aura-2-thalia-en"]
            STT --> LLM --> TTS
        end
    end

    subgraph Dev["Local development"]
        Ngrok["ngrok HTTPS / WSS tunnel"]
        UV["uv + PyCharm Python 3.13 env"]

        subgraph App["voice-agent localhost:5000"]
            Main["main.py asyncio + websockets"]

            subgraph Bridge["Per-call audio bridge twilio_handler"]
                TR["twilio_receiver start / media / stop buffer 20x160 mulaw"]
                SS["sts_sender forward chunks to Deepgram"]
                SR["sts_receiver TTS audio + JSON events"]
                BI["handle_barge_in UserStartedSpeaking to Twilio clear"]
                FC["handle_function_call_request FunctionCallRequest to local Python"]
            end

            Config["config.json mulaw 8 kHz, prompt, tool schemas, greeting"]
            Env[".env DEEPGRAM_API_KEY"]
            Pharmacy["pharmacy_functions.py"]

            subgraph Store["In-memory stores"]
                DrugDB["DRUG_DB"]
                OrdersDB["ORDERS_DB"]
            end
        end
    end

    Caller -->|"PSTN call"| Twilio
    Twilio -->|"Media Stream WebSocket"| Ngrok
    Ngrok -->|"WSS to ws://localhost:5000"| Main
    Main --> TR
    TR -->|"inbound mulaw chunks"| SS
    SS -->|"binary audio"| STT
    TTS -->|"binary mulaw TTS"| SR
    SR -->|"event media base64 mulaw"| Twilio
    Twilio -->|"playback"| Caller

    LLM -->|"FunctionCallRequest JSON"| SR
    SR --> FC
    FC --> Pharmacy
    Pharmacy --> DrugDB
    Pharmacy --> OrdersDB
    FC -->|"FunctionCallResponse"| LLM

    LLM -->|"UserStartedSpeaking"| SR
    SR --> BI
    BI -->|"event clear"| Twilio

    Config -->|"Settings on connect"| DG
    Env -->|"auth subprotocol token"| DG
    UV -.-> App

Call workflow

sequenceDiagram
    autonumber
    participant Phone as Caller
    participant Twilio as Twilio
    participant Ngrok as ngrok
    participant Server as main.py :5000
    participant DG as Deepgram Agent
    participant Tools as pharmacy_functions.py

    Phone->>Twilio: Dial provisioned number
    Twilio->>Ngrok: Open Media Stream WebSocket
    Ngrok->>Server: twilio_handler(twilio_ws)

    Server->>DG: Connect wss://agent.deepgram.com
    Server->>DG: Send config.json Settings

    par Concurrent asyncio tasks
        Twilio->>Server: event start streamSid
        Server->>Server: streamsid_queue
    and
        loop Inbound speech
            Phone->>Twilio: Voice (mulaw)
            Twilio->>Server: event media inbound payload
            Server->>Server: Decode base64, buffer to 20x160
            Server->>DG: Binary mulaw chunk
        end
    and
        DG->>Server: Greeting / TTS mulaw
        Server->>Twilio: event media base64 payload
        Twilio->>Phone: Play audio
    end

    opt Interruption
        DG->>Server: UserStartedSpeaking
        Server->>Twilio: event clear streamSid
        Note over Twilio,Phone: Stop current playback immediately
    end

    opt Tool / function calling
        DG->>Server: FunctionCallRequest
        Server->>Tools: get_drug_info / place_order / lookup_order
        Tools-->>Server: JSON result
        Server->>DG: FunctionCallResponse
        DG->>Server: Spoken reply as mulaw TTS
        Server->>Twilio: event media
        Twilio->>Phone: Play answer
    end

    Phone->>Twilio: Hang up
    Twilio->>Server: event stop
    Server->>Server: Close Deepgram + Twilio sockets

File-to-role map

flowchart LR
    subgraph Repo["Repository"]
        M["main.py"]
        P["pharmacy_functions.py"]
        C["config.json"]
        E[".env"]
        PY["pyproject.toml / uv.lock"]
    end

    subgraph Roles["Runtime roles"]
        R1["WebSocket server + audio bridge"]
        R2["Pharmacy tools + in-memory DB"]
        R3["Agent personality, STT/LLM/TTS, tool schemas"]
        R4["Deepgram API auth"]
        R5["Deps: websockets, python-dotenv"]
    end

    M --> R1
    P --> R2
    C --> R3
    E --> R4
    PY --> R5

What I worked on

Built the asyncio WebSocket server that accepts Twilio Media Streams and bridges audio to Deepgram.
Implemented per-call audio tasks: inbound buffering (20×160 mulaw), forwarding to Deepgram, and streaming TTS back to Twilio.
Wired Deepgram Settings on connect from config.json: mulaw 8 kHz, prompt, tool schemas, and greeting.
Added local pharmacy tools (get_drug_info, place_order, lookup_order) over in-memory DRUG_DB and ORDERS_DB.
Handled interruptions by mapping UserStartedSpeaking to a Twilio clear event so playback stops immediately.
Used ngrok to expose the local WSS endpoint to Twilio during development with uv and Python 3.13.

Results

End-to-end speech-to-speech over a live phone call with low-latency audio bridging
Function calling mid-conversation for get_drug_info, place_order, and lookup_order
Barge-in handling: UserStartedSpeaking immediately clears Twilio playback
Configurable agent personality, audio format, tool schemas, and greeting via config.json