wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1670 from pipecat-ai/mb/daily-twilio-sip-example

Browse Source 
Add standalone Daily + Twilio SIP example
This commit is contained in:
Mark Backman
2025-05-05 10:57:16 -04:00
committed by GitHub
parent 2bebdbc371 0513d0b6a8
commit ddc6a9c695
11 changed files with 655 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
examples/phone-chatbot-daily-twilio-sip/README.md Normal file
View File

@@ -0,0 +1,138 @@
# Daily + Twilio SIP Voice Bot
This project demonstrates how to create a voice bot that can receive phone calls via Twilio and use Daily's SIP capabilities to enable voice conversations.
## How It Works
1. Twilio receives an incoming call to your phone number
2. Twilio calls your webhook server (`/call` endpoint)
3. The server creates a Daily room with SIP capabilities
4. The server starts the bot process with the room details
5. The caller is put on hold with music
6. The bot joins the Daily room and signals readiness
7. Twilio forwards the call to Daily's SIP endpoint
8. The caller and bot are connected, and the bot handles the conversation
## Prerequisites
- A Daily account with an API key
- A Twilio account with a phone number that supports voice
- OpenAI API key for the bot's intelligence
- Cartesia API key for text-to-speech
## Setup
1. Create a virtual environment and install dependencies
```bash
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
pip install -r requirements.txt
```
2. Set up environment variables
Copy the example file and fill in your API keys:
```bash
cp .env.example .env
# Edit .env with your API keys
```
3. Configure your Twilio webhook
In the Twilio console:
- Go to your phone number's configuration
- Set the webhook for "A Call Comes In" to your server's URL + "/call"
- For local testing, you can use ngrok to expose your local server
```bash
ngrok http 8000
# Then use the provided URL (e.g., https://abc123.ngrok.io/call) in Twilio
```
## Running the Server
Start the webhook server:
```bash
python server.py
```
## Testing
Call your Twilio phone number. The system should answer the call, put you on hold briefly, then connect you with the bot.
## Customizing the Bot
You can customize the bot's behavior by modifying the system prompt in `bot.py`.
### Changing the Hold Music
To change the ringing sound or hold music that callers hear while waiting to be connected to the bot, update the URL in `server.py`:
```python
resp = VoiceResponse()
resp.play(
url="https://your-custom-audio-file-url.mp3",
loop=10,
)
```
> Read [Twilio's guide](https://www.twilio.com/en-us/blog/adding-mp3-to-voice-call-using-twilio) on how to set up an mp3 in a voice call.
## Handling Multiple SIP Endpoints
The bot is configured to handle multiple `on_dialin_ready` events that might occur with multiple SIP endpoints. It ensures that each call is only forwarded once using a simple flag:
```python
# Flag to track if call has been forwarded
call_already_forwarded = False
@transport.event_handler("on_dialin_ready")
async def on_dialin_ready(transport, cdata):
nonlocal call_already_forwarded
# Skip if already forwarded
if call_already_forwarded:
logger.info("Call already forwarded, ignoring this event.")
return
# ... forwarding code ...
call_already_forwarded = True
```
Note that normally calls only require a single SIP endpoint. If you are planning to forward the call to a different number, you will need to set up 2 SIP endpoints: one for the initial call and one for the forwarded call. IMPORTANT: ensure that your `on_dialin_ready` handler only handles the first call.
## Daily SIP Configuration
The bot configures Daily rooms with SIP capabilities using these settings:
```python
sip_params = DailyRoomSipParams(
display_name="phone-user", # This will show up in the Daily UI; optional display the dialer's number
video=False, # Audio-only call
sip_mode="dial-in", # For receiving calls (vs. dial-out)
num_endpoints=1, # Number of SIP endpoints to create
)
```
## Troubleshooting
### Call is not being answered
- Check that your Twilio webhook is correctly configured
- Verify your Twilio account has sufficient funds
- Check the logs of both the server and bot processes
### Call connects but no bot is heard
- Ensure your Daily API key is correct and has SIP capabilities
- Check that the SIP endpoint is being correctly passed to the bot
- Verify that the Cartesia API key and voice ID are correct
### Bot starts but disconnects immediately
- Check the Daily and Twilio logs for any error messages
- Ensure your server has stable internet connectivity

183
examples/phone-chatbot-daily-twilio-sip/bot.py Normal file
View File

@@ -0,0 +1,183 @@
"""Twilio + Daily voice bot implementation."""
import argparse
import asyncio
import os
import sys
from dotenv import load_dotenv
from loguru import logger
from twilio.rest import Client
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
# Setup logging
load_dotenv()
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
# Initialize Twilio client
twilio_client = Client(os.getenv("TWILIO_ACCOUNT_SID"), os.getenv("TWILIO_AUTH_TOKEN"))
async def run_bot(room_url: str, token: str, call_id: str, sip_uri: str) -> None:
"""Run the voice bot with the given parameters.
Args:
room_url: The Daily room URL
token: The Daily room token
call_id: The Twilio call ID
sip_uri: The Daily SIP URI for forwarding the call
"""
logger.info(f"Starting bot with room: {room_url}")
logger.info(f"SIP endpoint: {sip_uri}")
call_already_forwarded = False
# Setup the Daily transport
transport = DailyTransport(
room_url,
token,
"Phone Bot",
DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
transcription_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
# Setup TTS service
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
# Setup LLM service
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
# Initialize LLM context with system prompt
messages = [
{
"role": "system",
"content": (
"You are a friendly phone assistant. Your responses will be read aloud, "
"so keep them concise and conversational. Avoid special characters or "
"formatting. Begin by greeting the caller and asking how you can help them today."
),
},
]
# Setup the conversational context
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
# Build the pipeline
pipeline = Pipeline(
[
transport.input(),
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
]
)
# Create the pipeline task
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True # Enable barge-in so callers can interrupt the bot
),
)
# Handle participant joining
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
logger.info(f"First participant joined: {participant['id']}")
await transport.capture_participant_transcription(participant["id"])
await task.queue_frames([context_aggregator.user().get_context_frame()])
# Handle participant leaving
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
logger.info(f"Participant left: {participant['id']}, reason: {reason}")
await task.cancel()
# Handle call ready to forward
@transport.event_handler("on_dialin_ready")
async def on_dialin_ready(transport, cdata):
nonlocal call_already_forwarded
# We only want to forward the call once
# The on_dialin_ready event will be triggered for each sip endpoint provisioned
if call_already_forwarded:
logger.warning("Call already forwarded, ignoring this event.")
return
logger.info(f"Forwarding call {call_id} to {sip_uri}")
try:
# Update the Twilio call with TwiML to forward to the Daily SIP endpoint
twilio_client.calls(call_id).update(
twiml=f"<Response><Dial><Sip>{sip_uri}</Sip></Dial></Response>"
)
logger.info("Call forwarded successfully")
call_already_forwarded = True
except Exception as e:
logger.error(f"Failed to forward call: {str(e)}")
raise
@transport.event_handler("on_dialin_connected")
async def on_dialin_connected(transport, data):
logger.debug(f"Dial-in connected: {data}")
@transport.event_handler("on_dialin_stopped")
async def on_dialin_stopped(transport, data):
logger.debug(f"Dial-in stopped: {data}")
@transport.event_handler("on_dialin_error")
async def on_dialin_error(transport, data):
logger.error(f"Dial-in error: {data}")
# If there is an error, the bot should leave the call
# This may be also handled in on_participant_left with
# await task.cancel()
@transport.event_handler("on_dialin_warning")
async def on_dialin_warning(transport, data):
logger.warning(f"Dial-in warning: {data}")
# Run the pipeline
runner = PipelineRunner()
await runner.run(task)
async def main():
"""Parse command line arguments and run the bot."""
parser = argparse.ArgumentParser(description="Daily + Twilio Voice Bot")
parser.add_argument("-u", type=str, required=True, help="Daily room URL")
parser.add_argument("-t", type=str, required=True, help="Daily room token")
parser.add_argument("-i", type=str, required=True, help="Twilio call ID")
parser.add_argument("-s", type=str, required=True, help="Daily SIP URI")
args = parser.parse_args()
# Validate required arguments
if not all([args.u, args.t, args.i, args.s]):
logger.error("All arguments (-u, -t, -i, -s) are required")
parser.print_help()
sys.exit(1)
await run_bot(args.u, args.t, args.i, args.s)
if __name__ == "__main__":
asyncio.run(main())

