Documentation
Free API Key

Twilio and Deepgram Voice Agent

Learn how to use Twilio with Deepgram Voice Agent API.

Suggest Edits

Deepgram Voice Agent can integrate with the Twilio streaming API to enable dynamic interactions between callers and voice agents or bots. This guide will walk you through how to setup a Twilio phone number that can interact with Deepgram's Voice Agent API, allowing callers to engage with a voice agent in real-time.

Before you Begin

📘

Before you can use Deepgram, you'll need to create a Deepgram account. Signup is free and includes $200 in free credit and access to all of Deepgram's features!

📘

Before you start, you'll need to follow the steps in the Make Your First API Request guide to obtain a Deepgram API key, and configure your environment if you are choosing to use a Deepgram SDK.

Prerequisites

For the complete code used in this guide, please check out this repository.

You will need:

  • A free Twilio account with a Twilio phone number.
  • ngrok to let Twilio access a local server OR your own hosted server.
  • Understanding of Python and using Python virtual environments.

TwiML Bin Setup

First, you will need to set up a TwiML Bin. You can refer to the docs on how to do that in the Twilio Console.

<?xml version="1.0" encoding="UTF-8"?>

<Response>
    <Say language="en">"This call may be monitored or recorded."</Say>
    <Connect>
        <Stream url="wss://a127-75-172-116-97.ngrok-free.app/twilio" />
    </Connect>
</Response>
  • You should replace the url with wherever you decide to deploy the server we are about to create and ensure/twilio is at the end of the url.
  • In the TwiML Bin example above, ngrok is used to expose the server running locally.
  • Be sure to use the ngrok URL provided as your WSS endpoint: In your Twilio Bin configuration you will need to replace http:// with wss://.

Using ngrok

ngrok is recommended for quick development and testing but shouldn't be used for production instances. To use ngrok see their documentation.

Be sure to set the port correctly to 5000 to align with the server code provided by running this command when you start the ngrok server.

ngrok http 5000

📘

If you restart your ngrok server, your URL will change, which will require you to update your TwiML Bin.

Connecting a Twilio phone number

Your TwiML Bin must then be connected to one of your Twilio phone numbers so that it gets executed whenever someone calls that number. If you need to set up a new phone number and connect it to your TwiML Bin, refer to the Twilio Docs.

📘

In your TwiML Bin The <Connect> verb is required for bi-directional communication, i.e. in order to send audio from the Deepgram Agent to Twilio, you must use this verb.

Building the Server

Copy the server.pycode from the repository as we will use this in the steps below and save this code locally as with a file name of server.py.

At this point you'll want to start up a virtual environment for Python. Please refer to documentation for how to do that based on your personal Python preferences.

Depending on your situation you may also need to install specific packages used in this code. You can install the packages you need manually or use the requirements.txt file.

pip install -r requirements.txt

You can set your Deepgram API key in the sts_connect function to run the server:

subprotocols=["token", "YOUR_DEEPGRAM_API_KEY"]

If your TwiML Bin is setup correctly, you can now navigate to the correct file location in your terminal and run the server with the following command:

python server.py

OR

python3 server.py

Make a test call

You can now start making calls to the phone number your TwiML Bin is using. Without any further code modifications, you should hear Deepgram Aura say simply: "Hello, how are you today?"

Code Tour

Let's dive into the code used in the server.py file.

First, we have some import statements:

import asyncio
import base64
import json
import sys
import websockets
import ssl
  • We are using asyncio and websockets to build an asynchronous websocket server.
  • We will use base64 to handle encoding audio from Aura to pass data to Twilio.
  • We will use json to deal with parsing text messages from Twilio .
  • We will use sys to provides access to some variables and functions used or maintained by the Python interpreter.
  • We will use ssl(optional) to create secure encrypted connections between client and server.

The next block of code sts_connectdefines a function that establishes a WebSocket connection to Deepgram's agent service.

def sts_connect():
    sts_ws = websockets.connect(
    "wss://agent.deepgram.com/agent", subprotocols=["token", "YOUR_DEEPGRAM_API_KEY"]
    )
    return sts_ws

Let's break it down:

Connection Setup:

  • Creates a secure WebSocket connection to wss://agent.deepgram.com/agent
  • Includes authentication via subprotocols

Authentication Method:

  • Uses Deepgram's token-based authentication
  • Requires replacing "YOUR_DEEPGRAM_API_KEY" with an actual Deepgram API key

The next block of code, twilio_handler does several things:

In this first code block, we set up an asynchronous function to handle WebSocket messages from Twilio. We define additional asynchronous functions to manage messages received from Twilio, messages sent to Deepgram, and responses from Deepgram. To facilitate data sharing between tasks, we use two queues: one for audio from Twilio and another for Twilio's stream SID (a unique identifier). 

async def twilio_handler(twilio_ws):
    audio_queue = asyncio.Queue()
    streamsid_queue = asyncio.Queue()

Also included in twilio_handler is the Setting Configuration for our Agent. The most important thing to note here is the audio format we are using 8000 Hz, raw, un-containerized mulaw. This is the format Twilio will be sending, and the format we will need to send back to Twilio including some base64 encoding/decoding.

📘

To learn more about supported media inputs and outputs for the Voice Agent review the documentation.

