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

Merge branch 'main' into fixing_sound_mixer_small_webrtc

Browse Source 
# Conflicts:
#	CHANGELOG.md
This commit is contained in:
Filipi Fuchter
2025-05-05 17:43:47 -03:00
parent d75815ba48 b323a7bd88
commit e91fb6bb29
22 changed files with 750 additions and 409 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
CHANGELOG.md
View File

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- Added support to `RimeHttpTTSService` for the `arcana` model.
### Fixed
- Refactored how the `start` method is handled in `SmallWebRTCOutputTransport` by
@@ -14,6 +18,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
alongside `SmallWebRTCTransport`, preventing unnecessary CPU usage and avoiding the
output being flooded with silent frames when no new audio is available.
- Remove custom audio tracks from `DailyTransport` before leaving.
## [0.0.66] - 2025-05-02
### Added
@@ -129,7 +135,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
not a very common case.
- Added `RivaSegmentedSTTService`, which allows Riva offline/batch models, such
as to be "canary-1b-asr" used in Pipecat.
as to be "canary-1b-asr" used in Pipecat.
### Deprecated

80
examples/deployment/pipecat-cloud-example/bot.py
View File

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
import aiohttp
@@ -21,44 +22,23 @@ from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
# Check if we're in local development mode
LOCAL_RUN = os.getenv("LOCAL_RUN")
if LOCAL_RUN:
import asyncio
import webbrowser
try:
from local_runner import configure
except ImportError:
logger.error("Could not import local_runner module. Local development mode may not work.")
# Load environment variables
load_dotenv(override=True)
# Check if we're in local development mode
LOCAL_RUN = os.getenv("LOCAL_RUN")
async def main(room_url: str, token: str):
async def main(transport: DailyTransport):
"""Main pipeline setup and execution function.
Args:
room_url: The Daily room URL
token: The Daily room token
transport: The DailyTransport object for the bot
"""
logger.debug("Starting bot in room: {}", room_url)
transport = DailyTransport(
room_url,
token,
"bot",
DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
transcription_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
logger.debug("Starting bot")
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22"
api_key=os.getenv("CARTESIA_API_KEY"), voice_id="71a7ad14-091c-4e8e-a314-022ece01c121"
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
@@ -126,10 +106,25 @@ async def bot(args: DailySessionArguments):
body: The configuration object from the request body
session_id: The session ID for logging
"""
from pipecat.audio.filters.krisp_filter import KrispFilter
logger.info(f"Bot process initialized {args.room_url} {args.token}")
transport = DailyTransport(
args.room_url,
args.token,
"Pipecat Bot",
DailyParams(
audio_in_enabled=True,
audio_in_filter=None if LOCAL_RUN else KrispFilter(),
audio_out_enabled=True,
transcription_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
try:
await main(args.room_url, args.token)
await main(transport)
logger.info("Bot process completed")
except Exception as e:
logger.exception(f"Error in bot process: {str(e)}")
@@ -137,18 +132,27 @@ async def bot(args: DailySessionArguments):
# Local development functions
async def local_main():
async def local_daily():
"""Function for local development testing."""
from local_runner import configure
try:
async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)
logger.warning("_")
logger.warning("_")
logger.warning(f"Talk to your voice agent here: {room_url}")
logger.warning("_")
logger.warning("_")
webbrowser.open(room_url)
await main(room_url, token)
transport = DailyTransport(
room_url,
token,
"Pipecat Bot",
DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
transcription_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
await main(transport)
except Exception as e:
logger.exception(f"Error in local development mode: {e}")
@@ -156,6 +160,6 @@ async def local_main():
# Local development entry point
if LOCAL_RUN and __name__ == "__main__":
try:
asyncio.run(local_main())
asyncio.run(local_daily())
except Exception as e:
logger.exception(f"Failed to run in local mode: {e}")

4
examples/deployment/pipecat-cloud-example/env.example
View File

@@ -1,2 +1,4 @@
CARTESIA_API_KEY=
OPENAI_API_KEY=
OPENAI_API_KEY=
# Local dev only
DAILY_API_KEY=

1
examples/deployment/pipecat-cloud-example/local_runner.py
View File

@@ -7,6 +7,7 @@
import os
import aiohttp
from fastapi import HTTPException
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams

2
examples/deployment/pipecat-cloud-example/pcc-deploy.toml
View File

@@ -1,6 +1,8 @@
agent_name = "my-first-agent"
image = "your-username/my-first-agent:0.1"
image_credentials = "your-dockerhub-creds"
secret_set = "my-first-agent-secrets"
enable_krisp = true
[scaling]
min_instances = 0

3
examples/foundational/07q-interruptible-rime-http.py
View File

@@ -44,7 +44,8 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
tts = RimeHttpTTSService(
api_key=os.getenv("RIME_API_KEY", ""),
voice_id="rex",
voice_id="luna",
model="arcana",
aiohttp_session=session,
)

3
examples/foundational/13c-gladia-translation.py
View File

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
from dotenv import load_dotenv
@@ -39,7 +40,7 @@ class TranscriptionLogger(FrameProcessor):
print(f"Translation ({frame.language}): {frame.text}")
async def run_bot(webrtc_connection: SmallWebRTCConnection):
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
logger.info(f"Starting bot")
transport = SmallWebRTCTransport(

14
examples/foundational/35-pattern-pair-voice-switching.py
View File

@@ -45,6 +45,7 @@ Note:
"""
import argparse
import asyncio
import os
from dotenv import load_dotenv
@@ -102,8 +103,17 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
voice_name = match.content.strip().lower()
if voice_name in VOICE_IDS:
voice_id = VOICE_IDS[voice_name]
tts.set_voice(voice_id)
logger.info(f"Switched to {voice_name} voice")
# Create task to reset the TTS context after voice change
async def change_voice():
# First flush any existing audio to finish the current context
await tts.flush_audio()
# Then set the new voice
tts.set_voice(voice_id)
logger.info(f"Switched to {voice_name} voice")
# Schedule the voice change task
asyncio.create_task(change_voice())
else:
logger.warning(f"Unknown voice: {voice_name}")

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

11
src/pipecat/services/rime/tts.py
View File

@@ -400,6 +400,13 @@ class RimeHttpTTSService(TTSService):
payload["modelId"] = self._model_name
payload["samplingRate"] = self.sample_rate
# Arcana does not support PCM audio
if payload["modelId"] == "arcana":
headers["Accept"] = "audio/wav"
need_to_strip_wav_header = True
else:
need_to_strip_wav_header = False
try:
await self.start_ttfb_metrics()
@@ -420,6 +427,10 @@ class RimeHttpTTSService(TTSService):
CHUNK_SIZE = 1024
async for chunk in response.content.iter_chunked(CHUNK_SIZE):
if need_to_strip_wav_header and chunk.startswith(b"RIFF"):
chunk = chunk[44:]
need_to_strip_wav_header = False
if len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)

2
src/pipecat/services/ultravox/stt.py
View File

@@ -425,7 +425,7 @@ class UltravoxSTTService(AIService):
if "content" in delta:
new_text = delta["content"]
if new_text:
yield LLMTextFrame(text=new_text.strip())
yield LLMTextFrame(text=new_text)
# Stop processing metrics after completion
await self.stop_processing_metrics()

12
src/pipecat/transports/services/daily.py
View File

@@ -578,6 +578,10 @@ class DailyTransportClient(EventHandler):
if self._params.transcription_enabled:
await self._stop_transcription()
# Remove any custom tracks, if any.
for track_name, _ in self._audio_sources.items():
await self.remove_custom_audio_track(track_name)
try:
error = await self._leave()
if not error:
@@ -741,6 +745,14 @@ class DailyTransportClient(EventHandler):
return audio_source
async def remove_custom_audio_track(self, track_name: str):
future = self._get_event_loop().create_future()
self._client.remove_custom_audio_track(
track_name=track_name,
completion=completion_callback(future),
)
await future
async def update_transcription(self, participants=None, instance_id=None):
future = self._get_event_loop().create_future()
self._client.update_transcription(