11
examples/phone-chatbot-daily-twilio-sip/env.example Normal file
View File

@@ -0,0 +1,11 @@
# Daily credentials
DAILY_API_KEY=your_daily_api_key
DAILY_API_URL=https://api.daily.co/v1
# Twilio credentials
TWILIO_ACCOUNT_SID=your_twilio_account_sid
TWILIO_AUTH_TOKEN=your_twilio_auth_token
# Service keys
OPENAI_API_KEY=your_openai_api_key
CARTESIA_API_KEY=your_cartesia_api_key

5
examples/phone-chatbot-daily-twilio-sip/requirements.txt Normal file
View File

@@ -0,0 +1,5 @@
pipecat-ai[daily,elevenlabs,openai,silero]
fastapi==0.115.6
uvicorn
python-dotenv
twilio

116
examples/phone-chatbot-daily-twilio-sip/server.py Normal file
View File

@@ -0,0 +1,116 @@
"""Webhook server to handle Twilio calls and start the voice bot."""
import os
import shlex
import subprocess
from contextlib import asynccontextmanager
import aiohttp
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import PlainTextResponse
from twilio.twiml.voice_response import VoiceResponse
from utils.daily_helpers import create_sip_room
# Load environment variables
load_dotenv()
# Initialize FastAPI app with aiohttp session
@asynccontextmanager
async def lifespan(app: FastAPI):
# Create aiohttp session to be used for Daily API calls
app.state.session = aiohttp.ClientSession()
yield
# Close session when shutting down
await app.state.session.close()
app = FastAPI(lifespan=lifespan)
@app.post("/call", response_class=PlainTextResponse)
async def handle_call(request: Request):
"""Handle incoming Twilio call webhook."""
print("Received call webhook from Twilio")
try:
# Get form data from Twilio webhook
form_data = await request.form()
data = dict(form_data)
# Extract call ID (required to forward the call later)
call_sid = data.get("CallSid")
if not call_sid:
raise HTTPException(status_code=400, detail="Missing CallSid in request")
# Extract the caller's phone number
caller_phone = str(data.get("From", "unknown-caller"))
print(f"Processing call with ID: {call_sid} from {caller_phone}")
# Create a Daily room with SIP capabilities
try:
room_details = await create_sip_room(request.app.state.session, caller_phone)
except Exception as e:
print(f"Error creating Daily room: {e}")
raise HTTPException(status_code=500, detail=f"Failed to create Daily room: {str(e)}")
# Extract necessary details
room_url = room_details["room_url"]
token = room_details["token"]
sip_endpoint = room_details["sip_endpoint"]
# Make sure we have a SIP endpoint
if not sip_endpoint:
raise HTTPException(status_code=500, detail="No SIP endpoint provided by Daily")
# Start the bot process
bot_cmd = f"python bot.py -u {room_url} -t {token} -i {call_sid} -s {sip_endpoint}"
try:
# Use shlex to properly split the command for subprocess
cmd_parts = shlex.split(bot_cmd)
# CHANGE: Keep stdout/stderr for debugging
# Start the bot in the background but capture output
subprocess.Popen(
cmd_parts,
# Don't redirect output so we can see logs
# stdout=subprocess.DEVNULL,
# stderr=subprocess.DEVNULL
)
print(f"Started bot process with command: {bot_cmd}")
except Exception as e:
print(f"Error starting bot: {e}")
raise HTTPException(status_code=500, detail=f"Failed to start bot: {str(e)}")
# Generate TwiML response to put the caller on hold with music
# You can replace the URL with your own music file
# or use Twilio's built-in music on hold
# https://www.twilio.com/docs/voice/twiml/play#music-on-hold
resp = VoiceResponse()
resp.play(
url="https://therapeutic-crayon-2467.twil.io/assets/US_ringback_tone.mp3",
loop=10,
)
return str(resp)
except HTTPException:
raise
except Exception as e:
print(f"Unexpected error: {str(e)}")
raise HTTPException(status_code=500, detail=f"Server error: {str(e)}")
@app.get("/health")
async def health_check():
"""Simple health check endpoint."""
return {"status": "healthy"}
if __name__ == "__main__":
# Run the server
port = int(os.getenv("PORT", "8000"))
print(f"Starting server on port {port}")
uvicorn.run("server:app", host="0.0.0.0", port=port, reload=True)