async with sts_connect() as sts_ws:
        config_message = {
            "type": "SettingsConfiguration",
            "audio": {
                "input": {
                    "encoding": "mulaw",
                    "sample_rate": 8000,
                },
                "output": {
                    "encoding": "mulaw",
                    "sample_rate": 8000,
                    "container": "none",
                },
            },
            "agent": {
                "listen": {"model": "nova-2"},
                "think": {
                    "provider": {
                        "type": "anthropic",
                    },
                    "model": "claude-3-haiku-20240307",
                    "instructions": "You are a helpful car seller.",
                },
                "speak": {"model": "aura-asteria-en"},
            },
        }
    
    await sts_ws.send(json.dumps(config_message))

The next block of code is sts_sender. This function waits for audio from Twilio (via the audio queue) and continuously reads audio chunks from the queue forwarding the chunks to the Deepgram Voice Agent API.

async def sts_sender(sts_ws):
            print("sts_sender started")
            while True:
                chunk = await audio_queue.get()
                await sts_ws.send(chunk)

Next is sts_receiver which waits until it has received a stream SID from Twilio, and then loops over messages
received from the Deepgram Voice Agent API. If we receive a text message, we check to ensure that the user has started speaking. If they have, we treat this as barge-in and have Twilio clear the agent audio on the call using the stream SID.

Other audio messages should be binary messages containing the text-to-speech (TTS) output of the Deepgram Voice Agen API. We pack all of this up into valid Twilio messages (using the stream SID again), and send
them to Twilio to be played back on the phone for the caller to here.

📘

For more information about streaming audio to Twilio, see the following Documentation.

async def sts_receiver(sts_ws):
            print("sts_receiver started")
            # we will wait until the twilio ws connection figures out the streamsid
            streamsid = await streamsid_queue.get()
            # for each sts result received, forward it on to the call
            async for message in sts_ws:
                if type(message) is str:
                    print(message)
                    # handle barge-in
                    decoded = json.loads(message)
                    if decoded['type'] == 'UserStartedSpeaking':
                        clear_message = {
                            "event": "clear",
                            "streamSid": streamsid
                        }
                        await twilio_ws.send(json.dumps(clear_message))

                    continue

                print(type(message))
                raw_mulaw = message

                # construct a Twilio media message with the raw mulaw (see https://www.twilio.com/docs/voice/twiml/stream#websocket-messages---to-twilio)
                media_message = {
                    "event": "media",
                    "streamSid": streamsid,
                    "media": {"payload": base64.b64encode(raw_mulaw).decode("ascii")},
                }

                # send the TTS audio to the attached phonecall
                await twilio_ws.send(json.dumps(media_message))

The next block of code is twilio_recieverand loops over messages Twilio is sending our server. If we receive a "start" message, we can extract the stream SID, and send it to our other async task which needs it. If we receive a "media" message, we decode the audio from it, append it to a running buffer, and send it to the async task which
forwards it to Deepgram when it's of a reasonable size.

Be aware there can be throughput issues when sending lots of tiny chunks, so that's why we are doing this buffering approach.

  async def twilio_receiver(twilio_ws):
            print("twilio_receiver started")
            # twilio sends audio data as 160 byte messages containing 20ms of audio each
            # we will buffer 20 twilio messages corresponding to 0.4 seconds of audio to improve throughput performance
            BUFFER_SIZE = 20 * 160

            inbuffer = bytearray(b"")
            async for message in twilio_ws:
                try:
                    data = json.loads(message)
                    if data["event"] == "start":
                        print("got our streamsid")
                        start = data["start"]
                        streamsid = start["streamSid"]
                        streamsid_queue.put_nowait(streamsid)
                    if data["event"] == "connected":
                        continue
                    if data["event"] == "media":
                        media = data["media"]
                        chunk = base64.b64decode(media["payload"])
                        if media["track"] == "inbound":
                            inbuffer.extend(chunk)
                    if data["event"] == "stop":
                        break

                    # check if our buffer is ready to send to our audio_queue (and, thus, then to sts)
                    while len(inbuffer) >= BUFFER_SIZE:
                        chunk = inbuffer[:BUFFER_SIZE]
                        audio_queue.put_nowait(chunk)
                        inbuffer = inbuffer[BUFFER_SIZE:]
                except:
                    break

The next block of code runs the asynchronous tasks defined in twilio_reciever.

await asyncio.wait(
            [
                asyncio.ensure_future(sts_sender(sts_ws)),
                asyncio.ensure_future(sts_receiver(sts_ws)),
                asyncio.ensure_future(twilio_receiver(twilio_ws)),
            ]
        )

        await twilio_ws.close()

Finally the last block of code sets up and runs the server, making sure all incoming websocket connections get handled by twilio_handler.

async def router(websocket, path):
    print(f"Incoming connection on path: {path}")
    if path == "/twilio":
        print("Starting Twilio handler")
        await twilio_handler(websocket)

def main():
    # use this if using ssl
    # ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    # ssl_context.load_cert_chain('cert.pem', 'key.pem')
    # server = websockets.serve(router, '0.0.0.0', 443, ssl=ssl_context)

    # use this if not using ssl
    server = websockets.serve(router, "localhost", 5000)
    print("Server starting on ws://localhost:5000")

    asyncio.get_event_loop().run_until_complete(server)
    asyncio.get_event_loop().run_forever()


if __name__ == "__main__":
    sys.exit(main() or 0)

Updated 1 day ago

What’s Next