​
Things you’ll need

  • An active Twilio developer key.
  • One or more Twilio provisioned phone numbers (covered below).
  • The Twilio Python client library (pip install twilio).

Prefer to look at code? See the example project!

We have a complete dialin-ready project using Daily as both a transport and Twilio as a phone provider in the Pipecat repo. This guide references the project and steps through the important parts that make dial-in work.

​
Getting a phone number

  • Visit console.twilio.com and purchase a new phone number (or via the API.)
  • Ensure your purchased number supports Voice capabilities.
  • Ensure your purchased number appears in your ‘active numbers’ list.

​
Project setup

You’ll need to set two environment variables for your project: TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN, value for which you can obtain via the Twilio console.

.env
DAILY_API_KEY=...
TWILIO_ACCOUNT_SID=...
TWILIO_AUTH_TOKEN=...

# ... service API keys etc

​
Configuring your bot runner

You’ll need a HTTP service that can receive incoming call hooks and trigger a new agent session. We discussed the concept of a bot runner in the deployment section, which we’ll build on here to add support for incoming phone calls.

Within the start_bot method, we’ll setup the Twilio client like so:

bot_runner.py
@app.post("/start_bot", response_class=PlainTextResponse)
async def start_bot (request: Request):
    # Note: unlike Daily, Twilio uses FormData and PlainText responses
    data = None
    try:
        form_data = await request.form()
        data = dict(form_data)
    except Exception:
        pass

    # We'll need to pass through the caller ID to our Pipecat bot
    callId = data.get('CallSid')

    # Create a room and spawn bot
    # ... see below for creating a room
    proc = subprocess.Popen(
        [
            f"python3 -m bot_twilio -u {room.url} -t {token} -i {callId} -s {room.config.sip_endpoint}"
        ],
        shell=True,
        bufsize=1,
        cwd=os.path.dirname(os.path.abspath(__file__))
    )

    # We have the room and the SIP URI,
    # but we do not know if the Daily SIP Worker and the Bot have joined the call
    # put the call on hold until the 'on_dialin_ready' fires.
    # The bot will call forward_twilio_call when it ready.
    resp = VoiceResponse()
    resp.play(url="http://com.twilio.sounds.music.s3.amazonaws.com/MARKOVICHAMP-Borghestral.mp3", loop=10)
    return str(resp)

​
Creating a new SIP-enabled room

We’ll need to configure the Daily room to be setup to receive SIP connections. daily-helpers.py as part of the Pipecat service, has some useful imports that make this easy. We just need to pass through new SIP parameters as part of room creation:

bot_runner.py

from pipecat.transports.services.helpers.daily_rest import DailyRoomParams, DailyRoomProperties, DailyRoomSipParams

params = DailyRoomParams(
    properties=DailyRoomProperties(
        sip=DailyRoomSipParams(
            display_name = "sip-dialin"
            video = False
            sip_mode = "dial-in"
            num_endpoints = 1
        )
    )
)

# Create sip-enabled Daily room via REST
try:
    room: DailyRoomObject = daily_rest_helper.create_room(params=params)
except Exception as e:
    raise HTTPException(
        status_code=500,
        detail=f"Unable to provision room {e}")

print (f"Daily room returned {room.url} {room.config.sip_endpoint}")

​
Setup the Twilio webhook on your phone number

Twilio needs to know where our start_bot URL is located as a webhook which it can trigger when a user dials a phone number. You can do this via the Twilio console or via the Twilio API:

If you want to test locally, you can expose your web method using a service such as ngrok.

Example ngrok tunnel
python bot_runner.py --host localhost --port 7860 --reload
ngrok http localhost:7860

# E.g: https://123.ngrok.app/start_bot

​
A quick test

With the webhook configured, you should now be able to dial your Twilio number and trigger the start_bot URL on your bot runner.

Assuming your webhook is configured correctly, we now need to configure our bot to signal to Twilio when it is ready for the call to be forwarded.

​
Configuring your Pipecat bot

Let’s take a look at bot_twilio.py and step through the configuration.

First, it’s setup to receive additional command line parameters which are passed through to the DailyTransport object:

from pipecat.transports.services.daily import DailyParams, DailyTransport, DailyDialinSettings

from twilio.twiml.voice_response import Dial, VoiceResponse, Sip
from twilio.rest import Client

twilio_account_sid = os.getenv('TWILIO_ACCOUNT_SID')
twilio_auth_token = os.getenv('TWILIO_AUTH_TOKEN')
twilio = Client(twilio_account_sid, twilio_auth_token)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Pipecat Daily Example")
    parser.add_argument("-u", type=str, help="Room URL")
    parser.add_argument("-t", type=str, help="Token")
    parser.add_argument("-i", type=str, help="Call ID")
    parser.add_argument("-s", type=str, help="SIP URI")
    config = parser.parse_args()

    asyncio.run(main(config.u, config.t, config.i, config.d, config.s))

async def main(room_url: str, token: str, callId: str, sipUri: str):
    async with aiohttp.ClientSession() as session:
        # diallin_settings are only needed if Daily's SIP URI is used
        # If you are handling this via Twilio, Telnyx, set this to None
        # and handle call-forwarding when on_dialin_ready fires.
        diallin_settings = None

        transport = DailyTransport(
            room_url,
            token,
            "Chatbot",
            DailyParams(
                api_url=daily_api_path,
                api_key=daily_api_key,
                dialin_settings=diallin_settings,
                audio_in_enabled=True,
                audio_out_enabled=True,
                camera_out_enabled=False,
                vad_enabled=True,
                vad_analyzer=SileroVADAnalyzer(),
                transcription_enabled=True,
            )
        )

        # ... your bot code here

        @transport.event_handler("on_dialin_ready")
        async def on_dialin_ready(transport, cdata):
            # For Twilio, Telnyx, etc. You need to update the state of the call
            # and forward it to the sip_uri..
            try:
                # The TwiML is updated using Twilio's client library
                call = twilioclient.calls(callId).update(
                    twiml=f'<Response><Dial><Sip>{sipUri}</Sip></Dial></Response>'
                )
            except Exception as e:
                raise Exception(detail=f"Failed to forward call: {str(e)}")