76
examples/phone-chatbot-daily-twilio-sip/utils/daily_helpers.py Normal file
View File

@@ -0,0 +1,76 @@
"""Helper functions for interacting with the Daily API."""
import os
from typing import Dict, Optional
import aiohttp
from dotenv import load_dotenv
from pipecat.transports.services.helpers.daily_rest import (
DailyRESTHelper,
DailyRoomParams,
DailyRoomProperties,
DailyRoomSipParams,
)
load_dotenv()
# Initialize Daily API helper
async def get_daily_helper(session: Optional[aiohttp.ClientSession] = None) -> DailyRESTHelper:
"""Get a Daily REST helper with the configured API key."""
if session is None:
session = aiohttp.ClientSession()
return DailyRESTHelper(
daily_api_key=os.getenv("DAILY_API_KEY", ""),
daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
aiohttp_session=session,
)
async def create_sip_room(
session: Optional[aiohttp.ClientSession] = None, caller_phone: str = "unknown-caller"
) -> Dict[str, str]:
"""Create a Daily room with SIP capabilities for phone calls.
Args:
session: Optional aiohttp session to use for API calls
caller_phone: The phone number of the caller to use in display name
Returns:
Dictionary with room URL, token, and SIP endpoint
"""
daily_helper = await get_daily_helper(session)
# Configure SIP parameters
sip_params = DailyRoomSipParams(
display_name=caller_phone,
video=False,
sip_mode="dial-in",
num_endpoints=1,
)
# Create room properties with SIP enabled
properties = DailyRoomProperties(
sip=sip_params,
enable_dialout=True, # Needed for outbound calls if you expand the bot
enable_chat=False, # No need for chat in a voice bot
start_video_off=True, # Voice only
)
# Create room parameters
params = DailyRoomParams(properties=properties)
# Create the room
try:
room = await daily_helper.create_room(params=params)
print(f"Created room: {room.url} with SIP endpoint: {room.config.sip_endpoint}")
# Get token for the bot to join
token = await daily_helper.get_token(room.url, 24 * 60 * 60) # 24 hours validity
return {"room_url": room.url, "token": token, "sip_endpoint": room.config.sip_endpoint}
except Exception as e:
print(f"Error creating room: {e}")
raise

