Neuphonic Agents and Twilio Integration Guide
This demo illustrates how to integrate Neuphonic's agent features with Twilio to create an interactive and intelligent phone voice agent capable of handling both inbound and outbound phone calls.
Prerequisites
Before getting started, ensure you have:
- A Neuphonic account and API Key
- A Twilio account with a Twilio phone number
- Python 3.9+
- A free ngrok account with the ngrok CLI installed and authenticated
Implementation
In this section, you will learn how to set up a server that manages both incoming and outgoing calls. The server will interact with Twilio to handle incoming audio and forward responses back to the user's phone using Twilio's services.
- Python
- JavaScript SDK
Looking for a complete example? Check out this demo on GitHub which you can run instantly.
Initialize the Project
Create a folder to hold the python project.
mkdir twilio-agent
cd twilio-agentCreate a Virtual Environment
Create a virtual environment using your desired environment manager.
python -m venv .venv
source .venv/bin/activateInstall Dependencies
Next, install the necessary dependencies.
pip install "pyneuphonic~=1.8.0" "twilio~=9.3" "fastapi[standard]~=0.114" "python-dotenv~=1.0" "uvicorn[standard]~=0.30"
Configure Your Environment
Create and configure your
.env
file with the following secretsTWILIO_AUTH_TOKEN=
TWILIO_ACCOUNT_SID=
# The base url of this server, e.g., if using ngrok it would similar to 123d-45-678-912-3.ngrok-free.app
SERVER_BASE_URL=
FROM_NUMBER=
TO_NUMBER=
NEUPHONIC_API_KEY=NEUPHONIC_API_KEY
: Found on the Neuphonic portalTWILIO_AUTH_TOKEN
andTWILIO_ACCOUNT_SID
: Available in your Twilio portalFROM_NUMBER
: Your Twilio-purchased phone numberTO_NUMBER
: Your personal phone number for receiving test calls
Leave
SERVER_BASE_URL
blank for now - we'll fill this in later.Create the Project Files
Create an
app.py
andmake_outbound_call.py
file with the following codefrom dotenv import load_dotenv
from starlette.websockets import WebSocketState
import logging
import json
import base64
from pyneuphonic import Neuphonic, WebsocketEvents, AgentConfig
from pyneuphonic.models import APIResponse, TTSResponse
import os
from twilio.twiml.voice_response import VoiceResponse, Connect
from fastapi.responses import HTMLResponse
from fastapi import (
FastAPI,
APIRouter,
WebSocket,
WebSocketDisconnect,
Request
)
load_dotenv(override=True)
app = FastAPI()
router = APIRouter()
@app.get('/ping')
def ping():
return {'message': 'pong'}
@app.api_route("/twilio/inbound_call", methods=["GET", "POST"])
async def handle_incoming_call(request: Request):
"""Handle incoming call and return TwiML response."""
response = VoiceResponse()
connect = Connect()
connect.stream(url=f'wss://{os.getenv("SERVER_BASE_URL")}/twilio/agent')
response.append(connect)
return HTMLResponse(content=str(response), media_type="application/xml")
@app.websocket('/twilio/agent')
async def agent_websocket(
websocket: WebSocket,
):
"""Handles the WebSocket connection for a Twilio agent, allowing real-time voice conversation."""
await websocket.accept()
stream_sid = None
client = Neuphonic(api_key=os.getenv('NEUPHONIC_API_KEY'))
neuphonic_agent_websocket = client.agents.AsyncWebsocketClient()
async def on_message(message: APIResponse[TTSResponse]):
"""Handles messages that are returned from the Neuphonic server to the websocket client."""
if stream_sid is not None and message.data.type == 'audio_response':
# Forward audio to the users's phone
await websocket.send_json(
{
'event': 'media',
'streamSid': stream_sid,
'media': {
'payload': base64.b64encode(message.data.audio).decode('utf-8')
},
}
)
elif message.data.type == 'user_transcript':
logging.info(f'user_transcript: {message.data.text}')
elif message.data.type == 'llm_response':
logging.info(f'llm_response: {message.data.text}')
elif message.data.type == 'stop_audio_response':
# If the user interrupts then stop playing any currently queued audio
if stream_sid is not None:
await websocket.send_json(
{
'event': 'clear',
'streamSid': stream_sid,
}
)
neuphonic_agent_websocket.on(WebsocketEvents.MESSAGE, on_message)
await neuphonic_agent_websocket.open(
agent_config=AgentConfig(
incoming_sampling_rate=8000,
return_sampling_rate=8000,
incoming_encoding='pcm_mulaw',
return_encoding='pcm_mulaw',
voice_id='<VOICE_ID>',
)
)
try:
while websocket.client_state == WebSocketState.CONNECTED:
message = await websocket.receive()
if message is None or message['type'] == 'websocket.disconnect':
continue
data = json.loads(message['text'])
if data['event'] == 'connected':
logging.info(f'Connected Message received: {message}')
if data['event'] == 'start':
stream_sid = data['start']['streamSid']
if data['event'] == 'media':
# Send incoming audio to the Neuphonic agent
await neuphonic_agent_websocket.send(
{'audio': data['media']['payload']}
)
if data['event'] == 'closed':
logging.info(f'Closed Message received: {message}')
break
except WebSocketDisconnect as e:
logging.error(f'WebSocketDisconnect: {e}')
except Exception as e:
logging.error(f'Error occured: {e}')
finally:
await neuphonic_agent_websocket.close()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)import os
from twilio.rest import Client
from dotenv import load_dotenv
load_dotenv(override=True)
def make_outbound_call():
# Initialize the Twilio client with account credentials
client = Client(os.getenv('TWILIO_ACCOUNT_SID'), os.getenv('TWILIO_AUTH_TOKEN'))
# Construct the TwiML message to connect the call to the WebSocket endpoint
twiml_message = f'<?xml version="1.0" encoding="UTF-8"?><Response><Connect><Stream url="wss://{os.getenv("SERVER_BASE_URL")}/twilio/agent" /></Connect></Response>'
# Create the call with the specified parameters
call = client.calls.create(
twiml=twiml_message,
to=os.getenv('TO_NUMBER'),
from_=os.getenv('FROM_NUMBER'),
record=True,
)
# Print the call SID
print(call.sid)
if __name__ == '__main__':
# Execute the make_outbound_call function if the script is run directly
make_outbound_call()Your project should now look like this
twilio-agent
├── .venv/
├── .env
├── app.py
└── make_outbound_call.pyStart ngrok
Start ngrok using the following command.
ngrok http http://localhost:8000
Copy the displayed URL (excluding the
https://
prefix) and add it to your.env
file asSERVER_BASE_URL
. It will look something like123d-45-678-912-3.ngrok-free.app
.noteNote that this is an ephemeral address, every time you restart the ngrok service using the above command, you will be given a new URL.
Start the Server
Start the FastAPI server using the following command:
python app.py
Your server is now running and ready to accept incoming calls and make outgoing calls.
Initiate an Outbound Call
Initiate a call from your Twilio number
FROM_NUMBER
to your personal numberTO_NUMBER
usingpython make_outbound_call.py
Enjoy chatting away!
The JavaScript guide is coming soon! Contact us at support@neuphonic.com if you need help in the meantime.
Twilio Setup for Inbound Calls
This section explains how to set up the server you made in the above section to handle incoming calls to your Twilio phone number.
- Get Your ngrok URL
- Get the ngrok URL you created in the last section when you set up your server.
- Configure Twilio
- Go to the Twilio Console
- Navgiate to
Phone Number
->Manage
->Active Number
and select your phone number. - Under
Voice Configuration
, set the URL in theA call comes in
section to your ngrok url,https://123d-45-678-912-3.ngrok-free.app/twilio/inbound_call
. - Set the HTTP method to POST.
- Call Your Twilio Number Go ahead and call your Twilio number. Your phone call will be connected through to the server and you can chat away!
Next Steps
This demo provides a basic introduction to integrating Neuphonic Agents with Twilio. To explore more advanced features and capabilities check out the Twilio Voice Documentation and Neuphonic Agents Documentation.