271
examples/phone-chatbot/README.md
View File

@@ -235,10 +235,10 @@ For incoming calls from customers, Daily will send a webhook to your `/start` en
```json
{
"From": "+CALLERS_PHONE",
"To": "$PURCHASED_PHONE",
"callId": "callid-read-only-string",
"callDomain": "callDomain-read-only-string"
"From": "+CALLERS_PHONE",
"To": "$PURCHASED_PHONE",
"callId": "callid-read-only-string",
"callDomain": "callDomain-read-only-string"
}
```
@@ -266,63 +266,63 @@ When making requests to the `/start` endpoint, the config object can include:
```json
{
"config": {
"prompts": [
{
"name": "call_transfer_initial_prompt",
"text": "Your custom prompt here"
},
{
"name": "call_transfer_prompt",
"text": "Your custom prompt here"
},
{
"name": "call_transfer_finished_prompt",
"text": "Your custom prompt here"
},
{
"name": "voicemail_detection_prompt",
"text": "Your custom prompt here"
},
{
"name": "voicemail_prompt",
"text": "Your custom prompt here"
},
{
"name": "human_conversation_prompt",
"text": "Your custom prompt here"
}
],
"dialin_settings": {
"From": "+CALLERS_PHONE",
"To": "$PURCHASED_PHONE",
"callId": "callid-read-only-string",
"callDomain": "callDomain-read-only-string"
},
"dialout_settings": [
{
"phoneNumber": "+12345678910",
"callerId": "caller-id-uuid",
"sipUri": "sip:maria@example.com"
}
],
"call_transfer": {
"mode": "dialout",
"speakSummary": true,
"storeSummary": false,
"operatorNumber": "+12345678910",
"testInPrebuilt": false
},
"voicemail_detection": {
"testInPrebuilt": true
},
"simple_dialin": {
"testInPrebuilt": true
},
"simple_dialout": {
"testInPrebuilt": true
}
}
"config": {
"prompts": [
{
"name": "call_transfer_initial_prompt",
"text": "Your custom prompt here"
},
{
"name": "call_transfer_prompt",
"text": "Your custom prompt here"
},
{
"name": "call_transfer_finished_prompt",
"text": "Your custom prompt here"
},
{
"name": "voicemail_detection_prompt",
"text": "Your custom prompt here"
},
{
"name": "voicemail_prompt",
"text": "Your custom prompt here"
},
{
"name": "human_conversation_prompt",
"text": "Your custom prompt here"
}
],
"dialin_settings": {
"From": "+CALLERS_PHONE",
"To": "$PURCHASED_PHONE",
"callId": "callid-read-only-string",
"callDomain": "callDomain-read-only-string"
},
"dialout_settings": [
{
"phoneNumber": "+12345678910",
"callerId": "caller-id-uuid",
"sipUri": "sip:maria@example.com"
}
],
"call_transfer": {
"mode": "dialout",
"speakSummary": true,
"storeSummary": false,
"operatorNumber": "+12345678910",
"testInPrebuilt": false
},
"voicemail_detection": {
"testInPrebuilt": true
},
"simple_dialin": {
"testInPrebuilt": true
},
"simple_dialout": {
"testInPrebuilt": true
}
}
}
```
@@ -393,19 +393,19 @@ The following table shows which feature combinations are supported when making r
```json
{
"config": {
"dialin_settings": {
"from": "+12345678901",
"to": "+19876543210",
"call_id": "call-id-string",
"call_domain": "domain-string"
},
"call_transfer": {
"mode": "dialout",
"speakSummary": true,
"operatorNumber": "+12345678910"
}
}
"config": {
"dialin_settings": {
"from": "+12345678901",
"to": "+19876543210",
"call_id": "call-id-string",
"call_domain": "domain-string"
},
"call_transfer": {
"mode": "dialout",
"speakSummary": true,
"operatorNumber": "+12345678910"
}
}
}
```
@@ -413,14 +413,14 @@ The following table shows which feature combinations are supported when making r
```json
{
"config": {
"call_transfer": {
"mode": "dialout",
"speakSummary": true,
"operatorNumber": "+12345678910",
"testInPrebuilt": true
}
}
"config": {
"call_transfer": {
"mode": "dialout",
"speakSummary": true,
"operatorNumber": "+12345678910",
"testInPrebuilt": true
}
}
}
```
@@ -428,11 +428,11 @@ The following table shows which feature combinations are supported when making r
```json
{
"config": {
"voicemail_detection": {
"testInPrebuilt": true
}
}
"config": {
"voicemail_detection": {
"testInPrebuilt": true
}
}
}
```
@@ -440,16 +440,16 @@ The following table shows which feature combinations are supported when making r
```json
{
"config": {
"dialout_settings": [
{
"phoneNumber": "+12345678910"
}
],
"voicemail_detection": {
"testInPrebuilt": false
}
}
"config": {
"dialout_settings": [
{
"phoneNumber": "+12345678910"
}
],
"voicemail_detection": {
"testInPrebuilt": false
}
}
}
```
@@ -457,15 +457,15 @@ The following table shows which feature combinations are supported when making r
```json
{
"config": {
"dialin_settings": {
"from": "+12345678901",
"to": "+19876543210",
"call_id": "call-id-string",
"call_domain": "domain-string"
},
"simple_dialin": {}
}
"config": {
"dialin_settings": {
"from": "+12345678901",
"to": "+19876543210",
"call_id": "call-id-string",
"call_domain": "domain-string"
},
"simple_dialin": {}
}
}
```
@@ -473,11 +473,11 @@ The following table shows which feature combinations are supported when making r
```json
{
"config": {
"simple_dialin": {
"testInPrebuilt": true
}
}
"config": {
"simple_dialin": {
"testInPrebuilt": true
}
}
}
```
@@ -485,14 +485,14 @@ The following table shows which feature combinations are supported when making r
```json
{
"config": {
"dialout_settings": [
{
"phoneNumber": "+12345678910"
}
],
"simple_dialout": {}
}
"config": {
"dialout_settings": [
{
"phoneNumber": "+12345678910"
}
],
"simple_dialout": {}
}
}
```
@@ -500,37 +500,14 @@ The following table shows which feature combinations are supported when making r
```json
{
"config": {
"simple_dialout": {
"testInPrebuilt": true
}
}
"config": {
"simple_dialout": {
"testInPrebuilt": true
}
}
}
```
## Using Twilio (Alternative)
To use Twilio for call handling:
1. Start the bot runner:
```shell
python bot_runner.py --host localhost
```
2. Start ngrok:
```shell
ngrok http --domain yourdomain.ngrok.app 7860
```
3. In another terminal, run the Twilio bot:
```shell
python bot_twilio.py
```
Make requests to `/start_twilio_bot` for Twilio-specific functionality.
## Deployment
See Pipecat Cloud deployment docs for how to deploy this example: https://docs.pipecat.daily.co/agents/deploy

94
examples/phone-chatbot/bot_runner.py
View File

@@ -20,8 +20,7 @@ from bot_runner_helpers import (
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, PlainTextResponse
from twilio.twiml.voice_response import VoiceResponse
from fastapi.responses import JSONResponse
from pipecat.transports.services.helpers.daily_rest import (
DailyRESTHelper,
@@ -125,32 +124,6 @@ async def start_bot(room_details: Dict[str, str], body: Dict[str, Any], example:
raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}")
async def start_twilio_bot(room_details: Dict[str, str], call_id: str) -> bool:
"""Start a Twilio bot process with the given configuration.
Args:
room_details: Room URL, token, and SIP endpoint
call_id: Twilio call ID (CallSid)
Returns:
Boolean indicating success
"""
room_url = room_details["room"]
token = room_details["token"]
sip_endpoint = room_details["sip_endpoint"]
# Format command for Twilio bot
bot_proc = f"python3 -m bot_twilio -u {room_url} -t {token} -i {call_id} -s {sip_endpoint}"
print(f"Starting Twilio bot. Room: {room_url}")
try:
command_parts = shlex.split(bot_proc)
subprocess.Popen(command_parts, bufsize=1, cwd=os.path.dirname(os.path.abspath(__file__)))
return True
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}")
# ----------------- API Setup ----------------- #
@@ -180,47 +153,6 @@ app.add_middleware(
# ----------------- API Endpoints ----------------- #
@app.post("/twilio_start_bot", response_class=PlainTextResponse)
async def twilio_start_bot(request: Request):
"""Handle incoming Twilio webhook calls and start a Twilio bot.
This endpoint is called directly by Twilio as a webhook when a call is received.
It puts the call on hold with music and starts a bot that will handle the call.
"""
print("POST /twilio_start_bot")
# Get form data from Twilio webhook
try:
form_data = await request.form()
data = dict(form_data)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Failed to parse Twilio form data: {str(e)}")
# Get default room URL from environment
room_url = os.getenv("DAILY_SAMPLE_ROOM_URL", None)
# Extract call ID from Twilio data
call_id = data.get("CallSid")
if not call_id:
raise HTTPException(status_code=400, detail="Missing 'CallSid' in request")
print(f"CallId: {call_id}")
# Create Daily room for the Twilio call
room_details = await create_daily_room(room_url, None) # No special config for Twilio rooms
# Start the Twilio bot
await start_twilio_bot(room_details, call_id)
# Put the call on hold until the bot is ready to handle it
# The bot will update the call with the SIP URI when it's ready
resp = VoiceResponse()
resp.play(
url="http://com.twilio.sounds.music.s3.amazonaws.com/MARKOVICHAMP-Borghestral.mp3", loop=10
)
return str(resp)
@app.post("/start")
async def handle_start_request(request: Request) -> JSONResponse:
"""Unified endpoint to handle bot configuration for different scenarios."""
@@ -228,21 +160,7 @@ async def handle_start_request(request: Request) -> JSONResponse:
room_url = os.getenv("DAILY_SAMPLE_ROOM_URL", None)
try:
# Check if this is form data (from Twilio) or JSON
content_type = request.headers.get("content-type", "").lower()
if "application/x-www-form-urlencoded" in content_type:
# Handle form data from Twilio
form_data = await request.form()
data = dict(form_data)
# Check for CallSid which indicates this is a Twilio webhook
if "CallSid" in data:
# Redirect to Twilio handler for backward compatibility
return await twilio_start_bot(request)
else:
# Parse JSON request data
data = await request.json()
data = await request.json()
# Handle webhook test
if "test" in data:
@@ -298,14 +216,6 @@ async def handle_start_request(request: Request) -> JSONResponse:
return JSONResponse(response)
except json.JSONDecodeError:
# Check if this might be form data from Twilio
try:
content_type = request.headers.get("content-type", "").lower()
if "application/x-www-form-urlencoded" in content_type:
return await twilio_start_bot(request)
except Exception:
pass
raise HTTPException(status_code=400, detail="Invalid JSON in request body")
except Exception as e:
raise HTTPException(status_code=400, detail=f"Request processing error: {str(e)}")

122
examples/phone-chatbot/bot_twilio.py
View File

@@ -1,122 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import asyncio
import os
import sys
from dotenv import load_dotenv
from loguru import logger
from twilio.rest import Client
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
load_dotenv(override=True)
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
twilio_account_sid = os.getenv("TWILIO_ACCOUNT_SID")
twilio_auth_token = os.getenv("TWILIO_AUTH_TOKEN")
twilioclient = Client(twilio_account_sid, twilio_auth_token)
daily_api_key = os.getenv("DAILY_API_KEY", "")
async def main(room_url: str, token: str, callId: str, sipUri: str):
# dialin_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.
transport = DailyTransport(
room_url,
token,
"Chatbot",
DailyParams(
api_key=daily_api_key,
dialin_settings=None, # Not required for Twilio
audio_in_enabled=True,
audio_out_enabled=True,
video_out_enabled=False,
vad_analyzer=SileroVADAnalyzer(),
transcription_enabled=True,
),
)
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": "You are Chatbot, a friendly, helpful robot. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by saying 'Hello! Who dares dial me at this hour?!'.",
},
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(),
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
await transport.capture_participant_transcription(participant["id"])
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
await task.cancel()
@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..
print(f"Forwarding call: {callId} {sipUri}")
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(f"Failed to forward call: {str(e)}")
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Pipecat Simple ChatBot")
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.s))

2
examples/phone-chatbot/env.example
View File

@@ -5,8 +5,6 @@ DEEPGRAM_API_KEY=
OPENAI_API_KEY=
GOOGLE_API_KEY
CARTESIA_API_KEY=
TWILIO_ACCOUNT_SID=
TWILIO_AUTH_TOKEN=
DIAL_IN_FROM_NUMBER=
DIAL_OUT_TO_NUMBER=
OPERATOR_NUMBER=

1
examples/phone-chatbot/requirements.txt
View File

@@ -2,5 +2,4 @@ pipecat-ai[daily,cartesia,deepgram,openai,google,silero]
fastapi==0.115.6
uvicorn
python-dotenv
twilio
python-multipart