removed space from event handler

.github/workflows/publish_test.yaml

@@ -1,6 +1,10 @@
 name: publish-test
 
-on: workflow_dispatch
+on:
+  workflow_dispatch:
+  push:
+    branches:
+      - main
 
 jobs:
   build:
@@ -10,6 +14,7 @@ jobs:
       - name: Checkout repo
         uses: actions/checkout@v4
         with:
+          ref: ${{ github.event.inputs.gitref }}
           fetch-tags: true
           fetch-depth: 100
       - name: Set up Python

.github/workflows/tests.yaml

@@ -25,7 +25,14 @@ jobs:
         id: setup_python
         uses: actions/setup-python@v4
         with:
-          python-version: "3.10"
+          python-version: '3.10'
+      - name: Cache virtual environment
+        uses: actions/cache@v3
+        with:
+          # We are hashing requirements-dev.txt and requirements-extra.txt which
+          # contain all dependencies needed to run the tests and examples.
+          key: venv-${{ runner.os }}-${{ steps.setup_python.outputs.python-version}}-${{ hashFiles('linux-py3.10-requirements.txt') }}-${{ hashFiles('dev-requirements.txt') }}
+          path: .venv
       - name: Install system packages
         run: sudo apt-get install -y portaudio19-dev
       - name: Setup virtual environment
@@ -35,7 +42,7 @@ jobs:
         run: |
           source .venv/bin/activate
           python -m pip install --upgrade pip
-          pip install -r dev-requirements.txt
+          pip install -r linux-py3.10-requirements.txt -r dev-requirements.txt
       - name: Test with pytest
         run: |
           source .venv/bin/activate

CHANGELOG.md

@@ -5,375 +5,6 @@ All notable changes to **pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
-
-### Added
-
-- A clock can now be specified to `PipelineTask` (defaults to
-  `SystemClock`). This clock will be passed to each frame processor via the
-  `StartFrame`.
-
-- Added pipeline clocks. A pipeline clock is used by the output transport to
-  know when a frame needs to be presented. For that, all frames now have an
-  optional `pts` field (prensentation timestamp). There's currently just one
-  clock implementation `SystemClock` and the `pts` field is currently only used
-  for `TextFrame`s (audio and image frames will be next).
-
-- `DailyTransport` now supports setting the audio bitrate to improve audio
-  quality through the `DailyParams.audio_out_bitrate` parameter. The new
-  default is 96kbps.
-
-- `DailyTransport` now uses the number of audio output channels (1 or 2) to set
-  mono or stereo audio when needed.
-
-- Interruptions support has been added to `TwilioFrameSerializer` when using
-  `FastAPIWebsocketTransport`.
-
-- Added new `LmntTTSService` text-to-speech service.
-  (see https://www.lmnt.com/)
-
-- Added `TTSModelUpdateFrame`, `TTSLanguageUpdateFrame`, `STTModelUpdateFrame`,
-  and `STTLanguageUpdateFrame` frames to allow you to switch models, language
-  and voices in TTS and STT services.
-
-- Added new `transcriptions.Language` enum.
-
-### Changed
-
-- `CartesiaTTSService` and `ElevenLabsTTSService` now add presentation
-  timestamps to their text output. This allows the output transport to push the
-  text frames downstream at almost the same time the words are spoken. We say
-  "almost" because currently the audio frames don't have presentation timestamp
-  but they should be played at roughly the same time.
-
-- `DailyTransport.on_joined` event now returns the full session data instead of
-  just the participant.
-
-- `CartesiaTTSService` is now a subclass of `TTSService`.
-
-- `DeepgramSTTService` is now a subclass of `STTService`.
-
-- `WhisperSTTService` is now a subclass of `SegmentedSTTService`. A
-  `SegmentedSTTService` is a `STTService` where the provided audio is given in a
-  big chunk (i.e. from when the user starts speaking until the user stops
-  speaking) instead of a continous stream.
-
-### Fixed
-
-- `StartFrame` should be the first frame every processor receives to avoid
-  situations where things are not initialized (because initialization happens on
-  `StartFrame`) and other frames come in resulting in undesired behavior.
-
-### Performance
-
-- `obj_id()` and `obj_count()` now use `itertools.count` avoiding the need of
-  `threading.Lock`.
-
-## [0.0.41] - 2024-08-22
-
-### Added
-
-- Added `LivekitFrameSerializer` audio frame serializer.
-
-### Fixed
-
-- Fix `FastAPIWebsocketOutputTransport` variable name clash with subclass.
-
-- Fix an `AnthropicLLMService` issue with empty arguments in function calling.
-
-### Other
-
-- Fixed `studypal` example errors.
-
-## [0.0.40] - 2024-08-20
-
-### Added
-
-- VAD parameters can now be dynamicallt updated using the
-  `VADParamsUpdateFrame`.
-
-- `ErrorFrame` has now a `fatal` field to indicate the bot should exit if a
-  fatal error is pushed upstream (false by default). A new `FatalErrorFrame`
-  that sets this flag to true has been added.
-
-- `AnthropicLLMService` now supports function calling and initial support for
-  prompt caching.
-  (see https://www.anthropic.com/news/prompt-caching)
-
-- `ElevenLabsTTSService` can now specify ElevenLabs input parameters such as
-  `output_format`.
-
-- `TwilioFrameSerializer` can now specify Twilio's and Pipecat's desired sample
-  rates to use.
-
-- Added new `on_participant_updated` event to `DailyTransport`.
-
-- Added `DailyRESTHelper.delete_room_by_name()` and
-  `DailyRESTHelper.delete_room_by_url()`.
-
-- Added LLM and TTS usage metrics. Those are enabled when
-  `PipelineParams.enable_usage_metrics` is True.
-
-- `AudioRawFrame`s are now pushed downstream from the base output
-  transport. This allows capturing the exact words the bot says by adding an STT
-  service at the end of the pipeline.
-
-- Added new `GStreamerPipelineSource`. This processor can generate image or
-  audio frames from a GStreamer pipeline (e.g. reading an MP4 file, and RTP
-  stream or anything supported by GStreamer).
-
-- Added `TransportParams.audio_out_is_live`. This flag is False by default and
-  it is useful to indicate we should not synchronize audio with sporadic images.
-
-- Added new `BotStartedSpeakingFrame` and `BotStoppedSpeakingFrame` control
-  frames. These frames are pushed upstream and they should wrap
-  `BotSpeakingFrame`.
-
-- Transports now allow you to register event handlers without decorators.
-
-### Changed
-
-- Support RTVI message protocol 0.1. This includes new messages, support for
-  messages responses, support for actions, configuration, webhooks and a bunch
-  of new cool stuff.
-  (see https://docs.rtvi.ai/)
-
-- `SileroVAD` dependency is now imported via pip's `silero-vad` package.
-
-- `ElevenLabsTTSService` now uses `eleven_turbo_v2_5` model by default.
-
-- `BotSpeakingFrame` is now a control frame.
-
-- `StartFrame` is now a control frame similar to `EndFrame`.
-
-- `DeepgramTTSService` now is more customizable. You can adjust the encoding and
-  sample rate.
-
-### Fixed
-
-- `TTSStartFrame` and `TTSStopFrame` are now sent when TTS really starts and
-  stops. This allows for knowing when the bot starts and stops speaking even
-  with asynchronous services (like Cartesia).
-
-- Fixed `AzureSTTService` transcription frame timestamps.
-
-- Fixed an issue with `DailyRESTHelper.create_room()` expirations which would
-  cause this function to stop working after the initial expiration elapsed.
-
-- Improved `EndFrame` and `CancelFrame` handling. `EndFrame` should end things
-  gracefully while a `CancelFrame` should cancel all running tasks as soon as
-  possible.
-
-- Fixed an issue in `AIService` that would cause a yielded `None` value to be
-  processed.
-
-- RTVI's `bot-ready` message is now sent when the RTVI pipeline is ready and
-  a first participant joins.
-
-- Fixed a `BaseInputTransport` issue that was causing incoming system frames to
-  be queued instead of being pushed immediately.
-
-- Fixed a `BaseInputTransport` issue that was causing start/stop interruptions
-  incoming frames to not cancel tasks and be processed properly.
-
-### Other
-
-- Added `studypal` example (from to the Cartesia folks!).
-
-- Most examples now use Cartesia.
-
-- Added examples `foundational/19a-tools-anthropic.py`,
-  `foundational/19b-tools-video-anthropic.py` and
-  `foundational/19a-tools-togetherai.py`.
-
-- Added examples `foundational/18-gstreamer-filesrc.py` and
-  `foundational/18a-gstreamer-videotestsrc.py` that show how to use
-  `GStreamerPipelineSource`
-
-- Remove `requests` library usage.
-
-- Cleanup examples and use `DailyRESTHelper`.
-
-## [0.0.39] - 2024-07-23
-
-### Fixed
-
-- Fixed a regression introduced in 0.0.38 that would cause Daily transcription
-  to stop the Pipeline.
-
-## [0.0.38] - 2024-07-23
-
-### Added
-
-- Added `force_reload`, `skip_validation` and `trust_repo` to `SileroVAD` and
-  `SileroVADAnalyzer`. This allows caching and various GitHub repo validations.
-
-- Added `send_initial_empty_metrics` flag to `PipelineParams` to request for
-  initial empty metrics (zero values). True by default.
-
-### Fixed
-
-- Fixed initial metrics format. It was using the wrong keys name/time instead of
-  processor/value.
-
-- STT services should be using ISO 8601 time format for transcription frames.
-
-- Fixed an issue that would cause Daily transport to show a stop transcription
-  error when actually none occurred.
-
-## [0.0.37] - 2024-07-22
-
-### Added
-
-- Added `RTVIProcessor` which implements the RTVI-AI standard.
-  See https://github.com/rtvi-ai
-
-- Added `BotInterruptionFrame` which allows interrupting the bot while talking.
-
-- Added `LLMMessagesAppendFrame` which allows appending messages to the current
-  LLM context.
-
-- Added `LLMMessagesUpdateFrame` which allows changing the LLM context for the
-  one provided in this new frame.
-
-- Added `LLMModelUpdateFrame` which allows updating the LLM model.
-
-- Added `TTSSpeakFrame` which causes the bot say some text. This text will not
-  be part of the LLM context.
-
-- Added `TTSVoiceUpdateFrame` which allows updating the TTS voice.
-
-### Removed
-
-- We remove the `LLMResponseStartFrame` and `LLMResponseEndFrame` frames. These
-  were added in the past to properly handle interruptions for the
-  `LLMAssistantContextAggregator`. But the `LLMContextAggregator` is now based
-  on `LLMResponseAggregator` which handles interruptions properly by just
-  processing the `StartInterruptionFrame`, so there's no need for these extra
-  frames any more.
-
-### Fixed
-
-- Fixed an issue with `StatelessTextTransformer` where it was pushing a string
-  instead of a `TextFrame`.
-
-- `TTSService` end of sentence detection has been improved. It now works with
-  acronyms, numbers, hours and others.
-
-- Fixed an issue in `TTSService` that would not properly flush the current
-  aggregated sentence if an `LLMFullResponseEndFrame` was found.
-
-### Performance
-
-- `CartesiaTTSService` now uses websockets which improves speed. It also
-  leverages the new Cartesia contexts which maintains generated audio prosody
-  when multiple inputs are sent, therefore improving audio quality a lot.
-
-## [0.0.36] - 2024-07-02
-
-### Added
-
-- Added `GladiaSTTService`.
-  See https://docs.gladia.io/chapters/speech-to-text-api/pages/live-speech-recognition
-
-- Added `XTTSService`. This is a local Text-To-Speech service.
-  See https://github.com/coqui-ai/TTS
-
-- Added `UserIdleProcessor`. This processor can be used to wait for any
-  interaction with the user. If the user doesn't say anything within a given
-  timeout a provided callback is called.
-
-- Added `IdleFrameProcessor`. This processor can be used to wait for frames
-  within a given timeout. If no frame is received within the timeout a provided
-  callback is called.
-
-- Added new frame `BotSpeakingFrame`. This frame will be continuously pushed
-  upstream while the bot is talking.
-
-- It is now possible to specify a Silero VAD version when using `SileroVADAnalyzer`
-  or `SileroVAD`.
-
-- Added `AysncFrameProcessor` and `AsyncAIService`.  Some services like
-  `DeepgramSTTService` need to process things asynchronously. For example, audio
-  is sent to Deepgram but transcriptions are not returned immediately. In these
-  cases we still require all frames (except system frames) to be pushed
-  downstream from a single task. That's what `AsyncFrameProcessor` is for. It
-  creates a task and all frames should be pushed from that task. So, whenever a
-  new Deepgram transcription is ready that transcription will also be pushed
-  from this internal task.
-
-- The `MetricsFrame` now includes processing metrics if metrics are enabled. The
-  processing metrics indicate the time a processor needs to generate all its
-  output. Note that not all processors generate these kind of metrics.
-
-### Changed
-
-- `WhisperSTTService` model can now also be a string.
-
-- Added missing * keyword separators in services.
-
-### Fixed
-
-- `WebsocketServerTransport` doesn't try to send frames anymore if serializers
-  returns `None`.
-
-- Fixed an issue where exceptions that occurred inside frame processors were
-  being swallowed and not displayed.
-
-- Fixed an issue in `FastAPIWebsocketTransport` where it would still try to send
-  data to the websocket after being closed.
-
-### Other
-
-- Added Fly.io deployment example in `examples/deployment/flyio-example`.
-
-- Added new `17-detect-user-idle.py` example that shows how to use the new
-  `UserIdleProcessor`.
-
-## [0.0.35] - 2024-06-28
-
-### Changed
-
-- `FastAPIWebsocketParams` now require a serializer.
-
-- `TwilioFrameSerializer` now requires a `streamSid`.
-
-### Fixed
-
-- Silero VAD number of frames needs to be 512 for 16000 sample rate or 256 for
-  8000 sample rate.
-
-## [0.0.34] - 2024-06-25
-
-### Fixed
-
-- Fixed an issue with asynchronous STT services (Deepgram and Azure) that could
-  interruptions to ignore transcriptions.
-
-- Fixed an issue introduced in 0.0.33 that would cause the LLM to generate
-  shorter output.
-
-## [0.0.33] - 2024-06-25
-
-### Changed
-
-- Upgraded to Cartesia's new Python library 1.0.0. `CartesiaTTSService` now
-  expects a voice ID instead of a voice name (you can get the voice ID from
-  Cartesia's playground). You can also specify the audio `sample_rate` and
-  `encoding` instead of the previous `output_format`.
-
-### Fixed
-
-- Fixed an issue with asynchronous STT services (Deepgram and Azure) that could
-  cause static audio issues and interruptions to not work properly when dealing
-  with multiple LLMs sentences.
-
-- Fixed an issue that could mix new LLM responses with previous ones when
-  handling interruptions.
-
-- Fixed a Daily transport blocking situation that occurred while reading audio
-  frames after a participant left the room. Needs daily-python >= 0.10.1.
-
 ## [0.0.32] - 2024-06-22
 
 ### Added

README.md

@@ -4,7 +4,8 @@
 
 # Pipecat
 
-[![PyPI](https://img.shields.io/pypi/v/pipecat-ai)](https://pypi.org/project/pipecat-ai) [![Discord](https://img.shields.io/discord/1239284677165056021)](https://discord.gg/pipecat) <a href="https://app.commanddash.io/agent/github_pipecat-ai_pipecat"><img src="https://img.shields.io/badge/AI-Code%20Agent-EB9FDA"></a>
+[![PyPI](https://img.shields.io/pypi/v/pipecat-ai)](https://pypi.org/project/pipecat-ai) [![Discord](https://img.shields.io/discord/1239284677165056021
+)](https://discord.gg/pipecat)
 
 `pipecat` is a framework for building voice (and multimodal) conversational agents. Things like personal coaches, meeting assistants, [story-telling toys for kids](https://storytelling-chatbot.fly.dev/), customer support bots, [intake flows](https://www.youtube.com/watch?v=lDevgsp9vn0), and snarky social companions.
 
@@ -38,7 +39,7 @@ pip install "pipecat-ai[option,...]"
 
 Your project may or may not need these, so they're made available as optional requirements. Here is a list:
 
-- **AI services**: `anthropic`, `azure`, `deepgram`, `gladia`, `google`, `fal`, `lmnt`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`, `xtts`
+- **AI services**: `anthropic`, `azure`, `deepgram`, `google`, `fal`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`
 - **Transports**: `local`, `websocket`, `daily`
 
 ## Code examples
@@ -48,7 +49,7 @@ Your project may or may not need these, so they're made available as optional re
 
 ## A simple voice agent running locally
 
-Here is a very basic Pipecat bot that greets a user when they join a real-time session. We'll use [Daily](https://daily.co) for real-time media transport, and [Cartesia](https://cartesia.ai/) for text-to-speech.
+Here is a very basic Pipecat bot that greets a user when they join a real-time session. We'll use [Daily](https://daily.co) for real-time media transport, and [ElevenLabs](https://elevenlabs.io/) for text-to-speech.
 
 ```python
 #app.py
@@ -60,7 +61,7 @@ from pipecat.frames.frames import EndFrame, TextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.task import PipelineTask
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
 async def main():
@@ -69,13 +70,14 @@ async def main():
     transport = DailyTransport(
       room_url=...,
       token=...,
-      bot_name="Bot Name",
-      params=DailyParams(audio_out_enabled=True))
+      "Bot Name",
+      DailyParams(audio_out_enabled=True))
 
-    # Use Cartesia for Text-to-Speech
-    tts = CartesiaTTSService(
-        api_key=...,
-        voice_id=...
+    # Use Eleven Labs for Text-to-Speech
+    tts = ElevenLabsTTSService(
+      aiohttp_session=session,
+      api_key=...,
+      voice_id=...,
       )
 
     # Simple pipeline that will process text to speech and output the result
@@ -92,7 +94,7 @@ async def main():
     @transport.event_handler("on_participant_joined")
     async def on_new_participant_joined(transport, participant):
       participant_name = participant["info"]["userName"] or ''
-      # Queue a TextFrame that will get spoken by the TTS service (Cartesia)
+      # Queue a TextFrame that will get spoken by the TTS service (Eleven Labs)
       await task.queue_frames([TextFrame(f"Hello there, {participant_name}!"), EndFrame()])
 
     # Run the pipeline task
@@ -123,7 +125,7 @@ Sign up [here](https://dashboard.daily.co/u/signup) and [create a room](https://
 
 Voice Activity Detection &mdash; very important for knowing when a user has finished speaking to your bot. If you are not using press-to-talk, and want Pipecat to detect when the user has finished talking, VAD is an essential component for a natural feeling conversation.
 
-Pipecat makes use of WebRTC VAD by default when using a WebRTC transport layer. Optionally, you can use Silero VAD for improved accuracy at the cost of higher CPU usage.
+Pipecast makes use of WebRTC VAD by default when using a WebRTC transport layer. Optionally, you can use Silero VAD for improved accuracy at the cost of higher CPU usage.
 
 ```shell
 pip install pipecat-ai[silero]
@@ -144,20 +146,20 @@ source venv/bin/activate
 From the root of this repo, run the following:
 
 ```shell
-pip install -r dev-requirements.txt
+pip install -r dev-requirements.txt -r {env}-requirements.txt
 python -m build
 ```
 
-This builds the package. To use the package locally (e.g. to run sample files), run
+This builds the package. To use the package locally (eg to run sample files), run
 
 ```shell
-pip install --editable ".[option,...]"
+pip install --editable .
 ```
 
 If you want to use this package from another directory, you can run:
 
 ```shell
-pip install "path_to_this_repo[option,...]"
+pip install path_to_this_repo
 ```
 
 ### Running tests

dev-requirements.txt

@@ -1,8 +1,8 @@
-autopep8~=2.3.1
+autopep8~=2.1.0
 build~=1.2.1
 grpcio-tools~=1.62.2
 pip-tools~=7.4.1
-pyright~=1.1.376
-pytest~=8.3.2
-setuptools~=72.2.0
+pyright~=1.1.367
+pytest~=8.2.0
+setuptools~=69.5.1
 setuptools_scm~=8.1.0

dot-env.template

@@ -27,13 +27,6 @@ FAL_KEY=...
 # Fireworks
 FIREWORKS_API_KEY=...
 
-# Gladia
-GLADIA_API_KEY=...
-
-# LMNT
-LMNT_API_KEY=...
-LMNT_VOICE_ID=...
-
 # PlayHT
 PLAY_HT_USER_ID=...
 PLAY_HT_API_KEY=...

examples/README.md

@@ -41,7 +41,6 @@ Next, follow the steps in the README for each demo.
 | [Patient intake](patient-intake)             | A chatbot that can call functions in response to user input.                                                                               | Deepgram, ElevenLabs, OpenAI, Daily, Daily Prebuilt UI            |
 | [Dialin Chatbot](dialin-chatbot)             | A chatbot that connects to an incoming phone call from Daily or Twilio.                                                                    | Deepgram, ElevenLabs, OpenAI, Daily, Twilio                       |
 | [Twilio Chatbot](twilio-chatbot)             | A chatbot that connects to an incoming phone call from Twilio.                                                                             | Deepgram, ElevenLabs, OpenAI, Daily, Twilio                       |
-| [studypal](studypal)                         | A chatbot to have a conversation about any article on the web                                                                              |                                                                   |
 
 > [!IMPORTANT]
 > These example projects use Daily as a WebRTC transport and can be joined using their hosted Prebuilt UI.

examples/deployment/flyio-example/Dockerfile

@@ -1,13 +0,0 @@
-FROM python:3.11-bullseye
-
-# Open port 7860 for http service
-ENV FAST_API_PORT=7860
-EXPOSE 7860
-
-# Install Python dependencies
-COPY *.py .
-COPY ./requirements.txt requirements.txt
-RUN pip3 install --no-cache-dir --upgrade -r requirements.txt
-
-# Start the FastAPI server
-CMD python3 bot_runner.py --port ${FAST_API_PORT}

examples/deployment/flyio-example/README.md

@@ -1,39 +0,0 @@
-# Fly.io deployment example
-
-This project modifies the `bot_runner.py` server to launch a new machine for each user session. This is a recommended approach for production vs. running shell processess as your deployment will quickly run out of system resources under load.
-
-For this example, we are using Daily as a WebRTC transport and provisioning a new room and token for each session. You can use another transport, such as WebSockets, by modifying the `bot.py` and `bot_runner.py` files accordingly.
-
-## Setting up your fly.io deployment
-
-### Create your fly.toml file
-
-You can copy the `example-fly.toml` as a reference. Be sure to change the app name to something unique.
-
-### Create your .env file
-
-Copy the base `env.example` to `.env` and enter the necessary API keys.
-
-`FLY_APP_NAME` should match that in the `fly.toml` file.
-
-### Launch a new fly.io project
-
-`fly launch` or `fly launch --org your-org-name`
-
-### Set the necessary app secrets from your .env
-
-Note: you can do this manually via the fly.io dashboard under the "secrets" sub-section of your deployment (e.g. "https://fly.io/apps/fly-app-name/secrets") or run the following terminal command:
-
-`cat .env | tr '\n' ' ' | xargs flyctl secrets set`
-
-### Deploy your machine
-
-`fly deploy`
-
-## Connecting to your bot
-
-Send a post request to your running fly.io instance:
-
-`curl --location --request POST 'https://YOUR_FLY_APP_NAME/start_bot'`
-
-This request will wait until the machine enters into a `starting` state, before returning the a room URL and token to join.

examples/deployment/flyio-example/bot.py

@@ -1,100 +0,0 @@
-import asyncio
-import os
-import sys
-import argparse
-
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.llm_response import LLMAssistantResponseAggregator, LLMUserResponseAggregator
-from pipecat.frames.frames import LLMMessagesFrame, EndFrame
-from pipecat.services.openai import OpenAILLMService
-from pipecat.services.elevenlabs import ElevenLabsTTSService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.vad.silero import SileroVADAnalyzer
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-daily_api_key = os.getenv("DAILY_API_KEY", "")
-daily_api_url = os.getenv("DAILY_API_URL", "https://api.daily.co/v1")
-
-
-async def main(room_url: str, token: str):
-    transport = DailyTransport(
-        room_url,
-        token,
-        "Chatbot",
-        DailyParams(
-            api_url=daily_api_url,
-            api_key=daily_api_key,
-            audio_in_enabled=True,
-            audio_out_enabled=True,
-            camera_out_enabled=False,
-            vad_enabled=True,
-            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"),
-        model="gpt-4o")
-
-    messages = [
-        {
-            "role": "system",
-            "content": "You are Chatbot, a friendly, helpful robot. Your output will be converted to audio so don't include special characters other than '!' or '?' in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by saying hello.",
-        },
-    ]
-
-    tma_in = LLMUserResponseAggregator(messages)
-    tma_out = LLMAssistantResponseAggregator(messages)
-
-    pipeline = Pipeline([
-        transport.input(),
-        tma_in,
-        llm,
-        tts,
-        transport.output(),
-        tma_out,
-    ])
-
-    task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
-
-    @transport.event_handler("on_first_participant_joined")
-    async def on_first_participant_joined(transport, participant):
-        transport.capture_participant_transcription(participant["id"])
-        await task.queue_frames([LLMMessagesFrame(messages)])
-
-    @transport.event_handler("on_participant_left")
-    async def on_participant_left(transport, participant, reason):
-        await task.queue_frame(EndFrame())
-
-    @transport.event_handler("on_call_state_updated")
-    async def on_call_state_updated(transport, state):
-        if state == "left":
-            await task.queue_frame(EndFrame())
-
-    runner = PipelineRunner()
-
-    await runner.run(task)
-
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description="Pipecat Bot")
-    parser.add_argument("-u", type=str, help="Room URL")
-    parser.add_argument("-t", type=str, help="Token")
-    config = parser.parse_args()
-
-    asyncio.run(main(config.u, config.t))

examples/deployment/flyio-example/bot_runner.py

@@ -1,215 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
-import argparse
-import subprocess
-import os
-
-from contextlib import asynccontextmanager
-
-from fastapi import FastAPI, Request, HTTPException
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse
-
-from pipecat.transports.services.helpers.daily_rest import (
-    DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomParams)
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-
-# ------------ Configuration ------------ #
-
-MAX_SESSION_TIME = 5 * 60  # 5 minutes
-REQUIRED_ENV_VARS = [
-    'DAILY_API_KEY',
-    'OPENAI_API_KEY',
-    'ELEVENLABS_API_KEY',
-    'ELEVENLABS_VOICE_ID',
-    'FLY_API_KEY',
-    'FLY_APP_NAME',]
-
-FLY_API_HOST = os.getenv("FLY_API_HOST", "https://api.machines.dev/v1")
-FLY_APP_NAME = os.getenv("FLY_APP_NAME", "pipecat-fly-example")
-FLY_API_KEY = os.getenv("FLY_API_KEY", "")
-FLY_HEADERS = {
-    'Authorization': f"Bearer {FLY_API_KEY}",
-    'Content-Type': 'application/json'
-}
-
-daily_helpers = {}
-
-
-# ----------------- API ----------------- #
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    aiohttp_session = aiohttp.ClientSession()
-    daily_helpers["rest"] = DailyRESTHelper(
-        daily_api_key=os.getenv("DAILY_API_KEY", ""),
-        daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
-        aiohttp_session=aiohttp_session
-    )
-    yield
-    await aiohttp_session.close()
-
-app = FastAPI(lifespan=lifespan)
-
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"]
-)
-
-# ----------------- Main ----------------- #
-
-
-async def spawn_fly_machine(room_url: str, token: str):
-    async with aiohttp.ClientSession() as session:
-        # Use the same image as the bot runner
-        async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS) as r:
-            if r.status != 200:
-                text = await r.text()
-                raise Exception(f"Unable to get machine info from Fly: {text}")
-
-            data = await r.json()
-            image = data[0]['config']['image']
-
-        # Machine configuration
-        cmd = f"python3 bot.py -u {room_url} -t {token}"
-        cmd = cmd.split()
-        worker_props = {
-            "config": {
-                "image": image,
-                "auto_destroy": True,
-                "init": {
-                    "cmd": cmd
-                },
-                "restart": {
-                    "policy": "no"
-                },
-                "guest": {
-                    "cpu_kind": "shared",
-                    "cpus": 1,
-                    "memory_mb": 1024
-                }
-            },
-        }
-
-        # Spawn a new machine instance
-        async with session.post(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS, json=worker_props) as r:
-            if r.status != 200:
-                text = await r.text()
-                raise Exception(f"Problem starting a bot worker: {text}")
-
-            data = await r.json()
-            # Wait for the machine to enter the started state
-            vm_id = data['id']
-
-        async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines/{vm_id}/wait?state=started", headers=FLY_HEADERS) as r:
-            if r.status != 200:
-                text = await r.text()
-                raise Exception(f"Bot was unable to enter started state: {text}")
-
-    print(f"Machine joined room: {room_url}")
-
-
-@app.post("/start_bot")
-async def start_bot(request: Request) -> JSONResponse:
-    try:
-        data = await request.json()
-        # Is this a webhook creation request?
-        if "test" in data:
-            return JSONResponse({"test": True})
-    except Exception as e:
-        pass
-
-    # Use specified room URL, or create a new one if not specified
-    room_url = os.getenv("DAILY_SAMPLE_ROOM_URL", "")
-
-    if not room_url:
-        params = DailyRoomParams(
-            properties=DailyRoomProperties()
-        )
-        try:
-            room: DailyRoomObject = await daily_helpers["rest"].create_room(params=params)
-        except Exception as e:
-            raise HTTPException(
-                status_code=500,
-                detail=f"Unable to provision room {e}")
-    else:
-        # Check passed room URL exists, we should assume that it already has a sip set up
-        try:
-            room: DailyRoomObject = await daily_helpers["rest"].get_room_from_url(room_url)
-        except Exception:
-            raise HTTPException(
-                status_code=500, detail=f"Room not found: {room_url}")
-
-    # Give the agent a token to join the session
-    token = await daily_helpers["rest"].get_token(room.url, MAX_SESSION_TIME)
-
-    if not room or not token:
-        raise HTTPException(
-            status_code=500, detail=f"Failed to get token for room: {room_url}")
-
-    # Launch a new fly.io machine, or run as a shell process (not recommended)
-    run_as_process = os.getenv("RUN_AS_PROCESS", False)
-
-    if run_as_process:
-        try:
-            subprocess.Popen(
-                [f"python3 -m bot -u {room.url} -t {token}"],
-                shell=True,
-                bufsize=1,
-                cwd=os.path.dirname(os.path.abspath(__file__)))
-        except Exception as e:
-            raise HTTPException(
-                status_code=500, detail=f"Failed to start subprocess: {e}")
-    else:
-        try:
-            await spawn_fly_machine(room.url, token)
-        except Exception as e:
-            raise HTTPException(
-                status_code=500, detail=f"Failed to spawn VM: {e}")
-
-    # Grab a token for the user to join with
-    user_token = await daily_helpers["rest"].get_token(room.url, MAX_SESSION_TIME)
-
-    return JSONResponse({
-        "room_url": room.url,
-        "token": user_token,
-    })
-
-if __name__ == "__main__":
-    # Check environment variables
-    for env_var in REQUIRED_ENV_VARS:
-        if env_var not in os.environ:
-            raise Exception(f"Missing environment variable: {env_var}.")
-
-    parser = argparse.ArgumentParser(description="Pipecat Bot Runner")
-    parser.add_argument("--host", type=str,
-                        default=os.getenv("HOST", "0.0.0.0"), help="Host address")
-    parser.add_argument("--port", type=int,
-                        default=os.getenv("PORT", 7860), help="Port number")
-    parser.add_argument("--reload", action="store_true",
-                        default=False, help="Reload code on change")
-
-    config = parser.parse_args()
-
-    try:
-        import uvicorn
-
-        uvicorn.run(
-            "bot_runner:app",
-            host=config.host,
-            port=config.port,
-            reload=config.reload
-        )
-    except KeyboardInterrupt:
-        print("Pipecat runner shutting down...")

examples/deployment/flyio-example/env.example

@@ -1,8 +0,0 @@
-DAILY_API_KEY=
-DAILY_SAMPLE_ROOM_URL= # Enter a Daily room URL to use a set room URL each time (useful for local testing)
-OPENAI_API_KEY=
-ELEVENLABS_API_KEY=
-ELEVENLABS_VOICE_ID=
-FLY_API_KEY=
-FLY_APP_NAME=
-RUN_AS_PROCESS= # Spawn fly.io machine for each session or run as local process

examples/deployment/flyio-example/example-fly.toml

@@ -1,25 +0,0 @@
-# fly.toml app configuration file generated for pipecat-fly-example on 2024-07-01T15:04:53+01:00
-#
-# See https://fly.io/docs/reference/configuration/ for information about how to use this file.
-#
-
-app = 'pipecat-fly-example'
-primary_region = 'sjc'
-
-[build]
-
-[env]
-  FLY_APP_NAME = 'pipecat-fly-example'
-
-[http_service]
-  internal_port = 7860
-  force_https = true
-  auto_stop_machines = true
-  auto_start_machines = true
-  min_machines_running = 0
-  processes = ['app']
-
-[[vm]]
-  memory = 512
-  cpu_kind = 'shared'
-  cpus = 1

examples/deployment/flyio-example/requirements.txt

@@ -1,5 +0,0 @@
-pipecat-ai[daily,openai,silero]
-fastapi
-uvicorn
-python-dotenv
-loguru

examples/dialin-chatbot/bot_daily.py

@@ -1,4 +1,5 @@
 import asyncio
+import aiohttp
 import os
 import sys
 import argparse
@@ -28,74 +29,75 @@ daily_api_url = os.getenv("DAILY_API_URL", "https://api.daily.co/v1")
 
 
 async def main(room_url: str, token: str, callId: str, callDomain: str):
-    # 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 = DailyDialinSettings(
-        call_id=callId,
-        call_domain=callDomain
-    )
-
-    transport = DailyTransport(
-        room_url,
-        token,
-        "Chatbot",
-        DailyParams(
-            api_url=daily_api_url,
-            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,
+    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 = DailyDialinSettings(
+            call_id=callId,
+            call_domain=callDomain
         )
-    )
 
-    tts = ElevenLabsTTSService(
-        api_key=os.getenv("ELEVENLABS_API_KEY", ""),
-        voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
-    )
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Chatbot",
+            DailyParams(
+                api_url=daily_api_url,
+                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,
+            )
+        )
 
-    llm = OpenAILLMService(
-        api_key=os.getenv("OPENAI_API_KEY"),
-        model="gpt-4o"
-    )
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY", ""),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
+        )
 
-    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 'Oh, hello! Who dares dial me at this hour?!'.",
-        },
-    ]
+        llm = OpenAILLMService(
+            api_key=os.getenv("OPENAI_API_KEY"),
+            model="gpt-4o")
 
-    tma_in = LLMUserResponseAggregator(messages)
-    tma_out = LLMAssistantResponseAggregator(messages)
+        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 'Oh, hello! Who dares dial me at this hour?!'.",
+            },
+        ]
 
-    pipeline = Pipeline([
-        transport.input(),
-        tma_in,
-        llm,
-        tts,
-        transport.output(),
-        tma_out,
-    ])
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
-    task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+        pipeline = Pipeline([
+            transport.input(),
+            tma_in,
+            llm,
+            tts,
+            transport.output(),
+            tma_out,
+        ])
 
-    @transport.event_handler("on_first_participant_joined")
-    async def on_first_participant_joined(transport, participant):
-        transport.capture_participant_transcription(participant["id"])
-        await task.queue_frames([LLMMessagesFrame(messages)])
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
 
-    @transport.event_handler("on_participant_left")
-    async def on_participant_left(transport, participant, reason):
-        await task.queue_frame(EndFrame())
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            transport.capture_participant_transcription(participant["id"])
+            await task.queue_frames([LLMMessagesFrame(messages)])
 
-    runner = PipelineRunner()
+        @transport.event_handler("on_participant_left")
+        async def on_participant_left(transport, participant, reason):
+            await task.queue_frame(EndFrame())
 
-    await runner.run(task)
+        runner = PipelineRunner()
+
+        await runner.run(task)
 
 
 if __name__ == "__main__":

examples/dialin-chatbot/bot_runner.py

@@ -6,27 +6,15 @@ provisioning a room and starting a Pipecat bot in response.
 
 Refer to README for more information.
 """
-
-
-import aiohttp
 import os
 import argparse
 import subprocess
-
-from contextlib import asynccontextmanager
-
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomSipParams, DailyRoomParams
 from fastapi import FastAPI, Request, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse, PlainTextResponse
 from twilio.twiml.voice_response import VoiceResponse
 
-from pipecat.transports.services.helpers.daily_rest import (
-    DailyRESTHelper,
-    DailyRoomObject,
-    DailyRoomProperties,
-    DailyRoomSipParams,
-    DailyRoomParams)
-
 from dotenv import load_dotenv
 load_dotenv(override=True)
 
@@ -37,23 +25,14 @@ MAX_SESSION_TIME = 5 * 60  # 5 minutes
 REQUIRED_ENV_VARS = ['OPENAI_API_KEY', 'DAILY_API_KEY',
                      'ELEVENLABS_API_KEY', 'ELEVENLABS_VOICE_ID']
 
-daily_helpers = {}
+daily_rest_helper = DailyRESTHelper(
+    os.getenv("DAILY_API_KEY", ""),
+    os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'))
+
 
 # ----------------- API ----------------- #
 
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    aiohttp_session = aiohttp.ClientSession()
-    daily_helpers["rest"] = DailyRESTHelper(
-        daily_api_key=os.getenv("DAILY_API_KEY", ""),
-        daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
-        aiohttp_session=aiohttp_session
-    )
-    yield
-    await aiohttp_session.close()
-
-app = FastAPI(lifespan=lifespan)
+app = FastAPI()
 
 app.add_middleware(
     CORSMiddleware,
@@ -74,7 +53,7 @@ action using the Twilio Client library.
 """
 
 
-async def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
+def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
     if not room_url:
         params = DailyRoomParams(
             properties=DailyRoomProperties(
@@ -89,13 +68,14 @@ async def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
         )
 
         print(f"Creating new room...")
-        room: DailyRoomObject = await daily_helpers["rest"].create_room(params=params)
+        room: DailyRoomObject = daily_rest_helper.create_room(params=params)
 
     else:
         # Check passed room URL exist (we assume that it already has a sip set up!)
         try:
             print(f"Joining existing room: {room_url}")
-            room: DailyRoomObject = await daily_helpers["rest"].get_room_from_url(room_url)
+            room: DailyRoomObject = daily_rest_helper.get_room_from_url(
+                room_url)
         except Exception:
             raise HTTPException(
                 status_code=500, detail=f"Room not found: {room_url}")
@@ -103,7 +83,7 @@ async def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
     print(f"Daily room: {room.url} {room.config.sip_endpoint}")
 
     # Give the agent a token to join the session
-    token = await daily_helpers["rest"].get_token(room.url, MAX_SESSION_TIME)
+    token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
 
     if not room or not token:
         raise HTTPException(
@@ -112,11 +92,11 @@ async def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
     # Spawn a new agent, and join the user session
     # Note: this is mostly for demonstration purposes (refer to 'deployment' in docs)
     if vendor == "daily":
-        bot_proc = f"python3 - m bot_daily - u {room.url} - t {token} - i {
-            callId} - d {callDomain}"
+        bot_proc = f"python3 -m bot_daily -u {room.url} -t {token} -i {
+            callId} -d {callDomain}"
     else:
-        bot_proc = f"python3 - m bot_twilio - u {room.url} - t {
-            token} - i {callId} - s {room.config.sip_endpoint}"
+        bot_proc = f"python3 -m bot_twilio -u {room.url} -t {
+            token} -i {callId} -s {room.config.sip_endpoint}"
 
     try:
         subprocess.Popen(
@@ -160,7 +140,8 @@ async def twilio_start_bot(request: Request):
 
     # create room and tell the bot to join the created room
     # note: Twilio does not require a callDomain
-    room: DailyRoomObject = await _create_daily_room(room_url, callId, None, "twilio")
+    room: DailyRoomObject = _create_daily_room(
+        room_url, callId, None, "twilio")
 
     print(f"Put Twilio on hold...")
     # We have the room and the SIP URI,
@@ -197,7 +178,8 @@ async def daily_start_bot(request: Request) -> JSONResponse:
             detail="Missing properties 'callId' or 'callDomain'")
 
     print(f"CallId: {callId}, CallDomain: {callDomain}")
-    room: DailyRoomObject = await _create_daily_room(room_url, callId, callDomain, "daily")
+    room: DailyRoomObject = _create_daily_room(
+        room_url, callId, callDomain, "daily")
 
     # Grab a token for the user to join with
     return JSONResponse({

examples/dialin-chatbot/bot_twilio.py

@@ -1,4 +1,5 @@
 import asyncio
+import aiohttp
 import os
 import sys
 import argparse
@@ -35,81 +36,82 @@ 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,
-            camera_out_enabled=False,
-            vad_enabled=True,
-            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"),
-        model="gpt-4o"
-    )
-
-    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?!'.",
-        },
-    ]
-
-    tma_in = LLMUserResponseAggregator(messages)
-    tma_out = LLMAssistantResponseAggregator(messages)
-
-    pipeline = Pipeline([
-        transport.input(),
-        tma_in,
-        llm,
-        tts,
-        transport.output(),
-        tma_out,
-    ])
-
-    task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
-
-    @transport.event_handler("on_first_participant_joined")
-    async def on_first_participant_joined(transport, participant):
-        transport.capture_participant_transcription(participant["id"])
-        await task.queue_frames([LLMMessagesFrame(messages)])
-
-    @transport.event_handler("on_participant_left")
-    async def on_participant_left(transport, participant, reason):
-        await task.queue_frame(EndFrame())
-
-    @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>'
+    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.
+        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,
+                camera_out_enabled=False,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                transcription_enabled=True,
             )
-        except Exception as e:
-            raise Exception(f"Failed to forward call: {str(e)}")
+        )
 
-    runner = PipelineRunner()
-    await runner.run(task)
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY", ""),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
+        )
+
+        llm = OpenAILLMService(
+            api_key=os.getenv("OPENAI_API_KEY"),
+            model="gpt-4o")
+
+        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?!'.",
+            },
+        ]
+
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
+
+        pipeline = Pipeline([
+            transport.input(),
+            tma_in,
+            llm,
+            tts,
+            transport.output(),
+            tma_out,
+        ])
+
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            transport.capture_participant_transcription(participant["id"])
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        @transport.event_handler("on_participant_left")
+        async def on_participant_left(transport, participant, reason):
+            await task.queue_frame(EndFrame())
+
+        @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__":

examples/dialin-chatbot/requirements.txt

@@ -1,6 +1,7 @@
 pipecat-ai[daily,openai,silero]
 fastapi
 uvicorn
+requests
 python-dotenv
-twilio
-python-multipart
+loguru
+twilio

examples/fast-chatbot/.gitignore

@@ -0,0 +1,165 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+runpod.toml
+
+# custom script to recursively upgrade items in requirements.py
+upgrade_requirements.py
+.DS_Store

examples/fast-chatbot/bot.py

@@ -0,0 +1,164 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from loguru import logger
+import argparse
+import asyncio
+import aiohttp
+import os
+import sys
+import time
+from typing import Optional
+
+from pydantic import BaseModel, ValidationError
+
+from pipecat.vad.vad_analyzer import VADParams
+from pipecat.vad.silero import SileroVADAnalyzer
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.services.openai import OpenAILLMService
+from pipecat.services.deepgram import DeepgramSTTService
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.frames.frames import LLMMessagesFrame, EndFrame
+
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator, LLMUserResponseAggregator
+)
+
+from helpers import (
+    ClearableDeepgramTTSService,
+    AudioVolumeTimer,
+    TranscriptionTimingLogger
+)
+
+
+from dotenv import load_dotenv
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level=os.getenv("LOG_LEVEL", "DEBUG"))
+
+
+class BotSettings(BaseModel):
+    room_url: str
+    room_token: str
+    bot_name: str = "Pipecat"
+    prompt: Optional[str] = "You are a helpful assistant."
+    deepgram_api_key: Optional[str] = os.getenv("DEEPGRAM_API_KEY", None)
+    deepgram_voice: Optional[str] = os.getenv("DEEPGRAM_VOICE", "aura-asteria-en")
+    deepgram_tts_base_url: Optional[str] = os.getenv(
+        "DEEPGRAM_TTS_BASE_URL", "https://api.deepgram.com/v1/speak")
+    deepgram_stt_base_url: Optional[str] = os.getenv(
+        "DEEPGRAM_STT_BASE_URL", "https://api.deepgram.com/v1/speak")
+    openai_api_key: Optional[str] = os.getenv("OPENAI_API_KEY", None),
+    openai_model: Optional[str] = os.getenv("OPENAI_MODEL", None),
+    openai_base_url: Optional[str] = os.getenv("OPENAI_BASE_URL", None)
+    vad_stop_secs: Optional[float] = os.getenv("VAD_STOP_SECS", 0.200)
+
+
+async def main(settings: BotSettings):
+    async with aiohttp.ClientSession() as session:
+        transport = DailyTransport(
+            settings.room_url,
+            settings.room_token,
+            settings.bot_name,
+            DailyParams(
+                audio_out_enabled=True,
+                transcription_enabled=False,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(params=VADParams(
+                    stop_secs=settings.vad_stop_secs
+                )),
+                vad_audio_passthrough=True
+            )
+        )
+
+        stt = DeepgramSTTService(
+            name="STT",
+            api_key=settings.deepgram_api_key,
+            url=settings.deepgram_stt_base_url
+        )
+
+        tts = ClearableDeepgramTTSService(
+            name="Voice",
+            aiohttp_session=session,
+            api_key=settings.deepgram_api_key,
+            voice=settings.deepgram_voice,
+            **({'base_url': url} if (url := settings.deepgram_tts_base_url) else {})
+        )
+
+        llm = OpenAILLMService(
+            name="LLM",
+            api_key=settings.openai_api_key,
+            model=settings.openai_model,
+            base_url=settings.openai_base_url,
+        )
+
+        messages = [
+            {
+                "role": "system",
+                "content": settings.prompt,
+            },
+        ]
+
+        avt = AudioVolumeTimer()
+        tl = TranscriptionTimingLogger(avt)
+
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
+
+        pipeline = Pipeline([
+            transport.input(),   # Transport user input
+            avt,                 # Audio volume timer
+            stt,                 # Speech-to-text
+            tl,                  # Transcription timing logger
+            tma_in,              # User responses
+            llm,                 # LLM
+            tts,                 # TTS
+            transport.output(),  # Transport bot output
+            tma_out,             # Assistant spoken responses
+        ])
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                report_only_initial_ttfb=True
+            ))
+
+        # When the participant leaves, we exit the bot.
+        @transport.event_handler("on_participant_left")
+        async def on_participant_left(transport, participant, reason):
+            await task.queue_frame(EndFrame())
+
+        # When the first participant joins, the bot should introduce itself.
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            # Provide some air whilst tracks subscribe
+            time.sleep(2)
+            messages.append(
+                {
+                    "role": "system",
+                    "content": "Briefly introduce yourself by saying 'hello, I'm FastBot, how can I help you today?'"})
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        runner = PipelineRunner()
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Pipecat Bot")
+    parser.add_argument("-s", "--settings", type=str, required=True, help="Pipecat bot settings")
+
+    args, unknown = parser.parse_known_args()
+
+    try:
+        settings = BotSettings.model_validate_json(args.settings)
+        asyncio.run(main(settings))
+    except ValidationError as e:
+        print(e)

examples/fast-chatbot/bot_runner.py

@@ -0,0 +1,164 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+import argparse
+import subprocess
+
+from pydantic import BaseModel, ValidationError
+from typing import Optional
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomParams
+
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+
+from bot import BotSettings
+
+from dotenv import load_dotenv
+load_dotenv(override=True)
+
+
+# ------------ Configuration ------------ #
+
+MAX_SESSION_TIME = 5 * 60  # 5 minutes
+REQUIRED_ENV_VARS = ['DAILY_API_URL', 'DAILY_API_KEY', 'DEEPGRAM_API_KEY']
+
+daily_rest_helper = DailyRESTHelper(
+    os.getenv("DAILY_API_KEY", ""),
+    os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'))
+
+
+class RunnerSettings(BaseModel):
+    prompt: Optional[
+        str] = "You are a fast, low-latency chatbot. Your goal is to demonstrate voice-driven AI capabilities at human-like speeds. When introducing yourself briefly mention your goal is to showcase speed and conversational flow. The technology powering you is Daily for transport, Cerebrium for GPU hosting, Llama 3 (8-B version) LLM, and Deepgram for speech-to-text and text-to-speech. You are hosted on the east coast of the United States. Respond to what the user said in a creative and helpful way, but keep responses short and legible. Ensure responses contain only words. Check again that you have not included special characters other than '?' or '!'."
+    deepgram_voice: Optional[str] = os.getenv("DEEPGRAM_VOICE")
+    openai_model: Optional[str] = os.getenv("OPENAI_MODEL", "gpt-4o")
+    openai_api_key: Optional[str] = os.getenv("OPENAI_API_KEY")
+    test: Optional[bool] = None
+
+# ----------------- API ----------------- #
+
+
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"]
+)
+
+# ----------------- Main ----------------- #
+
+
+@app.post("/start_bot")
+async def start_bot(request: Request) -> JSONResponse:
+    runner_settings = RunnerSettings()
+    try:
+        request_body = await request.body()
+        if len(request_body) > 0:
+            runner_settings = RunnerSettings.model_validate_json(request_body)
+    except ValidationError as e:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid request: {e}")
+    except Exception as e:
+        # If no data in request, pass
+        pass
+
+    # Is this a webhook creation request?
+    if runner_settings.test is not None:
+        return JSONResponse({"test": True})
+
+    # Use specified room URL, or create a new one if not specified
+    room_url = os.getenv("DAILY_SAMPLE_ROOM_URL", "")
+
+    if not room_url:
+        params = DailyRoomParams(
+            properties=DailyRoomProperties()
+        )
+        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}")
+    else:
+        # Check passed room URL exists, we should assume that it already has a sip set up
+        try:
+            room: DailyRoomObject = daily_rest_helper.get_room_from_url(room_url)
+        except Exception:
+            raise HTTPException(
+                status_code=500, detail=f"Room not found: {room_url}")
+
+    # Give the agent a token to join the session
+    token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
+
+    if not room or not token:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to get token for room: {room_url}")
+
+    # Spawn a new agent, and join the user session
+    try:
+        bot_settings = BotSettings(
+            room_url=room.url,
+            room_token=token,
+            prompt=runner_settings.prompt,
+            deepgram_voice=runner_settings.deepgram_voice,
+            openai_model=runner_settings.openai_model,
+            openai_api_key=runner_settings.openai_api_key,
+        )
+        bot_settings_str = bot_settings.model_dump_json(exclude_none=True)
+
+        subprocess.Popen(
+            [f"python3 -m bot -s '{bot_settings_str}'"],
+            shell=True,
+            bufsize=1,
+            cwd=os.path.dirname(os.path.abspath(__file__)))
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to start subprocess: {e}")
+
+    # Grab a token for the user to join with
+    user_token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
+
+    return JSONResponse({
+        "room_url": room.url,
+        "token": user_token,
+    })
+
+
+if __name__ == "__main__":
+    # Check environment variables
+    for env_var in REQUIRED_ENV_VARS:
+        if env_var not in os.environ:
+            raise Exception(f"Missing environment variable: {env_var}.")
+
+    parser = argparse.ArgumentParser(description="Pipecat Bot Runner")
+    parser.add_argument("--host", type=str,
+                        default=os.getenv("HOST", "0.0.0.0"), help="Host address")
+    parser.add_argument("--port", type=int,
+                        default=os.getenv("PORT", 7860), help="Port number")
+    parser.add_argument("--reload", action="store_true",
+                        default=True, help="Reload code on change")
+
+    config = parser.parse_args()
+
+    try:
+        import uvicorn
+
+        uvicorn.run(
+            "bot_runner:app",
+            host=config.host,
+            port=config.port,
+            reload=config.reload
+        )
+
+    except KeyboardInterrupt:
+        print("Pipecat runner shutting down...")

examples/fast-chatbot/env.example

@@ -0,0 +1,12 @@
+DAILY_SAMPLE_ROOM_URL= #optional: use the same room each time, or create a new one if unset
+DAILY_API_KEY=
+DAILY_API_URL=
+
+DEEPGRAM_API_KEY=
+DEEPGRAM_VOICE=
+DEEPGRAM_STT_URL=
+DEEPGRAM_TTS_BASE_URL=
+
+OPENAI_API_KEY=
+OPENAI_MODEL=
+OPENAI_BASE_URL=

examples/fast-chatbot/helpers.py

@@ -0,0 +1,267 @@
+from loguru import logger
+import asyncio
+import math
+import struct
+import time
+from dataclasses import dataclass, field
+from typing import List
+
+
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.frames.frames import (
+    Frame,
+    AudioRawFrame,
+    InterimTranscriptionFrame,
+    TranscriptionFrame,
+    TextFrame,
+    StartInterruptionFrame,
+    LLMFullResponseStartFrame,
+    TTSStoppedFrame,
+    MetricsFrame
+)
+
+from pipecat.vad.vad_analyzer import VADAnalyzer, VADState
+from pipecat.services.deepgram import DeepgramTTSService
+from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame
+
+
+class GreedyLLMAggregator(FrameProcessor):
+    def __init__(self, context: OpenAILLMContext = None, **kwargs):
+        super().__init__(**kwargs)
+        self.context: OpenAILLMContext = context if context else OpenAILLMContext()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        logger.debug(f"{frame}")
+
+        try:
+            if isinstance(frame, InterimTranscriptionFrame):
+                return
+
+            if isinstance(frame, TranscriptionFrame):
+                # append transcribed text to last "user" frame
+                if self.context.messages and self.context.messages[-1]["role"] == "user":
+                    last_frame = self.context.messages.pop()
+                else:
+                    last_frame = {"role": "user", "content": ""}
+
+                last_frame["content"] += " " + frame.text
+                self.context.messages.append(last_frame)
+
+                oai_context_frame = OpenAILLMContextFrame(context=self.context)
+                logger.debug(f"pushing frame {oai_context_frame}")
+                await self.push_frame(oai_context_frame)
+                return
+
+            await self.push_frame(frame, direction)
+        except Exception as e:
+            logger.debug(f"error: {e}")
+
+
+class ClearableDeepgramTTSService(DeepgramTTSService):
+    def __init___(self, **kwargs):
+        super().__init(**kwargs)
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, StartInterruptionFrame):
+            self._current_sentence = ""
+
+
+@dataclass
+class BufferedSentence:
+    audio_frames: List[AudioRawFrame] = field(default_factory=list)
+    text_frame: TextFrame = None
+
+
+class VADGate(FrameProcessor):
+
+    def __init__(
+            self,
+            vad_analyzer: VADAnalyzer = None,
+            context: OpenAILLMContext = None,
+            **kwargs):
+        super().__init__(**kwargs)
+        self.vad_analyzer = vad_analyzer
+        self.context = context
+
+        self._audio_pusher_task = None
+        self._expect_text_frame_next = False
+        self._sentences: List[BufferedSentence] = []
+
+    # queue output from tts one sentence at a time. associate a buffer of audio frames with the content of
+    # each text frame.
+    #
+    # start a coroutine to service the queue and send sentences down the pipeline when possible.
+    # 1. do not send anything when we are not in VADState.QUIET
+    # 2. if we are in VADState.QUIET, send a sentence, estimate how long it will take for that sentence
+    #    to output, sleep until it's time to send another sentence
+    # 3. each time we send a sentence, append it to the conversation context
+    # 3. when the sentence buffer becomes empty, cancel the coroutine
+    # 4. if we get a new LLMFullResponse, treat that as a cancellation, too
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        try:
+
+            # A TTSService will emit a series of AudioRawFrame objects, then a TTSStoppedFrame,
+            # then a TextFrame.
+
+            if self._expect_text_frame_next:
+                self._expect_text_frame_next = False
+                if isinstance(frame, TextFrame):
+                    self._sentences[-1].text_frame = frame
+                else:
+                    logger.debug(f"expected a text frame, but received {frame}")
+                    await self.push_frame(frame, direction)
+                return
+            else:
+                if isinstance(frame, TextFrame):
+                    logger.error(f"XXXXXXXXXXXXXXXXXXX received a text frame, wasn't expecting it.")
+
+            if isinstance(frame, AudioRawFrame):
+                # if our buffer is empty or has a "finished" sentence at the end,
+                # then we need to start buffering a new sentence
+                if not self._sentences or self._sentences[-1].text_frame:
+                    self._sentences.append(BufferedSentence())
+                self._sentences[-1].audio_frames.append(frame)
+                await self.maybe_start_audio_pusher_task()
+                return
+
+            if isinstance(frame, TTSStoppedFrame):
+                self._expect_text_frame_next = True
+                await self.push_frame(frame, direction)
+                return
+
+            # There are two ways we can be interrupted. During greedy inference, a new
+            # LLM response can start. Or, during playout, we can get a traditional
+            # user interruption frame.
+            if (isinstance(frame, LLMFullResponseStartFrame) or
+                    isinstance(frame, StartInterruptionFrame)):
+                logger.debug(f"{frame} - Handle interruption in VADGate")
+                self._sentences = []
+                if self._audio_pusher_task:
+                    self._audio_pusher_task.cancel()
+                    self._audio_pusher_task = None
+                await self.push_frame(frame, direction)
+                return
+
+            await self.push_frame(frame, direction)
+        except Exception as e:
+            logger.debug(f"error: {e}")
+
+    async def maybe_start_audio_pusher_task(self):
+        try:
+            if self._audio_pusher_task:
+                return
+            self._audio_pusher_task = self.get_event_loop().create_task(self.push_audio())
+
+        except Exception as e:
+            logger.debug(f"Exception {e}")
+
+    async def push_audio(self):
+        try:
+            while True:
+                if not self._sentences:
+                    await asyncio.sleep(0.01)
+                    continue
+
+                if self.vad_analyzer._vad_state != VADState.QUIET:
+                    await asyncio.sleep(0.01)
+                    continue
+
+                # we only want to push completed sentence buffers
+                if not self._sentences[0].text_frame:
+                    await asyncio.sleep(0.01)
+                    continue
+
+                s = self._sentences.pop(0)
+                if not s.audio_frames:
+                    continue
+                sample_rate = s.audio_frames[0].sample_rate
+                duration = 0
+                logger.debug(f"Pushing {len(s.audio_frames)} audio frames")
+                for frame in s.audio_frames:
+                    await self.push_frame(frame)
+                    # assume linear16 encoding (2 bytes per sample). todo: add some more
+                    # metadata to AudioRawFrame, maybe
+                    duration += (len(frame.audio) / 2 / frame.num_channels) / sample_rate
+                await asyncio.sleep(duration - 20 / 1000)
+                if self.context:
+                    logger.debug(f"Appending assistant message to context: [{s.text_frame.text}]")
+                    self.context.messages.append(
+                        {"role": "assistant", "content": s.text_frame.text}
+                    )
+                await self.push_frame(s.text_frame)
+
+        except Exception as e:
+            logger.debug(f"Exception {e}")
+
+
+class TranscriptionTimingLogger(FrameProcessor):
+    def __init__(self, avt):
+        super().__init__()
+        self.name = "Transcription"
+        self._avt = avt
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        try:
+            await super().process_frame(frame, direction)
+            if isinstance(frame, TranscriptionFrame):
+                elapsed = time.time() - self._avt.last_transition_ts
+                logger.debug(f"Transcription TTF: {elapsed}")
+                await self.push_frame(MetricsFrame(ttfb={self.name: elapsed}))
+
+            await self.push_frame(frame, direction)
+        except Exception as e:
+            logger.debug(f"Exception {e}")
+
+
+class AudioVolumeTimer(FrameProcessor):
+    def __init__(self):
+        super().__init__()
+        self.last_transition_ts = 0
+        self._prev_volume = -80
+        self._speech_volume_threshold = -50
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, AudioRawFrame):
+            volume = self.calculate_volume(frame)
+            # print(f"Audio volume: {volume:.2f} dB")
+            if (volume >= self._speech_volume_threshold and
+                    self._prev_volume < self._speech_volume_threshold):
+                # logger.debug("transition above speech volume threshold")
+                self.last_transition_ts = time.time()
+            elif (volume < self._speech_volume_threshold and
+                    self._prev_volume >= self._speech_volume_threshold):
+                # logger.debug("transition below non-speech volume threshold")
+                self.last_transition_ts = time.time()
+            self._prev_volume = volume
+
+        await self.push_frame(frame, direction)
+
+    def calculate_volume(self, frame: AudioRawFrame) -> float:
+        if frame.num_channels != 1:
+            raise ValueError(f"Expected 1 channel, got {frame.num_channels}")
+
+        # Unpack audio data into 16-bit integers
+        fmt = f"{len(frame.audio) // 2}h"
+        audio_samples = struct.unpack(fmt, frame.audio)
+
+        # Calculate RMS
+        sum_squares = sum(sample**2 for sample in audio_samples)
+        rms = math.sqrt(sum_squares / len(audio_samples))
+
+        # Convert RMS to decibels (dB)
+        # Reference: maximum value for 16-bit audio is 32767
+        if rms > 0:
+            db = 20 * math.log10(rms / 32767)
+        else:
+            db = -96  # Minimum value (almost silent)
+
+        return db

examples/fast-chatbot/requirements.txt

@@ -0,0 +1,6 @@
+pipecat-ai[daily,openai,silero,deepgram]
+fastapi
+uvicorn
+requests
+python-dotenv
+loguru

examples/foundational/01-say-one-thing.py

@@ -9,11 +9,11 @@ import aiohttp
 import os
 import sys
 
-from pipecat.frames.frames import TextFrame
+from pipecat.frames.frames import EndFrame, TextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.task import PipelineTask
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
 from runner import configure
@@ -27,16 +27,15 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
+async def main(room_url):
     async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
-
         transport = DailyTransport(
             room_url, None, "Say One Thing", DailyParams(audio_out_enabled=True))
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         runner = PipelineRunner()
@@ -48,9 +47,10 @@ async def main():
         @transport.event_handler("on_participant_joined")
         async def on_new_participant_joined(transport, participant):
             participant_name = participant["info"]["userName"] or ''
-            await task.queue_frame(TextFrame(f"Hello there, {participant_name}!"))
+            await task.queue_frames([TextFrame(f"Hello there, {participant_name}!"), EndFrame()])
 
         await runner.run(task)
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url))

examples/foundational/01a-local-audio.py

@@ -9,11 +9,11 @@ import aiohttp
 import os
 import sys
 
-from pipecat.frames.frames import TextFrame
+from pipecat.frames.frames import EndFrame, TextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.transports.base_transport import TransportParams
 from pipecat.transports.local.audio import LocalAudioTransport
 
@@ -30,9 +30,10 @@ async def main():
     async with aiohttp.ClientSession() as session:
         transport = LocalAudioTransport(TransportParams(audio_out_enabled=True))
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         pipeline = Pipeline([tts, transport.output()])
@@ -41,7 +42,7 @@ async def main():
 
         async def say_something():
             await asyncio.sleep(1)
-            await task.queue_frame(TextFrame("Hello there!"))
+            await task.queue_frames([TextFrame("Hello there!"), EndFrame()])
 
         runner = PipelineRunner()
 

examples/foundational/02-llm-say-one-thing.py

@@ -9,11 +9,11 @@ import aiohttp
 import os
 import sys
 
-from pipecat.frames.frames import LLMMessagesFrame
+from pipecat.frames.frames import EndFrame, LLMMessagesFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
@@ -28,19 +28,18 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
+async def main(room_url):
     async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             None,
             "Say One Thing From an LLM",
             DailyParams(audio_out_enabled=True))
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         llm = OpenAILLMService(
@@ -59,10 +58,11 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await task.queue_frame(LLMMessagesFrame(messages))
+            await task.queue_frames([LLMMessagesFrame(messages), EndFrame()])
 
         await runner.run(task)
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url))

examples/foundational/03-still-frame.py

@@ -27,10 +27,8 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
+async def main(room_url):
     async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             None,
@@ -66,4 +64,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url))

examples/foundational/03a-local-still-frame.py

@@ -56,7 +56,7 @@ async def main():
         runner = PipelineRunner()
 
         async def run_tk():
-            while not task.has_finished():
+            while runner.is_active():
                 tk_root.update()
                 tk_root.update_idletasks()
                 await asyncio.sleep(0.1)

examples/foundational/04-utterance-and-speech.py

@@ -30,10 +30,8 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
+async def main(room_url: str):
     async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
-
         transport = DailyTransport(room_url, None, "Static And Dynamic Speech")
 
         meeting = TransportServiceOutput(transport, mic_enabled=True)
@@ -84,4 +82,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url))

examples/foundational/05-sync-speech-and-image.py

@@ -13,6 +13,7 @@ from dataclasses import dataclass
 
 from pipecat.frames.frames import (
     AppFrame,
+    EndFrame,
     Frame,
     ImageRawFrame,
     LLMFullResponseStartFrame,
@@ -72,10 +73,8 @@ class MonthPrepender(FrameProcessor):
             await self.push_frame(frame, direction)
 
 
-async def main():
+async def main(room_url):
     async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             None,
@@ -89,6 +88,7 @@ async def main():
         )
 
         tts = ElevenLabsTTSService(
+            aiohttp_session=session,
             api_key=os.getenv("ELEVENLABS_API_KEY"),
             voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
@@ -150,6 +150,8 @@ async def main():
             frames.append(MonthFrame(month=month))
             frames.append(LLMMessagesFrame(messages))
 
+        frames.append(EndFrame())
+
         runner = PipelineRunner()
 
         task = PipelineTask(pipeline)
@@ -160,4 +162,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url))

examples/foundational/05a-local-sync-speech-and-image.py

@@ -85,6 +85,7 @@ async def main():
                 model="gpt-4o")
 
             tts = ElevenLabsTTSService(
+                aiohttp_session=session,
                 api_key=os.getenv("ELEVENLABS_API_KEY"),
                 voice_id=os.getenv("ELEVENLABS_VOICE_ID"))
 
@@ -136,11 +137,14 @@ async def main():
 
         task = PipelineTask(pipeline)
 
-        # We only specify a few months as we create tasks all at once and we
-        # might get rate limited otherwise.
+        # We only specify 5 months as we create tasks all at once and we might
+        # get rate limited otherwise.
         months: list[str] = [
             "January",
             "February",
+            # "March",
+            # "April",
+            # "May",
         ]
 
         # We create one task per month. This will be executed concurrently.

examples/foundational/06-listen-and-respond.py

@@ -9,16 +9,16 @@ import aiohttp
 import os
 import sys
 
-from pipecat.frames.frames import Frame, LLMMessagesFrame, MetricsFrame
+from pipecat.frames.frames import LLMMessagesFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator,
     LLMUserResponseAggregator,
 )
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.processors.logger import FrameLogger
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -34,18 +34,8 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-class MetricsLogger(FrameProcessor):
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        if isinstance(frame, MetricsFrame):
-            print(
-                f"!!! MetricsFrame: {frame}, ttfb: {frame.ttfb}, processing: {frame.processing}, tokens: {frame.tokens}, characters: {frame.characters}")
-        await self.push_frame(frame, direction)
-
-
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -58,17 +48,19 @@ async def main():
             )
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         llm = OpenAILLMService(
             api_key=os.getenv("OPENAI_API_KEY"),
-            model="gpt-4o"
-        )
+            model="gpt-4o")
 
-        ml = MetricsLogger()
+        fl = FrameLogger("!!! after LLM", "red")
+        fltts = FrameLogger("@@@ out of tts", "green")
+        flend = FrameLogger("### out of the end", "magenta")
 
         messages = [
             {
@@ -83,18 +75,15 @@ async def main():
             transport.input(),
             tma_in,
             llm,
+            fl,
             tts,
-            ml,
+            fltts,
             transport.output(),
             tma_out,
+            flend
         ])
 
         task = PipelineTask(pipeline)
-        task = PipelineTask(pipeline, PipelineParams(
-            allow_interruptions=True,
-            enable_metrics=True,
-            report_only_initial_ttfb=False,
-        ))
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
@@ -110,4 +99,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/06a-image-sync.py

@@ -51,7 +51,7 @@ class ImageSyncAggregator(FrameProcessor):
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
-        if not isinstance(frame, SystemFrame) and direction == FrameDirection.DOWNSTREAM:
+        if not isinstance(frame, SystemFrame):
             await self.push_frame(ImageRawFrame(image=self._speaking_image_bytes, size=(1024, 1024), format=self._speaking_image_format))
             await self.push_frame(frame)
             await self.push_frame(ImageRawFrame(image=self._waiting_image_bytes, size=(1024, 1024), format=self._waiting_image_format))
@@ -59,26 +59,24 @@ class ImageSyncAggregator(FrameProcessor):
             await self.push_frame(frame)
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
             "Respond bot",
             DailyParams(
                 audio_out_enabled=True,
-                camera_out_enabled=True,
                 camera_out_width=1024,
                 camera_out_height=1024,
                 transcription_enabled=True,
                 vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer(),
+                vad_analyzer=SileroVADAnalyzer()
             )
         )
 
         tts = ElevenLabsTTSService(
+            aiohttp_session=session,
             api_key=os.getenv("ELEVENLABS_API_KEY"),
             voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
@@ -118,7 +116,7 @@ async def main():
         async def on_first_participant_joined(transport, participant):
             participant_name = participant["info"]["userName"] or ''
             transport.capture_participant_transcription(participant["id"])
-            await task.queue_frames([TextFrame(f"Hi there {participant_name}!")])
+            await task.queue_frames([TextFrame(f"Hi, this is {participant_name}.")])
 
         runner = PipelineRunner()
 
@@ -126,4 +124,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/07-interruptible.py

@@ -15,7 +15,7 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -31,10 +31,8 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -47,9 +45,10 @@ async def main():
             )
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         llm = OpenAILLMService(
@@ -78,7 +77,6 @@ async def main():
         task = PipelineTask(pipeline, PipelineParams(
             allow_interruptions=True,
             enable_metrics=True,
-            enable_usage_metrics=True,
             report_only_initial_ttfb=True,
         ))
 
@@ -96,4 +94,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/07a-interruptible-anthropic.py

@@ -15,7 +15,7 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.anthropic import AnthropicLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -31,10 +31,8 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -47,9 +45,10 @@ async def main():
             )
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         llm = AnthropicLLMService(
@@ -92,4 +91,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/07b-interruptible-langchain.py

@@ -17,7 +17,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator, LLMUserResponseAggregator)
 from pipecat.processors.frameworks.langchain import LangchainProcessor
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
 
@@ -47,10 +47,8 @@ def get_session_history(session_id: str) -> BaseChatMessageHistory:
     return message_store[session_id]
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -63,9 +61,10 @@ async def main():
             ),
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         prompt = ChatPromptTemplate.from_messages(
@@ -122,4 +121,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/07c-interruptible-deepgram.py

@@ -31,10 +31,8 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -95,4 +93,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/07d-interruptible-cartesia.py

@@ -0,0 +1,94 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+from pipecat.frames.frames import LLMMessagesFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main(room_url: str, token):
+    transport = DailyTransport(
+        room_url,
+        token,
+        "Respond bot",
+        DailyParams(
+            audio_out_enabled=True,
+            audio_out_sample_rate=44100,
+            transcription_enabled=True,
+            vad_enabled=True,
+            vad_analyzer=SileroVADAnalyzer()
+        )
+    )
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_name="British Lady",
+        output_format="pcm_44100"
+    )
+
+    llm = OpenAILLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        model="gpt-4o")
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. 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.",
+        },
+    ]
+
+    tma_in = LLMUserResponseAggregator(messages)
+    tma_out = LLMAssistantResponseAggregator(messages)
+
+    pipeline = Pipeline([
+        transport.input(),   # Transport user input
+        tma_in,              # User responses
+        llm,                 # LLM
+        tts,                 # TTS
+        transport.output(),  # Transport bot output
+        tma_out              # Assistant spoken responses
+    ])
+
+    task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+
+    @transport.event_handler("on_first_participant_joined")
+    async def on_first_participant_joined(transport, participant):
+        transport.capture_participant_transcription(participant["id"])
+        # Kick off the conversation.
+        messages.append(
+            {"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([LLMMessagesFrame(messages)])
+
+    runner = PipelineRunner()
+
+    await runner.run(task)
+
+
+if __name__ == "__main__":
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/07d-interruptible-elevenlabs.py

@@ -1,99 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-
-from pipecat.frames.frames import LLMMessagesFrame
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.llm_response import (
-    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.services.elevenlabs import ElevenLabsTTSService
-from pipecat.services.openai import OpenAILLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.vad.silero import SileroVADAnalyzer
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                transcription_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer()
-            )
-        )
-
-        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"),
-            model="gpt-4o")
-
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. 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.",
-            },
-        ]
-
-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
-
-        pipeline = Pipeline([
-            transport.input(),   # Transport user input
-            tma_in,              # User responses
-            llm,                 # LLM
-            tts,                 # TTS
-            transport.output(),  # Transport bot output
-            tma_out              # Assistant spoken responses
-        ])
-
-        task = PipelineTask(pipeline, PipelineParams(
-            allow_interruptions=True,
-            enable_metrics=True,
-            enable_usage_metrics=True,
-            report_only_initial_ttfb=True,
-        ))
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append(
-                {"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/07e-interruptible-playht.py

@@ -4,7 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import aiohttp
 import asyncio
 import os
 import sys
@@ -31,66 +30,64 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                audio_out_sample_rate=16000,
-                transcription_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer()
-            )
+async def main(room_url: str, token):
+    transport = DailyTransport(
+        room_url,
+        token,
+        "Respond bot",
+        DailyParams(
+            audio_out_enabled=True,
+            audio_out_sample_rate=16000,
+            transcription_enabled=True,
+            vad_enabled=True,
+            vad_analyzer=SileroVADAnalyzer()
         )
+    )
 
-        tts = PlayHTTTSService(
-            user_id=os.getenv("PLAYHT_USER_ID"),
-            api_key=os.getenv("PLAYHT_API_KEY"),
-            voice_url="s3://voice-cloning-zero-shot/801a663f-efd0-4254-98d0-5c175514c3e8/jennifer/manifest.json",
-        )
+    tts = PlayHTTTSService(
+        user_id=os.getenv("PLAYHT_USER_ID"),
+        api_key=os.getenv("PLAYHT_API_KEY"),
+        voice_url="s3://voice-cloning-zero-shot/801a663f-efd0-4254-98d0-5c175514c3e8/jennifer/manifest.json",
+    )
 
-        llm = OpenAILLMService(
-            api_key=os.getenv("OPENAI_API_KEY"),
-            model="gpt-4o")
+    llm = OpenAILLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        model="gpt-4o")
 
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. 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.",
-            },
-        ]
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. 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.",
+        },
+    ]
 
-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+    tma_in = LLMUserResponseAggregator(messages)
+    tma_out = LLMAssistantResponseAggregator(messages)
 
-        pipeline = Pipeline([
-            transport.input(),   # Transport user input
-            tma_in,              # User responses
-            llm,                 # LLM
-            tts,                 # TTS
-            transport.output(),  # Transport bot output
-            tma_out              # Assistant spoken responses
-        ])
+    pipeline = Pipeline([
+        transport.input(),   # Transport user input
+        tma_in,              # User responses
+        llm,                 # LLM
+        tts,                 # TTS
+        transport.output(),  # Transport bot output
+        tma_out              # Assistant spoken responses
+    ])
 
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+    task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
 
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append(
-                {"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
+    @transport.event_handler("on_first_participant_joined")
+    async def on_first_participant_joined(transport, participant):
+        transport.capture_participant_transcription(participant["id"])
+        # Kick off the conversation.
+        messages.append(
+            {"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([LLMMessagesFrame(messages)])
 
-        runner = PipelineRunner()
+    runner = PipelineRunner()
 
-        await runner.run(task)
+    await runner.run(task)
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/07f-interruptible-azure.py

@@ -4,7 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import aiohttp
 import asyncio
 import os
 import sys
@@ -31,73 +30,71 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                audio_out_sample_rate=16000,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer(),
-                vad_audio_passthrough=True,
-            )
+async def main(room_url: str, token):
+    transport = DailyTransport(
+        room_url,
+        token,
+        "Respond bot",
+        DailyParams(
+            audio_out_enabled=True,
+            audio_out_sample_rate=16000,
+            vad_enabled=True,
+            vad_analyzer=SileroVADAnalyzer(),
+            vad_audio_passthrough=True,
         )
+    )
 
-        stt = AzureSTTService(
-            api_key=os.getenv("AZURE_SPEECH_API_KEY"),
-            region=os.getenv("AZURE_SPEECH_REGION"),
-        )
+    stt = AzureSTTService(
+        api_key=os.getenv("AZURE_SPEECH_API_KEY"),
+        region=os.getenv("AZURE_SPEECH_REGION"),
+    )
 
-        tts = AzureTTSService(
-            api_key=os.getenv("AZURE_SPEECH_API_KEY"),
-            region=os.getenv("AZURE_SPEECH_REGION"),
-        )
+    tts = AzureTTSService(
+        api_key=os.getenv("AZURE_SPEECH_API_KEY"),
+        region=os.getenv("AZURE_SPEECH_REGION"),
+    )
 
-        llm = AzureLLMService(
-            api_key=os.getenv("AZURE_CHATGPT_API_KEY"),
-            endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"),
-            model=os.getenv("AZURE_CHATGPT_MODEL"),
-        )
+    llm = AzureLLMService(
+        api_key=os.getenv("AZURE_CHATGPT_API_KEY"),
+        endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"),
+        model=os.getenv("AZURE_CHATGPT_MODEL"),
+    )
 
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. 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.",
-            },
-        ]
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. 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.",
+        },
+    ]
 
-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+    tma_in = LLMUserResponseAggregator(messages)
+    tma_out = LLMAssistantResponseAggregator(messages)
 
-        pipeline = Pipeline([
-            transport.input(),   # Transport user input
-            stt,                 # STT
-            tma_in,              # User responses
-            llm,                 # LLM
-            tts,                 # TTS
-            transport.output(),  # Transport bot output
-            tma_out              # Assistant spoken responses
-        ])
+    pipeline = Pipeline([
+        transport.input(),   # Transport user input
+        stt,                 # STT
+        tma_in,              # User responses
+        llm,                 # LLM
+        tts,                 # TTS
+        transport.output(),  # Transport bot output
+        tma_out              # Assistant spoken responses
+    ])
 
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+    task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
 
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append(
-                {"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
+    @transport.event_handler("on_first_participant_joined")
+    async def on_first_participant_joined(transport, participant):
+        transport.capture_participant_transcription(participant["id"])
+        # Kick off the conversation.
+        messages.append(
+            {"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([LLMMessagesFrame(messages)])
 
-        runner = PipelineRunner()
+    runner = PipelineRunner()
 
-        await runner.run(task)
+    await runner.run(task)
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/07g-interruptible-openai-tts.py

@@ -4,7 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import aiohttp
 import asyncio
 import os
 import sys
@@ -31,65 +30,63 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                audio_out_sample_rate=24000,
-                transcription_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer()
-            )
+async def main(room_url: str, token):
+    transport = DailyTransport(
+        room_url,
+        token,
+        "Respond bot",
+        DailyParams(
+            audio_out_enabled=True,
+            audio_out_sample_rate=24000,
+            transcription_enabled=True,
+            vad_enabled=True,
+            vad_analyzer=SileroVADAnalyzer()
         )
+    )
 
-        tts = OpenAITTSService(
-            api_key=os.getenv("OPENAI_API_KEY"),
-            voice="alloy"
-        )
+    tts = OpenAITTSService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        voice="alloy"
+    )
 
-        llm = OpenAILLMService(
-            api_key=os.getenv("OPENAI_API_KEY"),
-            model="gpt-4o")
+    llm = OpenAILLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        model="gpt-4o")
 
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. 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.",
-            },
-        ]
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. 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.",
+        },
+    ]
 
-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+    tma_in = LLMUserResponseAggregator(messages)
+    tma_out = LLMAssistantResponseAggregator(messages)
 
-        pipeline = Pipeline([
-            transport.input(),   # Transport user input
-            tma_in,              # User responses
-            llm,                 # LLM
-            tts,                 # TTS
-            transport.output(),  # Transport bot output
-            tma_out              # Assistant spoken responses
-        ])
+    pipeline = Pipeline([
+        transport.input(),   # Transport user input
+        tma_in,              # User responses
+        llm,                 # LLM
+        tts,                 # TTS
+        transport.output(),  # Transport bot output
+        tma_out              # Assistant spoken responses
+    ])
 
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+    task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
 
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append(
-                {"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
+    @transport.event_handler("on_first_participant_joined")
+    async def on_first_participant_joined(transport, participant):
+        transport.capture_participant_transcription(participant["id"])
+        # Kick off the conversation.
+        messages.append(
+            {"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([LLMMessagesFrame(messages)])
 
-        runner = PipelineRunner()
+    runner = PipelineRunner()
 
-        await runner.run(task)
+    await runner.run(task)
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/07h-interruptible-openpipe.py

@@ -17,7 +17,7 @@ from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator,
     LLMUserResponseAggregator,
 )
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openpipe import OpenPipeLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -34,10 +34,8 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -50,9 +48,10 @@ async def main():
             )
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         timestamp = int(time.time())
@@ -99,4 +98,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/07i-interruptible-xtts.py

@@ -1,97 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-
-from pipecat.frames.frames import LLMMessagesFrame
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.llm_response import (
-    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.services.deepgram import DeepgramSTTService, DeepgramTTSService
-from pipecat.services.openai import OpenAILLMService
-from pipecat.services.xtts import XTTSService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.vad.silero import SileroVADAnalyzer
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                transcription_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer(),
-            )
-        )
-
-        tts = XTTSService(
-            aiohttp_session=session,
-            voice_id="Claribel Dervla",
-            language="en",
-            base_url="http://localhost:8000"
-        )
-
-        llm = OpenAILLMService(
-            api_key=os.getenv("OPENAI_API_KEY"),
-            model="gpt-4o")
-
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. 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.",
-            },
-        ]
-
-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
-
-        pipeline = Pipeline([
-            transport.input(),   # Transport user input
-            tma_in,              # User responses
-            llm,                 # LLM
-            tts,                 # TTS
-            transport.output(),  # Transport bot output
-            tma_out              # Assistant spoken responses
-        ])
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append(
-                {"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/07j-interruptible-gladia.py

@@ -1,100 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-
-from pipecat.frames.frames import LLMMessagesFrame
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.llm_response import (
-    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.services.cartesia import CartesiaTTSService
-from pipecat.services.gladia import GladiaSTTService
-from pipecat.services.openai import OpenAILLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.vad.silero import SileroVADAnalyzer
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer(),
-                vad_audio_passthrough=True,
-            )
-        )
-
-        stt = GladiaSTTService(
-            api_key=os.getenv("GLADIA_API_KEY"),
-        )
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        llm = OpenAILLMService(
-            api_key=os.getenv("OPENAI_API_KEY"),
-            model="gpt-4o")
-
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. 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.",
-            },
-        ]
-
-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
-
-        pipeline = Pipeline([
-            transport.input(),   # Transport user input
-            stt,                 # STT
-            tma_in,              # User responses
-            llm,                 # LLM
-            tts,                 # TTS
-            transport.output(),  # Transport bot output
-            tma_out              # Assistant spoken responses
-        ])
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append(
-                {"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/07k-interruptible-lmnt.py

@@ -1,95 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
-import asyncio
-import os
-import sys
-
-from pipecat.frames.frames import LLMMessagesFrame
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.llm_response import (
-    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.services.lmnt import LmntTTSService
-from pipecat.services.openai import OpenAILLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.vad.silero import SileroVADAnalyzer
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                audio_out_sample_rate=24000,
-                transcription_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer()
-            )
-        )
-
-        tts = LmntTTSService(
-            api_key=os.getenv("LMNT_API_KEY"),
-            voice_id="morgan"
-        )
-
-        llm = OpenAILLMService(
-            api_key=os.getenv("OPENAI_API_KEY"),
-            model="gpt-4o")
-
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. 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.",
-            },
-        ]
-
-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
-
-        pipeline = Pipeline([
-            transport.input(),   # Transport user input
-            tma_in,              # User responses
-            llm,                 # LLM
-            tts,                 # TTS
-            transport.output(),  # Transport bot output
-            tma_out              # Assistant spoken responses
-        ])
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append(
-                {"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/08-bots-arguing.py

@@ -22,10 +22,8 @@ logger = logging.getLogger("pipecat")
 logger.setLevel(logging.DEBUG)
 
 
-async def main():
+async def main(room_url: str):
     async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             None,
@@ -146,4 +144,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url))

examples/foundational/09-mirror.py

@@ -4,7 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import aiohttp
 import asyncio
 import sys
 
@@ -24,34 +23,32 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url, token, "Test",
-            DailyParams(
-                audio_in_enabled=True,
-                audio_out_enabled=True,
-                camera_out_enabled=True,
-                camera_out_is_live=True,
-                camera_out_width=1280,
-                camera_out_height=720
-            )
+async def main(room_url, token):
+    transport = DailyTransport(
+        room_url, token, "Test",
+        DailyParams(
+            audio_in_enabled=True,
+            audio_out_enabled=True,
+            camera_out_enabled=True,
+            camera_out_is_live=True,
+            camera_out_width=1280,
+            camera_out_height=720
         )
+    )
 
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_video(participant["id"])
+    @transport.event_handler("on_first_participant_joined")
+    async def on_first_participant_joined(transport, participant):
+        transport.capture_participant_video(participant["id"])
 
-        pipeline = Pipeline([transport.input(), transport.output()])
+    pipeline = Pipeline([transport.input(), transport.output()])
 
-        runner = PipelineRunner()
+    runner = PipelineRunner()
 
-        task = PipelineTask(pipeline)
+    task = PipelineTask(pipeline)
 
-        await runner.run(task)
+    await runner.run(task)
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/09a-local-mirror.py

@@ -4,7 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import aiohttp
 import asyncio
 import sys
 
@@ -28,44 +27,40 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
+async def main(room_url, token):
+    tk_root = tk.Tk()
+    tk_root.title("Local Mirror")
 
-        tk_root = tk.Tk()
-        tk_root.title("Local Mirror")
+    daily_transport = DailyTransport(room_url, token, "Test", DailyParams(audio_in_enabled=True))
 
-        daily_transport = DailyTransport(
-            room_url, token, "Test", DailyParams(
-                audio_in_enabled=True))
+    tk_transport = TkLocalTransport(
+        tk_root,
+        TransportParams(
+            audio_out_enabled=True,
+            camera_out_enabled=True,
+            camera_out_is_live=True,
+            camera_out_width=1280,
+            camera_out_height=720))
 
-        tk_transport = TkLocalTransport(
-            tk_root,
-            TransportParams(
-                audio_out_enabled=True,
-                camera_out_enabled=True,
-                camera_out_is_live=True,
-                camera_out_width=1280,
-                camera_out_height=720))
+    @daily_transport.event_handler("on_first_participant_joined")
+    async def on_first_participant_joined(transport, participant):
+        transport.capture_participant_video(participant["id"])
 
-        @daily_transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_video(participant["id"])
+    pipeline = Pipeline([daily_transport.input(), tk_transport.output()])
 
-        pipeline = Pipeline([daily_transport.input(), tk_transport.output()])
+    task = PipelineTask(pipeline)
 
-        task = PipelineTask(pipeline)
+    async def run_tk():
+        while not task.has_finished():
+            tk_root.update()
+            tk_root.update_idletasks()
+            await asyncio.sleep(0.1)
 
-        async def run_tk():
-            while not task.has_finished():
-                tk_root.update()
-                tk_root.update_idletasks()
-                await asyncio.sleep(0.1)
+    runner = PipelineRunner()
 
-        runner = PipelineRunner()
-
-        await asyncio.gather(runner.run(task), run_tk())
+    await asyncio.gather(runner.run(task), run_tk())
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/10-wake-phrase.py

@@ -15,7 +15,7 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -31,10 +31,9 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
+async def main(room_url: str, token):
 
+    async with aiohttp.ClientSession() as session:
         transport = DailyTransport(
             room_url,
             token,
@@ -47,9 +46,10 @@ async def main():
             )
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         llm = OpenAILLMService(
@@ -90,4 +90,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/11-sound-effects.py

@@ -83,10 +83,8 @@ class InboundSoundEffectWrapper(FrameProcessor):
             await self.push_frame(frame, direction)
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -104,6 +102,7 @@ async def main():
             model="gpt-4o")
 
         tts = ElevenLabsTTSService(
+            aiohttp_session=session,
             api_key=os.getenv("ELEVENLABS_API_KEY"),
             voice_id="ErXwobaYiN019PkySvjV",
         )
@@ -149,4 +148,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/12-describe-video.py

@@ -16,7 +16,7 @@ from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.user_response import UserResponseAggregator
 from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.moondream import MoondreamService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -49,10 +49,8 @@ class UserImageRequester(FrameProcessor):
         await self.push_frame(frame, direction)
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -65,6 +63,12 @@ async def main():
             )
         )
 
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
+        )
+
         user_response = UserResponseAggregator()
 
         image_requester = UserImageRequester()
@@ -74,9 +78,10 @@ async def main():
         # If you run into weird description, try with use_cpu=True
         moondream = MoondreamService()
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         @transport.event_handler("on_first_participant_joined")
@@ -103,4 +108,5 @@ async def main():
         await runner.run(task)
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/12a-describe-video-gemini-flash.py

@@ -16,7 +16,7 @@ from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.user_response import UserResponseAggregator
 from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.google import GoogleLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -49,10 +49,8 @@ class UserImageRequester(FrameProcessor):
         await self.push_frame(frame, direction)
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -76,9 +74,10 @@ async def main():
             model="gemini-1.5-flash-latest",
             api_key=os.getenv("GOOGLE_API_KEY"))
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         @transport.event_handler("on_first_participant_joined")
@@ -105,4 +104,5 @@ async def main():
         await runner.run(task)
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/12b-describe-video-gpt-4o.py

@@ -16,7 +16,7 @@ from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.user_response import UserResponseAggregator
 from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -49,10 +49,8 @@ class UserImageRequester(FrameProcessor):
         await self.push_frame(frame, direction)
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -76,9 +74,10 @@ async def main():
             model="gpt-4o"
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         @transport.event_handler("on_first_participant_joined")
@@ -105,4 +104,5 @@ async def main():
         await runner.run(task)
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/12c-describe-video-anthropic.py

@@ -16,7 +16,7 @@ from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.user_response import UserResponseAggregator
 from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.anthropic import AnthropicLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -49,10 +49,8 @@ class UserImageRequester(FrameProcessor):
         await self.push_frame(frame, direction)
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -72,13 +70,14 @@ async def main():
         vision_aggregator = VisionImageFrameAggregator()
 
         anthropic = AnthropicLLMService(
-            api_key=os.getenv("ANTHROPIC_API_KEY")
+            api_key=os.getenv("ANTHROPIC_API_KEY"),
+            model="claude-3-sonnet-20240229"
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-            sample_rate=16000,
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         @transport.event_handler("on_first_participant_joined")
@@ -105,4 +104,5 @@ async def main():
         await runner.run(task)
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/13-whisper-transcription.py

@@ -4,7 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import aiohttp
 import asyncio
 import sys
 
@@ -36,25 +35,23 @@ class TranscriptionLogger(FrameProcessor):
             print(f"Transcription: {frame.text}")
 
 
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
+async def main(room_url: str):
+    transport = DailyTransport(room_url, None, "Transcription bot",
+                               DailyParams(audio_in_enabled=True))
 
-        transport = DailyTransport(room_url, None, "Transcription bot",
-                                   DailyParams(audio_in_enabled=True))
+    stt = WhisperSTTService()
 
-        stt = WhisperSTTService()
+    tl = TranscriptionLogger()
 
-        tl = TranscriptionLogger()
+    pipeline = Pipeline([transport.input(), stt, tl])
 
-        pipeline = Pipeline([transport.input(), stt, tl])
+    task = PipelineTask(pipeline)
 
-        task = PipelineTask(pipeline)
+    runner = PipelineRunner()
 
-        runner = PipelineRunner()
-
-        await runner.run(task)
+    await runner.run(task)
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url))

examples/foundational/13b-deepgram-transcription.py

@@ -4,7 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import aiohttp
 import asyncio
 import os
 import sys
@@ -37,25 +36,23 @@ class TranscriptionLogger(FrameProcessor):
             print(f"Transcription: {frame.text}")
 
 
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
+async def main(room_url: str):
+    transport = DailyTransport(room_url, None, "Transcription bot",
+                               DailyParams(audio_in_enabled=True))
 
-        transport = DailyTransport(room_url, None, "Transcription bot",
-                                   DailyParams(audio_in_enabled=True))
+    stt = DeepgramSTTService(os.getenv("DEEPGRAM_API_KEY"))
 
-        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+    tl = TranscriptionLogger()
 
-        tl = TranscriptionLogger()
+    pipeline = Pipeline([transport.input(), stt, tl])
 
-        pipeline = Pipeline([transport.input(), stt, tl])
+    task = PipelineTask(pipeline)
 
-        task = PipelineTask(pipeline)
+    runner = PipelineRunner()
 
-        runner = PipelineRunner()
-
-        await runner.run(task)
+    await runner.run(task)
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url))

examples/foundational/14-function-calling.py

@@ -13,8 +13,12 @@ from pipecat.frames.frames import TextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantContextAggregator,
+    LLMUserContextAggregator,
+)
 from pipecat.processors.logger import FrameLogger
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMContext, OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -32,18 +36,16 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def start_fetch_weather(function_name, llm, context):
-    await llm.push_frame(TextFrame("Let me check on that."))
+async def start_fetch_weather(llm):
+    await llm.push_frame(TextFrame("Let me think."))
 
 
-async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
-    await result_callback({"conditions": "nice", "temperature": "75"})
+async def fetch_weather_from_api(llm, args):
+    return {"conditions": "nice", "temperature": "75"}
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -56,18 +58,17 @@ async def main():
             )
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
 
         llm = OpenAILLMService(
             api_key=os.getenv("OPENAI_API_KEY"),
             model="gpt-4o")
-        # Register a function_name of None to get all functions
-        # sent to the same callback with an additional function_name parameter.
         llm.register_function(
-            None,
+            "get_current_weather",
             fetch_weather_from_api,
             start_callback=start_fetch_weather)
 
@@ -108,22 +109,22 @@ async def main():
         ]
 
         context = OpenAILLMContext(messages, tools)
-        context_aggregator = llm.create_context_aggregator(context)
-
+        tma_in = LLMUserContextAggregator(context)
+        tma_out = LLMAssistantContextAggregator(context)
         pipeline = Pipeline([
             fl_in,
             transport.input(),
-            context_aggregator.user(),
+            tma_in,
             llm,
             fl_out,
             tts,
             transport.output(),
-            context_aggregator.assistant(),
+            tma_out
         ])
 
         task = PipelineTask(pipeline)
 
-        @transport.event_handler("on_first_participant_joined")
+        @ transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
@@ -133,5 +134,7 @@ async def main():
 
         await runner.run(task)
 
+
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/15-switch-voices.py

@@ -4,8 +4,8 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import aiohttp
 import asyncio
+import aiohttp
 import os
 import sys
 
@@ -14,6 +14,10 @@ from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantContextAggregator,
+    LLMUserContextAggregator
+)
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.filters.function_filter import FunctionFilter
 from pipecat.services.cartesia import CartesiaTTSService
@@ -36,10 +40,10 @@ logger.add(sys.stderr, level="DEBUG")
 current_voice = "News Lady"
 
 
-async def switch_voice(function_name, tool_call_id, args, llm, context, result_callback):
+async def switch_voice(llm, args):
     global current_voice
     current_voice = args["voice"]
-    await result_callback({"voice": f"You are now using your {current_voice} voice. Your responses should now be as if you were a {current_voice}."})
+    return {"voice": f"You are now using your {current_voice} voice. Your responses should now be as if you were a {current_voice}."}
 
 
 async def news_lady_filter(frame) -> bool:
@@ -54,16 +58,15 @@ async def barbershop_man_filter(frame) -> bool:
     return current_voice == "Barbershop Man"
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
             "Pipecat",
             DailyParams(
                 audio_out_enabled=True,
+                audio_out_sample_rate=44100,
                 transcription_enabled=True,
                 vad_enabled=True,
                 vad_analyzer=SileroVADAnalyzer()
@@ -72,17 +75,20 @@ async def main():
 
         news_lady = CartesiaTTSService(
             api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="bf991597-6c13-47e4-8411-91ec2de5c466",  # Newslady
+            voice_name="Newslady",
+            output_format="pcm_44100"
         )
 
         british_lady = CartesiaTTSService(
             api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+            voice_name="British Lady",
+            output_format="pcm_44100"
         )
 
         barbershop_man = CartesiaTTSService(
             api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="a0e99841-438c-4a64-b679-ae501e7d6091",  # Barbershop Man
+            voice_name="Barbershop Man",
+            output_format="pcm_44100"
         )
 
         llm = OpenAILLMService(
@@ -115,11 +121,12 @@ async def main():
         ]
 
         context = OpenAILLMContext(messages, tools)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserContextAggregator(context)
+        tma_out = LLMAssistantContextAggregator(context)
 
         pipeline = Pipeline([
             transport.input(),   # Transport user input
-            context_aggregator.user(),  # User responses
+            tma_in,              # User responses
             llm,                 # LLM
             ParallelPipeline(    # TTS (one of the following vocies)
                 [FunctionFilter(news_lady_filter), news_lady],            # News Lady voice
@@ -127,7 +134,7 @@ async def main():
                 [FunctionFilter(barbershop_man_filter), barbershop_man],  # Barbershop Man voice
             ),
             transport.output(),  # Transport bot output
-            context_aggregator.assistant(),  # Assistant spoken responses
+            tma_out              # Assistant spoken responses
         ])
 
         task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@@ -148,4 +155,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/15a-switch-languages.py

@@ -14,9 +14,13 @@ from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantContextAggregator,
+    LLMUserContextAggregator
+)
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.filters.function_filter import FunctionFilter
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.services.whisper import Model, WhisperSTTService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -37,10 +41,10 @@ logger.add(sys.stderr, level="DEBUG")
 current_language = "English"
 
 
-async def switch_language(function_name, tool_call_id, args, llm, context, result_callback):
+async def switch_language(llm, args):
     global current_language
     current_language = args["language"]
-    await result_callback({"voice": f"Your answers from now on should be in {current_language}."})
+    return {"voice": f"Your answers from now on should be in {current_language}."}
 
 
 async def english_filter(frame) -> bool:
@@ -51,10 +55,8 @@ async def spanish_filter(frame) -> bool:
     return current_language == "Spanish"
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -70,14 +72,17 @@ async def main():
 
         stt = WhisperSTTService(model=Model.LARGE)
 
-        english_tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        english_tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id="pNInz6obpgDQGcFmaJgB",
         )
 
-        spanish_tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="846d6cb0-2301-48b6-9683-48f5618ea2f6",  # Spanish-speaking Lady
+        spanish_tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            model="eleven_multilingual_v2",
+            voice_id="9F4C8ztpNUmXkdDDbz3J",
         )
 
         llm = OpenAILLMService(
@@ -110,19 +115,20 @@ async def main():
         ]
 
         context = OpenAILLMContext(messages, tools)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserContextAggregator(context)
+        tma_out = LLMAssistantContextAggregator(context)
 
         pipeline = Pipeline([
             transport.input(),   # Transport user input
             stt,                 # STT
-            context_aggregator.user(),  # User responses
+            tma_in,              # User responses
             llm,                 # LLM
             ParallelPipeline(    # TTS (bot will speak the chosen language)
                 [FunctionFilter(english_filter), english_tts],  # English
                 [FunctionFilter(spanish_filter), spanish_tts],  # Spanish
             ),
             transport.output(),  # Transport bot output
-            context_aggregator.assistant()  # Assistant spoken responses
+            tma_out              # Assistant spoken responses
         ])
 
         task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@@ -143,4 +149,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/16-gpu-container-local-bot.py

@@ -8,6 +8,7 @@ import asyncio
 import aiohttp
 import os
 import sys
+import json
 
 from pipecat.frames.frames import LLMMessagesFrame
 from pipecat.pipeline.pipeline import Pipeline
@@ -31,10 +32,8 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -127,4 +126,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/foundational/17-detect-user-idle.py

@@ -1,107 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-
-from pipecat.frames.frames import LLMMessagesFrame
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.llm_response import (
-    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.processors.user_idle_processor import UserIdleProcessor
-from pipecat.services.cartesia import CartesiaTTSService
-from pipecat.services.openai import OpenAILLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.vad.silero import SileroVADAnalyzer
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                transcription_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer()
-            )
-        )
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        llm = OpenAILLMService(
-            api_key=os.getenv("OPENAI_API_KEY"),
-            model="gpt-4o")
-
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. 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.",
-            },
-        ]
-
-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
-
-        async def user_idle_callback(user_idle: UserIdleProcessor):
-            messages.append(
-                {"role": "system", "content": "Ask the user if they are still there and try to prompt for some input, but be short."})
-            await user_idle.queue_frame(LLMMessagesFrame(messages))
-
-        user_idle = UserIdleProcessor(callback=user_idle_callback, timeout=5.0)
-
-        pipeline = Pipeline([
-            transport.input(),   # Transport user input
-            user_idle,           # Idle user check-in
-            tma_in,              # User responses
-            llm,                 # LLM
-            tts,                 # TTS
-            transport.output(),  # Transport bot output
-            tma_out              # Assistant spoken responses
-        ])
-
-        task = PipelineTask(pipeline, PipelineParams(
-            allow_interruptions=True,
-            enable_metrics=True,
-            report_only_initial_ttfb=True,
-        ))
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append(
-                {"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/18-gstreamer-filesrc.py

@@ -1,78 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import argparse
-import sys
-
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.gstreamer.pipeline_source import GStreamerPipelineSource
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-from runner import configure_with_args
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
-        parser.add_argument(
-            "-i",
-            "--input",
-            type=str,
-            required=True,
-            help="Input video file")
-
-        (room_url, _, args) = await configure_with_args(session, parser)
-
-        transport = DailyTransport(
-            room_url,
-            None,
-            "GStreamer",
-            DailyParams(
-                audio_out_enabled=True,
-                audio_out_is_live=True,
-                camera_out_enabled=True,
-                camera_out_width=1280,
-                camera_out_height=720,
-                camera_out_is_live=True,
-            )
-        )
-
-        gst = GStreamerPipelineSource(
-            pipeline=f"filesrc location={args.input}",
-            out_params=GStreamerPipelineSource.OutputParams(
-                video_width=1280,
-                video_height=720,
-                audio_sample_rate=16000,
-                audio_channels=1,
-            )
-        )
-
-        pipeline = Pipeline([
-            gst,                 # GStreamer file source
-            transport.output(),  # Transport bot output
-        ])
-
-        task = PipelineTask(pipeline)
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/18a-gstreamer-videotestsrc.py

@@ -1,64 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import sys
-
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.gstreamer.pipeline_source import GStreamerPipelineSource
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            None,
-            "GStreamer",
-            DailyParams(
-                camera_out_enabled=True,
-                camera_out_width=1280,
-                camera_out_height=720,
-                camera_out_is_live=True,
-            )
-        )
-
-        gst = GStreamerPipelineSource(
-            pipeline="videotestsrc ! capsfilter caps=\"video/x-raw,width=1280,height=720,framerate=30/1\"",
-            out_params=GStreamerPipelineSource.OutputParams(
-                video_width=1280,
-                video_height=720,
-                clock_sync=False))
-
-        pipeline = Pipeline([
-            gst,                 # GStreamer file source
-            transport.output(),  # Transport bot output
-        ])
-
-        task = PipelineTask(pipeline)
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/19a-tools-anthropic.py

@@ -1,116 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-
-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 import CartesiaTTSService
-from pipecat.services.anthropic import AnthropicLLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.vad.silero import SileroVADAnalyzer
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def get_weather(function_name, tool_call_id, arguments, llm, context, result_callback):
-    location = arguments["location"]
-    await result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                transcription_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer()
-            )
-        )
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        llm = AnthropicLLMService(
-            api_key=os.getenv("ANTHROPIC_API_KEY"),
-            model="claude-3-5-sonnet-20240620"
-        )
-        llm.register_function("get_weather", get_weather)
-
-        tools = [
-            {
-                "name": "get_weather",
-                "description": "Get the current weather in a given location",
-                "input_schema": {
-                    "type": "object",
-                    "properties": {
-                        "location": {
-                            "type": "string",
-                            "description": "The city and state, e.g. San Francisco, CA",
-                        }
-                    },
-                    "required": ["location"],
-                },
-            }
-        ]
-
-        # todo: test with very short initial user message
-
-        # messages = [{"role": "system",
-        #              "content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation."},
-        #             {"role": "user",
-        #              "content": " Start the conversation by introducing yourself."}]
-
-        messages = [{"role": "user", "content": "Say 'hello' to start the conversation."}]
-
-        context = OpenAILLMContext(messages, tools)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline([
-            transport.input(),               # Transport user input
-            context_aggregator.user(),       # User spoken responses
-            llm,                             # LLM
-            tts,                             # TTS
-            transport.output(),              # Transport bot output
-            context_aggregator.assistant(),  # Assistant spoken responses and tool context
-        ])
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True))
-
-        @ transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/19b-tools-video-anthropic.py

@@ -1,172 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-
-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 import CartesiaTTSService
-from pipecat.services.anthropic import AnthropicLLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.vad.silero import SileroVADAnalyzer
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-video_participant_id = None
-
-
-async def get_weather(function_name, tool_call_id, arguments, llm, context, result_callback):
-    location = arguments["location"]
-    await result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
-
-
-async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
-    question = arguments["question"]
-    await llm.request_image_frame(user_id=video_participant_id, text_content=question)
-
-
-async def main():
-    global llm
-
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                transcription_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer()
-            )
-        )
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        llm = AnthropicLLMService(
-            api_key=os.getenv("ANTHROPIC_API_KEY"),
-            model="claude-3-5-sonnet-20240620",
-            enable_prompt_caching_beta=True
-        )
-        llm.register_function("get_weather", get_weather)
-        llm.register_function("get_image", get_image)
-
-        tools = [
-            {
-                "name": "get_weather",
-                "description": "Get the current weather in a given location",
-                "input_schema": {
-                    "type": "object",
-                    "properties": {
-                        "location": {
-                            "type": "string",
-                            "description": "The city and state, e.g. San Francisco, CA",
-                        }
-                    },
-                    "required": ["location"],
-                },
-            },
-            {
-                "name": "get_image",
-                "description": "Get an image from the video stream.",
-                "input_schema": {
-                    "type": "object",
-                    "properties": {
-                        "question": {
-                            "type": "string",
-                            "description": "The question that the user is asking about the image.",
-                        }
-                    },
-                    "required": ["question"],
-                },
-            }
-        ]
-
-        # todo: test with very short initial user message
-
-        system_prompt = """\
-You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions.
-
-Your response will be turned into speech so use only simple words and punctuation.
-
-You have access to two tools: get_weather and get_image.
-
-You can respond to questions about the weather using the get_weather tool.
-
-You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \
-indicate you should use the get_image tool are:
-  - What do you see?
-  - What's in the video?
-  - Can you describe the video?
-  - Tell me about what you see.
-  - Tell me something interesting about what you see.
-  - What's happening in the video?
-
-If you need to use a tool, simply use the tool. Do not tell the user the tool you are using. Be brief and concise.
-        """
-
-        messages = [
-            {
-                "role": "system",
-                "content": [
-                    {
-                        "type": "text",
-                        "text": system_prompt,
-                    }
-                ]
-            },
-            {
-                "role": "user",
-                "content": "Start the conversation by introducing yourself."
-            }]
-
-        context = OpenAILLMContext(messages, tools)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline([
-            transport.input(),               # Transport user input
-            context_aggregator.user(),       # User speech to text
-            llm,                             # LLM
-            tts,                             # TTS
-            transport.output(),              # Transport bot output
-            context_aggregator.assistant(),  # Assistant spoken responses and tool context
-        ])
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True))
-
-        @ transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            global video_participant_id
-            video_participant_id = participant["id"]
-            transport.capture_participant_transcription(video_participant_id)
-            transport.capture_participant_video(video_participant_id, framerate=0)
-            # Kick off the conversation.
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
-
-        runner = PipelineRunner()
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/19c-tools-togetherai.py

@@ -1,138 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-import json
-
-from pipecat.frames.frames import LLMMessagesFrame
-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 import CartesiaTTSService
-from pipecat.services.together import TogetherLLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.vad.silero import SileroVADAnalyzer
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def get_current_weather(
-        function_name,
-        tool_call_id,
-        arguments,
-        llm,
-        context,
-        result_callback):
-    logger.debug("IN get_current_weather")
-    location = arguments["location"]
-    await result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                transcription_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer()
-            )
-        )
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        llm = TogetherLLMService(
-            api_key=os.getenv("TOGETHER_API_KEY"),
-            model=os.getenv("TOGETHER_MODEL"),
-        )
-        llm.register_function("get_current_weather", get_current_weather)
-
-        weatherTool = {
-            "name": "get_current_weather",
-            "description": "Get the current weather in a given location",
-            "parameters": {
-                "type": "object",
-                "properties": {
-                    "location": {
-                        "type": "string",
-                        "description": "The city and state, e.g. San Francisco, CA",
-                    },
-                },
-                "required": ["location"],
-            },
-        }
-
-        system_prompt = f"""\
-You have access to the following functions:
-
-Use the function '{weatherTool["name"]}' to '{weatherTool["description"]}':
-{json.dumps(weatherTool)}
-
-If you choose to call a function ONLY reply in the following format with no prefix or suffix:
-
-<function=example_function_name>{{\"example_name\": \"example_value\"}}</function>
-
-Reminder:
-- Function calls MUST follow the specified format, start with <function= and end with </function>
-- Required parameters MUST be specified
-- Only call one function at a time
-- Put the entire function call reply on one line
-- If there is no function call available, answer the question like normal with your current knowledge and do not tell the user about function calls
-
-"""
-
-        messages = [{"role": "system",
-                     "content": system_prompt},
-                    {"role": "user",
-                     "content": "Wait for the user to say something."}]
-
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline([
-            transport.input(),               # Transport user input
-            context_aggregator.user(),       # User speech to text
-            llm,                             # LLM
-            tts,                             # TTS
-            transport.output(),              # Transport bot output
-            context_aggregator.assistant(),  # Assistant spoken responses and tool context
-        ])
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True))
-
-        @ transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/runner.py

@@ -1,26 +1,12 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
 import argparse
 import os
-
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+import time
+import urllib
+import requests
 
 
-async def configure(aiohttp_session: aiohttp.ClientSession):
-    (url, token, _) = await configure_with_args(aiohttp_session)
-    return (url, token)
-
-
-async def configure_with_args(
-        aiohttp_session: aiohttp.ClientSession,
-        parser: argparse.ArgumentParser | None = None):
-    if not parser:
-        parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
+def configure():
+    parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
     parser.add_argument(
         "-u",
         "--url",
@@ -47,15 +33,26 @@ async def configure_with_args(
     if not key:
         raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
 
-    daily_rest_helper = DailyRESTHelper(
-        daily_api_key=key,
-        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
-        aiohttp_session=aiohttp_session)
-
     # Create a meeting token for the given room with an expiration 1 hour in
     # the future.
-    expiry_time: float = 60 * 60
+    room_name: str = urllib.parse.urlparse(url).path[1:]
+    expiration: float = time.time() + 60 * 60
 
-    token = await daily_rest_helper.get_token(url, expiry_time)
+    res: requests.Response = requests.post(
+        f"https://api.daily.co/v1/meeting-tokens",
+        headers={
+            "Authorization": f"Bearer {key}"},
+        json={
+            "properties": {
+                "room_name": room_name,
+                "is_owner": True,
+                "exp": expiration}},
+    )
 
-    return (url, token, args)
+    if res.status_code != 200:
+        raise Exception(
+            f"Failed to create meeting token: {res.status_code} {res.text}")
+
+    token: str = res.json()["token"]
+
+    return (url, token)

examples/moondream-chatbot/bot.py

@@ -1,9 +1,3 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
 import asyncio
 import aiohttp
 import os
@@ -31,7 +25,7 @@ from pipecat.processors.aggregators.llm_response import LLMUserResponseAggregato
 from pipecat.processors.aggregators.sentence import SentenceAggregator
 from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.moondream import MoondreamService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -134,10 +128,8 @@ class ImageFilterProcessor(FrameProcessor):
             await self.push_frame(frame)
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -153,9 +145,10 @@ async def main():
             )
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id="pNInz6obpgDQGcFmaJgB",
         )
 
         llm = OpenAILLMService(
@@ -211,4 +204,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/moondream-chatbot/requirements.txt

@@ -1,4 +1,5 @@
 python-dotenv
+requests
 fastapi[all]
 uvicorn
 pipecat-ai[daily,moondream,openai,silero]

examples/moondream-chatbot/runner.py

@@ -1,17 +1,11 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
 import argparse
 import os
-
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+import time
+import urllib
+import requests
 
 
-async def configure(aiohttp_session: aiohttp.ClientSession):
+def configure():
     parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
     parser.add_argument(
         "-u",
@@ -39,16 +33,26 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
     if not key:
         raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
 
-    daily_rest_helper = DailyRESTHelper(
-        daily_api_key=key,
-        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
-        aiohttp_session=aiohttp_session
-    )
-
     # Create a meeting token for the given room with an expiration 1 hour in
     # the future.
-    expiry_time: float = 60 * 60
+    room_name: str = urllib.parse.urlparse(url).path[1:]
+    expiration: float = time.time() + 60 * 60
 
-    token = await daily_rest_helper.get_token(url, expiry_time)
+    res: requests.Response = requests.post(
+        f"https://api.daily.co/v1/meeting-tokens",
+        headers={
+            "Authorization": f"Bearer {key}"},
+        json={
+            "properties": {
+                "room_name": room_name,
+                "is_owner": True,
+                "exp": expiration}},
+    )
+
+    if res.status_code != 200:
+        raise Exception(
+            f"Failed to create meeting token: {res.status_code} {res.text}")
+
+    token: str = res.json()["token"]
 
     return (url, token)

examples/moondream-chatbot/server.py

@@ -1,51 +1,31 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
 import os
 import argparse
 import subprocess
-
-from contextlib import asynccontextmanager
+import atexit
 
 from fastapi import FastAPI, Request, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse, RedirectResponse
 
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
+from utils.daily_helpers import create_room as _create_room, get_token
 
 MAX_BOTS_PER_ROOM = 1
 
 # Bot sub-process dict for status reporting and concurrency control
 bot_procs = {}
 
-daily_helpers = {}
-
 
 def cleanup():
     # Clean up function, just to be extra safe
-    for entry in bot_procs.values():
-        proc = entry[0]
+    for proc in bot_procs.values():
         proc.terminate()
         proc.wait()
 
 
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    aiohttp_session = aiohttp.ClientSession()
-    daily_helpers["rest"] = DailyRESTHelper(
-        daily_api_key=os.getenv("DAILY_API_KEY", ""),
-        daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
-        aiohttp_session=aiohttp_session
-    )
-    yield
-    await aiohttp_session.close()
-    cleanup()
+atexit.register(cleanup)
 
-app = FastAPI(lifespan=lifespan)
+
+app = FastAPI()
 
 app.add_middleware(
     CORSMiddleware,
@@ -59,45 +39,45 @@ app.add_middleware(
 @app.get("/start")
 async def start_agent(request: Request):
     print(f"!!! Creating room")
-    room = await daily_helpers["rest"].create_room(DailyRoomParams())
-    print(f"!!! Room URL: {room.url}")
+    room_url, room_name = _create_room()
+    print(f"!!! Room URL: {room_url}")
     # Ensure the room property is present
-    if not room.url:
+    if not room_url:
         raise HTTPException(
             status_code=500,
             detail="Missing 'room' property in request data. Cannot start agent without a target room!")
 
     # Check if there is already an existing process running in this room
     num_bots_in_room = sum(
-        1 for proc in bot_procs.values() if proc[1] == room.url and proc[0].poll() is None)
+        1 for proc in bot_procs.values() if proc[1] == room_url and proc[0].poll() is None)
     if num_bots_in_room >= MAX_BOTS_PER_ROOM:
         raise HTTPException(
-            status_code=500, detail=f"Max bot limited reach for room: {room.url}")
+            status_code=500, detail=f"Max bot limited reach for room: {room_url}")
 
     # Get the token for the room
-    token = await daily_helpers["rest"].get_token(room.url)
+    token = get_token(room_url)
 
     if not token:
         raise HTTPException(
-            status_code=500, detail=f"Failed to get token for room: {room.url}")
+            status_code=500, detail=f"Failed to get token for room: {room_url}")
 
     # Spawn a new agent, and join the user session
     # Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
     try:
         proc = subprocess.Popen(
             [
-                f"python3 -m bot -u {room.url} -t {token}"
+                f"python3 -m bot -u {room_url} -t {token}"
             ],
             shell=True,
             bufsize=1,
             cwd=os.path.dirname(os.path.abspath(__file__))
         )
-        bot_procs[proc.pid] = (proc, room.url)
+        bot_procs[proc.pid] = (proc, room_url)
     except Exception as e:
         raise HTTPException(
             status_code=500, detail=f"Failed to start subprocess: {e}")
 
-    return RedirectResponse(room.url)
+    return RedirectResponse(room_url)
 
 
 @app.get("/status/{pid}")

examples/moondream-chatbot/utils/daily_helpers.py

@@ -0,0 +1,109 @@
+
+import urllib.parse
+import os
+import time
+import urllib
+import requests
+
+from dotenv import load_dotenv
+load_dotenv()
+
+
+daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
+daily_api_key = os.getenv("DAILY_API_KEY")
+
+
+def create_room() -> tuple[str, str]:
+    """
+    Helper function to create a Daily room.
+    # See: https://docs.daily.co/reference/rest-api/rooms
+
+    Returns:
+        tuple: A tuple containing the room URL and room name.
+
+    Raises:
+        Exception: If the request to create the room fails or if the response does not contain the room URL or room name.
+    """
+    room_props = {
+        "exp": time.time() + 60 * 60,  # 1 hour
+        "enable_chat": True,
+        "enable_emoji_reactions": True,
+        "eject_at_room_exp": True,
+        "enable_prejoin_ui": False,  # Important for the bot to be able to join headlessly
+    }
+    res = requests.post(
+        f"https://{daily_api_path}/rooms",
+        headers={"Authorization": f"Bearer {daily_api_key}"},
+        json={
+            "properties": room_props
+        },
+    )
+    if res.status_code != 200:
+        raise Exception(f"Unable to create room: {res.text}")
+
+    data = res.json()
+    room_url: str = data.get("url")
+    room_name: str = data.get("name")
+    if room_url is None or room_name is None:
+        raise Exception("Missing room URL or room name in response")
+
+    return room_url, room_name
+
+
+def get_name_from_url(room_url: str) -> str:
+    """
+    Extracts the name from a given room URL.
+
+    Args:
+        room_url (str): The URL of the room.
+
+    Returns:
+        str: The extracted name from the room URL.
+    """
+    return urllib.parse.urlparse(room_url).path[1:]
+
+
+def get_token(room_url: str) -> str:
+    """
+    Retrieves a meeting token for the specified Daily room URL.
+    # See: https://docs.daily.co/reference/rest-api/meeting-tokens
+
+    Args:
+        room_url (str): The URL of the Daily room.
+
+    Returns:
+        str: The meeting token.
+
+    Raises:
+        Exception: If no room URL is specified or if no Daily API key is specified.
+        Exception: If there is an error creating the meeting token.
+    """
+    if not room_url:
+        raise Exception(
+            "No Daily room specified. You must specify a Daily room in order a token to be generated.")
+
+    if not daily_api_key:
+        raise Exception(
+            "No Daily API key specified. set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
+
+    expiration: float = time.time() + 60 * 60
+    room_name = get_name_from_url(room_url)
+
+    res: requests.Response = requests.post(
+        f"https://{daily_api_path}/meeting-tokens",
+        headers={
+            "Authorization": f"Bearer {daily_api_key}"},
+        json={
+            "properties": {
+                "room_name": room_name,
+                "is_owner": True,  # Owner tokens required for transcription
+                "exp": expiration}},
+    )
+
+    if res.status_code != 200:
+        raise Exception(
+            f"Failed to create meeting token: {res.status_code} {res.text}")
+
+    token: str = res.json()["token"]
+
+    return token

examples/patient-intake/bot.py

@@ -10,14 +10,24 @@ import os
 import sys
 import wave
 
+from typing import List
+
+from openai._types import NotGiven, NOT_GIVEN
+
+from openai.types.chat import (
+    ChatCompletionToolParam,
+)
+
 from pipecat.frames.frames import AudioRawFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator, LLMAssistantContextAggregator
 from pipecat.processors.logger import FrameLogger
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService
+from pipecat.services.ai_services import AIService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
 
@@ -54,11 +64,20 @@ for file in sound_files:
 
 
 class IntakeProcessor:
-
-    def __init__(self, context: OpenAILLMContext):
+    def __init__(
+        self,
+        context: OpenAILLMContext,
+        llm: AIService,
+        tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN,
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+        self._context: OpenAILLMContext = context
+        self._llm = llm
         print(f"Initializing context from IntakeProcessor")
-        context.add_message({"role": "system", "content": "You are Jessica, an agent for a company called Tri-County Health Services. Your job is to collect important information from the user before their doctor visit. You're talking to Chad Bailey. You should address the user by their first name and be polite and professional. You're not a medical professional, so you shouldn't provide any advice. Keep your responses short. Your job is to collect information to give to a doctor. Don't make assumptions about what values to plug into functions. Ask for clarification if a user response is ambiguous. Start by introducing yourself. Then, ask the user to confirm their identity by telling you their birthday, including the year. When they answer with their birthday, call the verify_birthday function."})
-        context.set_tools([
+        self._context.add_message({"role": "system", "content": "You are Jessica, an agent for a company called Tri-County Health Services. Your job is to collect important information from the user before their doctor visit. You're talking to Chad Bailey. You should address the user by their first name and be polite and professional. You're not a medical professional, so you shouldn't provide any advice. Keep your responses short. Your job is to collect information to give to a doctor. Don't make assumptions about what values to plug into functions. Ask for clarification if a user response is ambiguous. Start by introducing yourself. Then, ask the user to confirm their identity by telling you their birthday, including the year. When they answer with their birthday, call the verify_birthday function."})
+        self._context.set_tools([
             {
                 "type": "function",
                 "function": {
@@ -74,17 +93,18 @@ class IntakeProcessor:
                     },
                 },
             }])
+        # Create an allowlist of functions that the LLM can call
+        self._functions = [
+            "verify_birthday",
+            "list_prescriptions",
+            "list_allergies",
+            "list_conditions",
+            "list_visit_reasons",
+        ]
 
-    async def verify_birthday(
-            self,
-            function_name,
-            tool_call_id,
-            args,
-            llm,
-            context,
-            result_callback):
+    async def verify_birthday(self, llm, args):
         if args["birthday"] == "1983-01-01":
-            context.set_tools(
+            self._context.set_tools(
                 [
                     {
                         "type": "function",
@@ -114,18 +134,18 @@ class IntakeProcessor:
                         },
                     }])
             # It's a bit weird to push this to the LLM, but it gets it into the pipeline
-            # await llm.push_frame(sounds["ding2.wav"], FrameDirection.DOWNSTREAM)
+            await llm.push_frame(sounds["ding2.wav"], FrameDirection.DOWNSTREAM)
             # We don't need the function call in the context, so just return a new
             # system message and let the framework re-prompt
-            await result_callback([{"role": "system", "content": "Next, thank the user for confirming their identity, then ask the user to list their current prescriptions. Each prescription needs to have a medication name and a dosage. Do not call the list_prescriptions function with any unknown dosages."}])
+            return [{"role": "system", "content": "Next, thank the user for confirming their identity, then ask the user to list their current prescriptions. Each prescription needs to have a medication name and a dosage. Do not call the list_prescriptions function with any unknown dosages."}]
         else:
             # The user provided an incorrect birthday; ask them to try again
-            await result_callback([{"role": "system", "content": "The user provided an incorrect birthday. Ask them for their birthday again. When they answer, call the verify_birthday function."}])
+            return [{"role": "system", "content": "The user provided an incorrect birthday. Ask them for their birthday again. When they answer, call the verify_birthday function."}]
 
-    async def start_prescriptions(self, function_name, llm, context):
+    async def start_prescriptions(self, llm):
         print(f"!!! doing start prescriptions")
         # Move on to allergies
-        context.set_tools(
+        self._context.set_tools(
             [
                 {
                     "type": "function",
@@ -149,18 +169,18 @@ class IntakeProcessor:
                         },
                     },
                 }])
-        context.add_message(
+        self._context.add_message(
             {
                 "role": "system",
                 "content": "Next, ask the user if they have any allergies. Once they have listed their allergies or confirmed they don't have any, call the list_allergies function."})
         print(f"!!! about to await llm process frame in start prescrpitions")
-        await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
+        await llm.process_frame(OpenAILLMContextFrame(self._context), FrameDirection.DOWNSTREAM)
         print(f"!!! past await process frame in start prescriptions")
 
-    async def start_allergies(self, function_name, llm, context):
+    async def start_allergies(self, llm):
         print("!!! doing start allergies")
         # Move on to conditions
-        context.set_tools(
+        self._context.set_tools(
             [
                 {
                     "type": "function",
@@ -185,16 +205,16 @@ class IntakeProcessor:
                     },
                 },
             ])
-        context.add_message(
+        self._context.add_message(
             {
                 "role": "system",
                 "content": "Now ask the user if they have any medical conditions the doctor should know about. Once they've answered the question, call the list_conditions function."})
-        await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
+        await llm.process_frame(OpenAILLMContextFrame(self._context), FrameDirection.DOWNSTREAM)
 
-    async def start_conditions(self, function_name, llm, context):
+    async def start_conditions(self, llm):
         print("!!! doing start conditions")
         # Move on to visit reasons
-        context.set_tools(
+        self._context.set_tools(
             [
                 {
                     "type": "function",
@@ -218,37 +238,36 @@ class IntakeProcessor:
                         },
                     },
                 }])
-        context.add_message(
-            {
-                "role": "system",
-                "content": "Finally, ask the user the reason for their doctor visit today. Once they answer, call the list_visit_reasons function."})
-        await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
+        self._context.add_message(
+            {"role": "system", "content": "Finally, ask the user the reason for their doctor visit today. Once they answer, call the list_visit_reasons function."})
+        await llm.process_frame(OpenAILLMContextFrame(self._context), FrameDirection.DOWNSTREAM)
 
-    async def start_visit_reasons(self, function_name, llm, context):
+    async def start_visit_reasons(self, llm):
         print("!!! doing start visit reasons")
         # move to finish call
-        context.set_tools([])
-        context.add_message({"role": "system",
-                             "content": "Now, thank the user and end the conversation."})
-        await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
+        self._context.set_tools([])
+        self._context.add_message({"role": "system",
+                                   "content": "Now, thank the user and end the conversation."})
+        await llm.process_frame(OpenAILLMContextFrame(self._context), FrameDirection.DOWNSTREAM)
 
-    async def save_data(self, function_name, tool_call_id, args, llm, context, result_callback):
+    async def save_data(self, llm, args):
         logger.info(f"!!! Saving data: {args}")
         # Since this is supposed to be "async", returning None from the callback
         # will prevent adding anything to context or re-prompting
-        await result_callback(None)
+        return None
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
             "Chatbot",
             DailyParams(
                 audio_out_enabled=True,
+                camera_out_enabled=True,
+                camera_out_width=1024,
+                camera_out_height=576,
                 vad_enabled=True,
                 vad_analyzer=SileroVADAnalyzer(),
                 transcription_enabled=True,
@@ -263,15 +282,20 @@ async def main():
             )
         )
 
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            #
+            # English
+            #
+            voice_id="pNInz6obpgDQGcFmaJgB",
 
-        # tts = CartesiaTTSService(
-        #     api_key=os.getenv("CARTESIA_API_KEY"),
-        #     voice_id="846d6cb0-2301-48b6-9683-48f5618ea2f6",  # Spanish-speaking Lady
-        # )
+            #
+            # Spanish
+            #
+            # model="eleven_multilingual_v2",
+            # voice_id="gD1IexrzCvsXPHUuT0s3",
+        )
 
         llm = OpenAILLMService(
             api_key=os.getenv("OPENAI_API_KEY"),
@@ -279,9 +303,10 @@ async def main():
 
         messages = []
         context = OpenAILLMContext(messages=messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        user_context = LLMUserContextAggregator(context)
+        assistant_context = LLMAssistantContextAggregator(context)
 
-        intake = IntakeProcessor(context)
+        intake = IntakeProcessor(context, llm)
         llm.register_function("verify_birthday", intake.verify_birthday)
         llm.register_function(
             "list_prescriptions",
@@ -304,12 +329,12 @@ async def main():
 
         pipeline = Pipeline([
             transport.input(),   # Transport input
-            context_aggregator.user(),  # User responses
+            user_context,        # User responses
             llm,                 # LLM
             fl,                  # Frame logger
             tts,                 # TTS
             transport.output(),  # Transport output
-            context_aggregator.assistant(),  # Assistant responses
+            assistant_context,   # Assistant responses
         ])
 
         task = PipelineTask(pipeline, PipelineParams(allow_interruptions=False))
@@ -326,4 +351,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/patient-intake/requirements.txt

@@ -1,4 +1,5 @@
 python-dotenv
+requests
 fastapi[all]
 uvicorn
 pipecat-ai[daily,openai,silero]

examples/patient-intake/runner.py

@@ -1,17 +1,11 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
 import argparse
 import os
-
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+import time
+import urllib
+import requests
 
 
-async def configure(aiohttp_session: aiohttp.ClientSession):
+def configure():
     parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
     parser.add_argument(
         "-u",
@@ -39,15 +33,26 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
     if not key:
         raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
 
-    daily_rest_helper = DailyRESTHelper(
-        daily_api_key=key,
-        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
-        aiohttp_session=aiohttp_session)
-
     # Create a meeting token for the given room with an expiration 1 hour in
     # the future.
-    expiry_time: float = 60 * 60
+    room_name: str = urllib.parse.urlparse(url).path[1:]
+    expiration: float = time.time() + 60 * 60
 
-    token = await daily_rest_helper.get_token(url, expiry_time)
+    res: requests.Response = requests.post(
+        f"https://api.daily.co/v1/meeting-tokens",
+        headers={
+            "Authorization": f"Bearer {key}"},
+        json={
+            "properties": {
+                "room_name": room_name,
+                "is_owner": True,
+                "exp": expiration}},
+    )
+
+    if res.status_code != 200:
+        raise Exception(
+            f"Failed to create meeting token: {res.status_code} {res.text}")
+
+    token: str = res.json()["token"]
 
     return (url, token)

examples/patient-intake/server.py

@@ -1,51 +1,31 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
 import os
 import argparse
 import subprocess
-
-from contextlib import asynccontextmanager
+import atexit
 
 from fastapi import FastAPI, Request, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse, RedirectResponse
 
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
+from utils.daily_helpers import create_room as _create_room, get_token
 
 MAX_BOTS_PER_ROOM = 1
 
 # Bot sub-process dict for status reporting and concurrency control
 bot_procs = {}
 
-daily_helpers = {}
-
 
 def cleanup():
     # Clean up function, just to be extra safe
-    for entry in bot_procs.values():
-        proc = entry[0]
+    for proc in bot_procs.values():
         proc.terminate()
         proc.wait()
 
 
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    aiohttp_session = aiohttp.ClientSession()
-    daily_helpers["rest"] = DailyRESTHelper(
-        daily_api_key=os.getenv("DAILY_API_KEY", ""),
-        daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
-        aiohttp_session=aiohttp_session
-    )
-    yield
-    await aiohttp_session.close()
-    cleanup()
+atexit.register(cleanup)
 
-app = FastAPI(lifespan=lifespan)
+
+app = FastAPI()
 
 app.add_middleware(
     CORSMiddleware,
@@ -59,45 +39,45 @@ app.add_middleware(
 @app.get("/start")
 async def start_agent(request: Request):
     print(f"!!! Creating room")
-    room = await daily_helpers["rest"].create_room(DailyRoomParams())
-    print(f"!!! Room URL: {room.url}")
+    room_url, room_name = _create_room()
+    print(f"!!! Room URL: {room_url}")
     # Ensure the room property is present
-    if not room.url:
+    if not room_url:
         raise HTTPException(
             status_code=500,
             detail="Missing 'room' property in request data. Cannot start agent without a target room!")
 
     # Check if there is already an existing process running in this room
     num_bots_in_room = sum(
-        1 for proc in bot_procs.values() if proc[1] == room.url and proc[0].poll() is None)
+        1 for proc in bot_procs.values() if proc[1] == room_url and proc[0].poll() is None)
     if num_bots_in_room >= MAX_BOTS_PER_ROOM:
         raise HTTPException(
-            status_code=500, detail=f"Max bot limited reach for room: {room.url}")
+            status_code=500, detail=f"Max bot limited reach for room: {room_url}")
 
     # Get the token for the room
-    token = await daily_helpers["rest"].get_token(room.url)
+    token = get_token(room_url)
 
     if not token:
         raise HTTPException(
-            status_code=500, detail=f"Failed to get token for room: {room.url}")
+            status_code=500, detail=f"Failed to get token for room: {room_url}")
 
     # Spawn a new agent, and join the user session
     # Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
     try:
         proc = subprocess.Popen(
             [
-                f"python3 -m bot -u {room.url} -t {token}"
+                f"python3 -m bot -u {room_url} -t {token}"
             ],
             shell=True,
             bufsize=1,
             cwd=os.path.dirname(os.path.abspath(__file__))
         )
-        bot_procs[proc.pid] = (proc, room.url)
+        bot_procs[proc.pid] = (proc, room_url)
     except Exception as e:
         raise HTTPException(
             status_code=500, detail=f"Failed to start subprocess: {e}")
 
-    return RedirectResponse(room.url)
+    return RedirectResponse(room_url)
 
 
 @app.get("/status/{pid}")

examples/patient-intake/utils/daily_helpers.py

@@ -0,0 +1,109 @@
+
+import urllib.parse
+import os
+import time
+import urllib
+import requests
+
+from dotenv import load_dotenv
+load_dotenv()
+
+
+daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
+daily_api_key = os.getenv("DAILY_API_KEY")
+
+
+def create_room() -> tuple[str, str]:
+    """
+    Helper function to create a Daily room.
+    # See: https://docs.daily.co/reference/rest-api/rooms
+
+    Returns:
+        tuple: A tuple containing the room URL and room name.
+
+    Raises:
+        Exception: If the request to create the room fails or if the response does not contain the room URL or room name.
+    """
+    room_props = {
+        "exp": time.time() + 60 * 60,  # 1 hour
+        "enable_chat": True,
+        "enable_emoji_reactions": True,
+        "eject_at_room_exp": True,
+        "enable_prejoin_ui": False,  # Important for the bot to be able to join headlessly
+    }
+    res = requests.post(
+        f"https://{daily_api_path}/rooms",
+        headers={"Authorization": f"Bearer {daily_api_key}"},
+        json={
+            "properties": room_props
+        },
+    )
+    if res.status_code != 200:
+        raise Exception(f"Unable to create room: {res.text}")
+
+    data = res.json()
+    room_url: str = data.get("url")
+    room_name: str = data.get("name")
+    if room_url is None or room_name is None:
+        raise Exception("Missing room URL or room name in response")
+
+    return room_url, room_name
+
+
+def get_name_from_url(room_url: str) -> str:
+    """
+    Extracts the name from a given room URL.
+
+    Args:
+        room_url (str): The URL of the room.
+
+    Returns:
+        str: The extracted name from the room URL.
+    """
+    return urllib.parse.urlparse(room_url).path[1:]
+
+
+def get_token(room_url: str) -> str:
+    """
+    Retrieves a meeting token for the specified Daily room URL.
+    # See: https://docs.daily.co/reference/rest-api/meeting-tokens
+
+    Args:
+        room_url (str): The URL of the Daily room.
+
+    Returns:
+        str: The meeting token.
+
+    Raises:
+        Exception: If no room URL is specified or if no Daily API key is specified.
+        Exception: If there is an error creating the meeting token.
+    """
+    if not room_url:
+        raise Exception(
+            "No Daily room specified. You must specify a Daily room in order a token to be generated.")
+
+    if not daily_api_key:
+        raise Exception(
+            "No Daily API key specified. set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
+
+    expiration: float = time.time() + 60 * 60
+    room_name = get_name_from_url(room_url)
+
+    res: requests.Response = requests.post(
+        f"https://{daily_api_path}/meeting-tokens",
+        headers={
+            "Authorization": f"Bearer {daily_api_key}"},
+        json={
+            "properties": {
+                "room_name": room_name,
+                "is_owner": True,  # Owner tokens required for transcription
+                "exp": expiration}},
+    )
+
+    if res.status_code != 200:
+        raise Exception(
+            f"Failed to create meeting token: {res.status_code} {res.text}")
+
+    token: str = res.json()["token"]
+
+    return token

examples/simple-chatbot/bot.py

@@ -1,9 +1,3 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
 import asyncio
 import aiohttp
 import os
@@ -26,7 +20,7 @@ from pipecat.frames.frames import (
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.transports.services.daily import DailyParams, DailyTranscriptionSettings, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
@@ -83,10 +77,8 @@ class TalkingAnimation(FrameProcessor):
         await self.push_frame(frame)
 
 
-async def main():
+async def main(room_url: str, token):
     async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
         transport = DailyTransport(
             room_url,
             token,
@@ -111,6 +103,7 @@ async def main():
         )
 
         tts = ElevenLabsTTSService(
+            aiohttp_session=session,
             api_key=os.getenv("ELEVENLABS_API_KEY"),
             #
             # English
@@ -172,4 +165,5 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    (url, token) = configure()
+    asyncio.run(main(url, token))

examples/simple-chatbot/requirements.txt

@@ -1,4 +1,5 @@
 python-dotenv
+requests
 fastapi[all]
 uvicorn
 pipecat-ai[daily,openai,silero]

examples/simple-chatbot/runner.py

@@ -1,17 +1,11 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
 import argparse
 import os
-
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+import time
+import urllib
+import requests
 
 
-async def configure(aiohttp_session: aiohttp.ClientSession):
+def configure():
     parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
     parser.add_argument(
         "-u",
@@ -39,16 +33,26 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
     if not key:
         raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
 
-    daily_rest_helper = DailyRESTHelper(
-        daily_api_key=key,
-        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
-        aiohttp_session=aiohttp_session
-    )
-
     # Create a meeting token for the given room with an expiration 1 hour in
     # the future.
-    expiry_time: float = 60 * 60
+    room_name: str = urllib.parse.urlparse(url).path[1:]
+    expiration: float = time.time() + 60 * 60
 
-    token = await daily_rest_helper.get_token(url, expiry_time)
+    res: requests.Response = requests.post(
+        f"https://api.daily.co/v1/meeting-tokens",
+        headers={
+            "Authorization": f"Bearer {key}"},
+        json={
+            "properties": {
+                "room_name": room_name,
+                "is_owner": True,
+                "exp": expiration}},
+    )
+
+    if res.status_code != 200:
+        raise Exception(
+            f"Failed to create meeting token: {res.status_code} {res.text}")
+
+    token: str = res.json()["token"]
 
     return (url, token)

examples/simple-chatbot/server.py

@@ -1,51 +1,31 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
 import os
 import argparse
 import subprocess
-
-from contextlib import asynccontextmanager
+import atexit
 
 from fastapi import FastAPI, Request, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse, RedirectResponse
 
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
+from utils.daily_helpers import create_room as _create_room, get_token
 
 MAX_BOTS_PER_ROOM = 1
 
 # Bot sub-process dict for status reporting and concurrency control
 bot_procs = {}
 
-daily_helpers = {}
-
 
 def cleanup():
     # Clean up function, just to be extra safe
-    for entry in bot_procs.values():
-        proc = entry[0]
+    for proc in bot_procs.values():
         proc.terminate()
         proc.wait()
 
 
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    aiohttp_session = aiohttp.ClientSession()
-    daily_helpers["rest"] = DailyRESTHelper(
-        daily_api_key=os.getenv("DAILY_API_KEY", ""),
-        daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
-        aiohttp_session=aiohttp_session
-    )
-    yield
-    await aiohttp_session.close()
-    cleanup()
+atexit.register(cleanup)
 
-app = FastAPI(lifespan=lifespan)
+
+app = FastAPI()
 
 app.add_middleware(
     CORSMiddleware,
@@ -59,45 +39,45 @@ app.add_middleware(
 @app.get("/start")
 async def start_agent(request: Request):
     print(f"!!! Creating room")
-    room = await daily_helpers["rest"].create_room(DailyRoomParams())
-    print(f"!!! Room URL: {room.url}")
+    room_url, room_name = _create_room()
+    print(f"!!! Room URL: {room_url}")
     # Ensure the room property is present
-    if not room.url:
+    if not room_url:
         raise HTTPException(
             status_code=500,
             detail="Missing 'room' property in request data. Cannot start agent without a target room!")
 
     # Check if there is already an existing process running in this room
     num_bots_in_room = sum(
-        1 for proc in bot_procs.values() if proc[1] == room.url and proc[0].poll() is None)
+        1 for proc in bot_procs.values() if proc[1] == room_url and proc[0].poll() is None)
     if num_bots_in_room >= MAX_BOTS_PER_ROOM:
         raise HTTPException(
-            status_code=500, detail=f"Max bot limited reach for room: {room.url}")
+            status_code=500, detail=f"Max bot limited reach for room: {room_url}")
 
     # Get the token for the room
-    token = await daily_helpers["rest"].get_token(room.url)
+    token = get_token(room_url)
 
     if not token:
         raise HTTPException(
-            status_code=500, detail=f"Failed to get token for room: {room.url}")
+            status_code=500, detail=f"Failed to get token for room: {room_url}")
 
     # Spawn a new agent, and join the user session
     # Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
     try:
         proc = subprocess.Popen(
             [
-                f"python3 -m bot -u {room.url} -t {token}"
+                f"python3 -m bot -u {room_url} -t {token}"
             ],
             shell=True,
             bufsize=1,
             cwd=os.path.dirname(os.path.abspath(__file__))
         )
-        bot_procs[proc.pid] = (proc, room.url)
+        bot_procs[proc.pid] = (proc, room_url)
     except Exception as e:
         raise HTTPException(
             status_code=500, detail=f"Failed to start subprocess: {e}")
 
-    return RedirectResponse(room.url)
+    return RedirectResponse(room_url)
 
 
 @app.get("/status/{pid}")

examples/simple-chatbot/utils/daily_helpers.py

@@ -0,0 +1,109 @@
+
+import urllib.parse
+import os
+import time
+import urllib
+import requests
+
+from dotenv import load_dotenv
+load_dotenv()
+
+
+daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
+daily_api_key = os.getenv("DAILY_API_KEY")
+
+
+def create_room() -> tuple[str, str]:
+    """
+    Helper function to create a Daily room.
+    # See: https://docs.daily.co/reference/rest-api/rooms
+
+    Returns:
+        tuple: A tuple containing the room URL and room name.
+
+    Raises:
+        Exception: If the request to create the room fails or if the response does not contain the room URL or room name.
+    """
+    room_props = {
+        "exp": time.time() + 60 * 60,  # 1 hour
+        "enable_chat": True,
+        "enable_emoji_reactions": True,
+        "eject_at_room_exp": True,
+        "enable_prejoin_ui": False,  # Important for the bot to be able to join headlessly
+    }
+    res = requests.post(
+        f"https://{daily_api_path}/rooms",
+        headers={"Authorization": f"Bearer {daily_api_key}"},
+        json={
+            "properties": room_props
+        },
+    )
+    if res.status_code != 200:
+        raise Exception(f"Unable to create room: {res.text}")
+
+    data = res.json()
+    room_url: str = data.get("url")
+    room_name: str = data.get("name")
+    if room_url is None or room_name is None:
+        raise Exception("Missing room URL or room name in response")
+
+    return room_url, room_name
+
+
+def get_name_from_url(room_url: str) -> str:
+    """
+    Extracts the name from a given room URL.
+
+    Args:
+        room_url (str): The URL of the room.
+
+    Returns:
+        str: The extracted name from the room URL.
+    """
+    return urllib.parse.urlparse(room_url).path[1:]
+
+
+def get_token(room_url: str) -> str:
+    """
+    Retrieves a meeting token for the specified Daily room URL.
+    # See: https://docs.daily.co/reference/rest-api/meeting-tokens
+
+    Args:
+        room_url (str): The URL of the Daily room.
+
+    Returns:
+        str: The meeting token.
+
+    Raises:
+        Exception: If no room URL is specified or if no Daily API key is specified.
+        Exception: If there is an error creating the meeting token.
+    """
+    if not room_url:
+        raise Exception(
+            "No Daily room specified. You must specify a Daily room in order a token to be generated.")
+
+    if not daily_api_key:
+        raise Exception(
+            "No Daily API key specified. set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
+
+    expiration: float = time.time() + 60 * 60
+    room_name = get_name_from_url(room_url)
+
+    res: requests.Response = requests.post(
+        f"https://{daily_api_path}/meeting-tokens",
+        headers={
+            "Authorization": f"Bearer {daily_api_key}"},
+        json={
+            "properties": {
+                "room_name": room_name,
+                "is_owner": True,  # Owner tokens required for transcription
+                "exp": expiration}},
+    )
+
+    if res.status_code != 200:
+        raise Exception(
+            f"Failed to create meeting token: {res.status_code} {res.text}")
+
+    token: str = res.json()["token"]
+
+    return token

examples/storytelling-chatbot/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.11-slim-bookworm
+FROM python:3.11-bullseye
 
 ARG DEBIAN_FRONTEND=noninteractive
 ARG USE_PERSISTENT_DATA
@@ -51,4 +51,4 @@ COPY --chown=user ./frontend/ frontend/
 RUN cd frontend && npm install && npm run build
 
 # Start the FastAPI server
-CMD python3 src/bot_runner.py --port ${FAST_API_PORT}
+CMD python3 src/server.py --port ${FAST_API_PORT}

examples/storytelling-chatbot/README.md

@@ -48,8 +48,6 @@ pip install -r requirements.txt
 mv env.example .env
 ```
 
-When deploying to production, to ensure only this app can spawn a new bot, set your `ENV` to `production`
-
 **Build the frontend:**
 
 This project uses a custom frontend, which needs to built. Note: this is done automatically as part of the Docker deployment.
@@ -66,11 +64,11 @@ The build UI files can be found in `frontend/out`
 
 Start the API / bot manager:
 
-`python src/bot_runner.py`
+`python src/server.py`
 
 If you'd like to run a custom domain or port:
 
-`python src/bot_runner.py --host somehost --p someport`
+`python src/server.py --host somehost --p 7777`
 
 ➡️ Open the host URL in your browser `http://localhost:7860`
 

examples/storytelling-chatbot/env.example

@@ -1,9 +1,5 @@
-DAILY_API_KEY=
-DAILY_SAMPLE_ROOM_URL=
-ELEVENLABS_API_KEY=
-ELEVENLABS_VOICE_ID=
-FAL_KEY=
-OPENAI_API_KEY=
-
-ENV= # dev | production
-RUN_AS_VM= # Set this if you want to run bots on process (not launch a new VM)
+DAILY_API_KEY=7df...
+ELEVENLABS_API_KEY=aeb...
+ELEVENLABS_VOICE_ID=7S...
+FAL_KEY=8c...
+OPENAI_API_KEY=sk-PL...

examples/storytelling-chatbot/frontend/components/App.tsx

@@ -27,11 +27,14 @@ export default function Call() {
 
     // Create a new room for the story session
     try {
-      const response = await fetch("/start_bot", {
+      const response = await fetch("/create", {
         method: "POST",
         headers: {
           "Content-Type": "application/json",
         },
+        body: JSON.stringify({
+          room_url: process.env.NEXT_PUBLIC_ROOM_URL || null,
+        }),
       });
 
       const { room_url, token } = await response.json();
@@ -52,9 +55,21 @@ export default function Call() {
       // Disable local audio, the bot will say hello first
       daily.setLocalAudio(false);
 
+      // Start the bot
+      const resp = await fetch("/start", {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify({
+          room_url,
+        }),
+      });
+
       setState("started");
     } catch (error) {
       setState("error");
+      leave();
     }
   }
 
@@ -64,13 +79,7 @@ export default function Call() {
   }
 
   if (state === "error") {
-    return (
-      <div className="flex items-center mx-auto">
-        <p className="text-red-500 font-semibold bg-white px-4 py-2 shadow-xl rounded-lg">
-          This demo is currently at capacity. Please try again later.
-        </p>
-      </div>
-    );
+    return <div>An Error occured</div>;
   }
 
   if (state === "started") {

examples/storytelling-chatbot/frontend/components/DevicePicker/index.tsx

@@ -108,26 +108,26 @@ export default function DevicePicker({}: Props) {
       {hasMicError && (
         <div className="error">
           {micState === "blocked" ? (
-            <p className="text-red-500">
+            <p>
               Please check your browser and system permissions. Make sure that
               this app is allowed to access your microphone.
             </p>
           ) : micState === "in-use" ? (
-            <p className="text-red-500">
+            <p>
               Your microphone is being used by another app. Please close any
               other apps using your microphone and restart this app.
             </p>
           ) : micState === "not-found" ? (
-            <p className="text-red-500">
+            <p>
               No microphone seems to be connected. Please connect a microphone.
             </p>
           ) : micState === "not-supported" ? (
-            <p className="text-red-500">
+            <p>
               This app is not supported on your device. Please update your
               software or use a different device.
             </p>
           ) : (
-            <p className="text-red-500">
+            <p>
               There seems to be an issue accessing your microphone. Try
               restarting the app or consult a system administrator.
             </p>

examples/storytelling-chatbot/frontend/components/Setup.tsx

@@ -1,7 +1,7 @@
 import React from "react";
 import { Button } from "@/components/ui/button";
 import DevicePicker from "@/components/DevicePicker";
-import { IconAlertCircle, IconEar, IconLoader2 } from "@tabler/icons-react";
+import { IconEar, IconLoader2 } from "@tabler/icons-react";
 
 type SetupProps = {
   handleStart: () => void;
@@ -24,6 +24,7 @@ export const Setup: React.FC<SetupProps> = ({ handleStart }) => {
           <h1 className="text-4xl font-bold text-pretty tracking-tighter mb-4">
             Welcome to <span className="text-sky-500">Storytime</span>
           </h1>
+
           {state === "intro" ? (
             <>
               <p className="text-gray-600 leading-relaxed text-pretty">
@@ -37,9 +38,6 @@ export const Setup: React.FC<SetupProps> = ({ handleStart }) => {
                 <IconEar size={24} /> For best results, try in a quiet
                 environment!
               </p>
-              <p className="flex flex-row gap-2 text-gray-600 font-medium text-red-500">
-                <IconAlertCircle size={24} /> This demo expires after 5 minutes.
-              </p>
             </>
           ) : (
             <>
@@ -51,6 +49,7 @@ export const Setup: React.FC<SetupProps> = ({ handleStart }) => {
               <DevicePicker />
             </>
           )}
+
           <hr className="border-gray-150 my-2" />
 
           <Button

examples/storytelling-chatbot/frontend/env.example

@@ -1 +1,2 @@
+NEXT_PUBLIC_ROOM_URL=
 SITE_URL=

examples/storytelling-chatbot/requirements.txt

@@ -1,5 +1,6 @@
 async_timeout
 fastapi
 uvicorn
+requests
 python-dotenv
 pipecat-ai[daily,openai,fal]

examples/storytelling-chatbot/src/bot.py

@@ -5,7 +5,7 @@ import os
 import sys
 
 
-from pipecat.frames.frames import LLMMessagesFrame, StopTaskFrame, EndFrame
+from pipecat.frames.frames import LLMMessagesFrame, StopTaskFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
@@ -60,6 +60,7 @@ async def main(room_url, token=None):
         )
 
         tts_service = ElevenLabsTTSService(
+            aiohttp_session=session,
             api_key=os.getenv("ELEVENLABS_API_KEY"),
             voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )
@@ -138,16 +139,6 @@ async def main(room_url, token=None):
 
         main_task = PipelineTask(main_pipeline)
 
-        @transport.event_handler("on_participant_left")
-        async def on_participant_left(transport, participant, reason):
-            intro_task.queue_frame(EndFrame())
-            await main_task.queue_frame(EndFrame())
-
-        @transport.event_handler("on_call_state_updated")
-        async def on_call_state_updated(transport, state):
-            if state == "left":
-                await main_task.queue_frame(EndFrame())
-
         await runner.run(main_task)
 
 if __name__ == "__main__":

examples/storytelling-chatbot/src/bot_runner.py

@@ -1,251 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
-import argparse
-import subprocess
-import os
-
-from pathlib import Path
-from typing import Optional
-
-from contextlib import asynccontextmanager
-
-from fastapi import FastAPI, Request, HTTPException
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.staticfiles import StaticFiles
-from fastapi.responses import FileResponse, JSONResponse
-
-from pipecat.transports.services.helpers.daily_rest import (
-    DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomParams)
-
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-# ------------ Fast API Config ------------ #
-
-MAX_SESSION_TIME = 5 * 60  # 5 minutes
-
-daily_helpers = {}
-
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    aiohttp_session = aiohttp.ClientSession()
-    daily_helpers["rest"] = DailyRESTHelper(
-        daily_api_key=os.getenv("DAILY_API_KEY", ""),
-        daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
-        aiohttp_session=aiohttp_session
-    )
-    yield
-    await aiohttp_session.close()
-
-app = FastAPI(lifespan=lifespan)
-
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# Mount the static directory
-STATIC_DIR = "frontend/out"
-
-
-# ------------ Fast API Routes ------------ #
-
-app.mount("/static", StaticFiles(directory=STATIC_DIR, html=True), name="static")
-
-
-@app.post("/start_bot")
-async def start_bot(request: Request) -> JSONResponse:
-    if os.getenv("ENV", "dev") == "production":
-        # Only allow requests from the specified domain
-        host_header = request.headers.get("host")
-        allowed_domains = ["storytelling-chatbot.fly.dev", "www.storytelling-chatbot.fly.dev"]
-        # Check if the Host header matches the allowed domain
-        if host_header not in allowed_domains:
-            raise HTTPException(status_code=403, detail="Access denied")
-
-    try:
-        data = await request.json()
-        # Is this a webhook creation request?
-        if "test" in data:
-            return JSONResponse({"test": True})
-    except Exception as e:
-        pass
-
-    # Use specified room URL, or create a new one if not specified
-    room_url = os.getenv("DAILY_SAMPLE_ROOM_URL", "")
-
-    if not room_url:
-        params = DailyRoomParams(
-            properties=DailyRoomProperties()
-        )
-        try:
-            room: DailyRoomObject = await daily_helpers["rest"].create_room(params=params)
-        except Exception as e:
-            raise HTTPException(
-                status_code=500,
-                detail=f"Unable to provision room {e}")
-    else:
-        # Check passed room URL exists, we should assume that it already has a sip set up
-        try:
-            room: DailyRoomObject = await daily_helpers["rest"].get_room_from_url(room_url)
-        except Exception:
-            raise HTTPException(
-                status_code=500, detail=f"Room not found: {room_url}")
-
-    # Give the agent a token to join the session
-    token = await daily_helpers["rest"].get_token(room.url, MAX_SESSION_TIME)
-
-    if not room or not token:
-        raise HTTPException(
-            status_code=500, detail=f"Failed to get token for room: {room_url}")
-
-    # Launch a new VM, or run as a shell process (not recommended)
-    if os.getenv("RUN_AS_VM", False):
-        try:
-            await virtualize_bot(room.url, token)
-        except Exception as e:
-            raise HTTPException(
-                status_code=500, detail=f"Failed to spawn VM: {e}")
-    else:
-        try:
-            subprocess.Popen(
-                [f"python3 -m bot -u {room.url} -t {token}"],
-                shell=True,
-                bufsize=1,
-                cwd=os.path.dirname(os.path.abspath(__file__)))
-        except Exception as e:
-            raise HTTPException(
-                status_code=500, detail=f"Failed to start subprocess: {e}")
-
-    # Grab a token for the user to join with
-    user_token = await daily_helpers["rest"].get_token(room.url, MAX_SESSION_TIME)
-
-    return JSONResponse({
-        "room_url": room.url,
-        "token": user_token,
-    })
-
-
-@app.get("/{path_name:path}", response_class=FileResponse)
-async def catch_all(path_name: Optional[str] = ""):
-    if path_name == "":
-        return FileResponse(f"{STATIC_DIR}/index.html")
-
-    file_path = Path(STATIC_DIR) / (path_name or "")
-
-    if file_path.is_file():
-        return file_path
-
-    html_file_path = file_path.with_suffix(".html")
-    if html_file_path.is_file():
-        return FileResponse(html_file_path)
-
-    raise HTTPException(status_code=450, detail="Incorrect API call")
-
-
-# ------------ Virtualization ------------ #
-
-async def virtualize_bot(room_url: str, token: str):
-    """
-    This is an example of how to virtualize the bot using Fly.io
-    You can adapt this method to use whichever cloud provider you prefer.
-    """
-    FLY_API_HOST = os.getenv("FLY_API_HOST", "https://api.machines.dev/v1")
-    FLY_APP_NAME = os.getenv("FLY_APP_NAME", "storytelling-chatbot")
-    FLY_API_KEY = os.getenv("FLY_API_KEY", "")
-    FLY_HEADERS = {
-        'Authorization': f"Bearer {FLY_API_KEY}",
-        'Content-Type': 'application/json'
-    }
-
-    async with aiohttp.ClientSession() as session:
-        # Use the same image as the bot runner
-        async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS) as r:
-            if r.status != 200:
-                text = await r.text()
-                raise Exception(f"Unable to get machine info from Fly: {text}")
-
-            data = await r.json()
-            image = data[0]['config']['image']
-
-        # Machine configuration
-        cmd = f"python3 src/bot.py -u {room_url} -t {token}"
-        cmd = cmd.split()
-        worker_props = {
-            "config": {
-                "image": image,
-                "auto_destroy": True,
-                "init": {
-                    "cmd": cmd
-                },
-                "restart": {
-                    "policy": "no"
-                },
-                "guest": {
-                    "cpu_kind": "shared",
-                    "cpus": 1,
-                    "memory_mb": 512
-                }
-            },
-        }
-
-        # Spawn a new machine instance
-        async with session.post(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS, json=worker_props) as r:
-            if r.status != 200:
-                text = await r.text()
-                raise Exception(f"Problem starting a bot worker: {text}")
-
-            data = await r.json()
-            # Wait for the machine to enter the started state
-            vm_id = data['id']
-
-        async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines/{vm_id}/wait?state=started", headers=FLY_HEADERS) as r:
-            if r.status != 200:
-                text = await r.text()
-                raise Exception(f"Bot was unable to enter started state: {text}")
-
-        print(f"Machine joined room: {room_url}")
-
-
-# ------------ Main ------------ #
-
-if __name__ == "__main__":
-    # Check environment variables
-    required_env_vars = ['OPENAI_API_KEY', 'DAILY_API_KEY',
-                         'FAL_KEY', 'ELEVENLABS_VOICE_ID', 'ELEVENLABS_API_KEY']
-    for env_var in required_env_vars:
-        if env_var not in os.environ:
-            raise Exception(f"Missing environment variable: {env_var}.")
-
-    import uvicorn
-
-    default_host = os.getenv("HOST", "0.0.0.0")
-    default_port = int(os.getenv("FAST_API_PORT", "7860"))
-
-    parser = argparse.ArgumentParser(
-        description="Daily Storyteller FastAPI server")
-    parser.add_argument("--host", type=str,
-                        default=default_host, help="Host address")
-    parser.add_argument("--port", type=int,
-                        default=default_port, help="Port number")
-    parser.add_argument("--reload", action="store_true",
-                        help="Reload code on change")
-
-    config = parser.parse_args()
-
-    uvicorn.run(
-        "bot_runner:app",
-        host=config.host,
-        port=config.port,
-        reload=config.reload
-    )

examples/storytelling-chatbot/src/server.py

@@ -0,0 +1,175 @@
+import os
+import argparse
+import subprocess
+import atexit
+from pathlib import Path
+from typing import Optional
+
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from fastapi.responses import FileResponse, JSONResponse
+
+from utils.daily_helpers import create_room as _create_room, get_token, get_name_from_url
+
+MAX_BOTS_PER_ROOM = 1
+
+# Bot sub-process dict for status reporting and concurrency control
+bot_procs = {}
+
+
+def cleanup():
+    # Clean up function, just to be extra safe
+    for proc in bot_procs.values():
+        proc.terminate()
+        proc.wait()
+
+
+atexit.register(cleanup)
+
+
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Mount the static directory
+STATIC_DIR = "frontend/out"
+
+app.mount("/static", StaticFiles(directory=STATIC_DIR, html=True), name="static")
+
+
+@app.post("/create")
+async def create_room(request: Request) -> JSONResponse:
+    data = await request.json()
+
+    if data.get('room_url') is not None:
+        room_url = data.get('room_url')
+        room_name = get_name_from_url(room_url)
+    else:
+        room_url, room_name = _create_room()
+
+    token = get_token(room_url)
+
+    return JSONResponse({"room_url": room_url, "room_name": room_name, "token": token})
+
+
+@app.post("/start")
+async def start_agent(request: Request) -> JSONResponse:
+    data = await request.json()
+
+    # Is this a webhook creation request?
+    if "test" in data:
+        return JSONResponse({"test": True})
+
+    # Ensure the room property is present
+    room_url = data.get('room_url')
+    if not room_url:
+        raise HTTPException(
+            status_code=500,
+            detail="Missing 'room' property in request data. Cannot start agent without a target room!")
+
+    # Check if there is already an existing process running in this room
+    num_bots_in_room = sum(
+        1 for proc in bot_procs.values() if proc[1] == room_url and proc[0].poll() is None)
+    if num_bots_in_room >= MAX_BOTS_PER_ROOM:
+        raise HTTPException(
+            status_code=500, detail=f"Max bot limited reach for room: {room_url}")
+
+    # Get the token for the room
+    token = get_token(room_url)
+
+    if not token:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to get token for room: {room_url}")
+
+    # Spawn a new agent, and join the user session
+    # Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
+    try:
+        proc = subprocess.Popen(
+            [
+                f"python3 -m bot -u {room_url} -t {token}"
+            ],
+            shell=True,
+            bufsize=1,
+            cwd=os.path.dirname(os.path.abspath(__file__))
+        )
+        bot_procs[proc.pid] = (proc, room_url)
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to start subprocess: {e}")
+
+    return JSONResponse({"bot_id": proc.pid, "room_url": room_url})
+
+
+@app.get("/status/{pid}")
+def get_status(pid: int):
+    # Look up the subprocess
+    proc = bot_procs.get(pid)
+
+    # If the subprocess doesn't exist, return an error
+    if not proc:
+        raise HTTPException(
+            status_code=404, detail=f"Bot with process id: {pid} not found")
+
+    # Check the status of the subprocess
+    if proc[0].poll() is None:
+        status = "running"
+    else:
+        status = "finished"
+
+    return JSONResponse({"bot_id": pid, "status": status})
+
+
+@app.get("/{path_name:path}", response_class=FileResponse)
+async def catch_all(path_name: Optional[str] = ""):
+    if path_name == "":
+        return FileResponse(f"{STATIC_DIR}/index.html")
+
+    file_path = Path(STATIC_DIR) / (path_name or "")
+
+    if file_path.is_file():
+        return file_path
+
+    html_file_path = file_path.with_suffix(".html")
+    if html_file_path.is_file():
+        return FileResponse(html_file_path)
+
+    raise HTTPException(status_code=450, detail="Incorrect API call")
+
+
+if __name__ == "__main__":
+    # Check environment variables
+    required_env_vars = ['OPENAI_API_KEY', 'DAILY_API_KEY',
+                         'FAL_KEY', 'ELEVENLABS_VOICE_ID', 'ELEVENLABS_API_KEY']
+    for env_var in required_env_vars:
+        if env_var not in os.environ:
+            raise Exception(f"Missing environment variable: {env_var}.")
+
+    import uvicorn
+
+    default_host = os.getenv("HOST", "0.0.0.0")
+    default_port = int(os.getenv("FAST_API_PORT", "7860"))
+
+    parser = argparse.ArgumentParser(
+        description="Daily Storyteller FastAPI server")
+    parser.add_argument("--host", type=str,
+                        default=default_host, help="Host address")
+    parser.add_argument("--port", type=int,
+                        default=default_port, help="Port number")
+    parser.add_argument("--reload", action="store_true",
+                        help="Reload code on change")
+
+    config = parser.parse_args()
+
+    uvicorn.run(
+        "server:app",
+        host=config.host,
+        port=config.port,
+        reload=config.reload
+    )

examples/storytelling-chatbot/src/utils/daily_helpers.py

@@ -0,0 +1,109 @@
+
+import urllib.parse
+import os
+import time
+import urllib
+import requests
+
+from dotenv import load_dotenv
+load_dotenv()
+
+
+daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
+daily_api_key = os.getenv("DAILY_API_KEY")
+
+
+def create_room() -> tuple[str, str]:
+    """
+    Helper function to create a Daily room.
+    # See: https://docs.daily.co/reference/rest-api/rooms
+
+    Returns:
+        tuple: A tuple containing the room URL and room name.
+
+    Raises:
+        Exception: If the request to create the room fails or if the response does not contain the room URL or room name.
+    """
+    room_props = {
+        "exp": time.time() + 60 * 60,  # 1 hour
+        "enable_chat": True,
+        "enable_emoji_reactions": True,
+        "eject_at_room_exp": True,
+        "enable_prejoin_ui": False,  # Important for the bot to be able to join headlessly
+    }
+    res = requests.post(
+        f"https://{daily_api_path}/rooms",
+        headers={"Authorization": f"Bearer {daily_api_key}"},
+        json={
+            "properties": room_props
+        },
+    )
+    if res.status_code != 200:
+        raise Exception(f"Unable to create room: {res.text}")
+
+    data = res.json()
+    room_url: str = data.get("url")
+    room_name: str = data.get("name")
+    if room_url is None or room_name is None:
+        raise Exception("Missing room URL or room name in response")
+
+    return room_url, room_name
+
+
+def get_name_from_url(room_url: str) -> str:
+    """
+    Extracts the name from a given room URL.
+
+    Args:
+        room_url (str): The URL of the room.
+
+    Returns:
+        str: The extracted name from the room URL.
+    """
+    return urllib.parse.urlparse(room_url).path[1:]
+
+
+def get_token(room_url: str) -> str:
+    """
+    Retrieves a meeting token for the specified Daily room URL.
+    # See: https://docs.daily.co/reference/rest-api/meeting-tokens
+
+    Args:
+        room_url (str): The URL of the Daily room.
+
+    Returns:
+        str: The meeting token.
+
+    Raises:
+        Exception: If no room URL is specified or if no Daily API key is specified.
+        Exception: If there is an error creating the meeting token.
+    """
+    if not room_url:
+        raise Exception(
+            "No Daily room specified. You must specify a Daily room in order a token to be generated.")
+
+    if not daily_api_key:
+        raise Exception(
+            "No Daily API key specified. set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
+
+    expiration: float = time.time() + 60 * 60
+    room_name = get_name_from_url(room_url)
+
+    res: requests.Response = requests.post(
+        f"https://{daily_api_path}/meeting-tokens",
+        headers={
+            "Authorization": f"Bearer {daily_api_key}"},
+        json={
+            "properties": {
+                "room_name": room_name,
+                "is_owner": True,  # Owner tokens required for transcription
+                "exp": expiration}},
+    )
+
+    if res.status_code != 200:
+        raise Exception(
+            f"Failed to create meeting token: {res.status_code} {res.text}")
+
+    token: str = res.json()["token"]
+
+    return token

examples/studypal/README.md

@@ -1,13 +0,0 @@
-# studypal
-
-### Have a conversation about any article on the web
-
-studypal is a fast conversational AI built using [Daily](https://www.daily.co/) for real-time media transport and [Cartesia](https://cartesia.ai) for text-to-speech. Everything is orchestrated together (VAD -> STT -> LLM -> TTS) using [Pipecat](https://www.pipecat.ai/).
-
-## Setup
-
-1. Clone the repository
-2. Copy `env.example` to a `.env` file and add API keys
-3. Install the required packages: `pip install -r requirements.txt`
-4. Run `python3 studypal.py` from your command line.
-5. While the app is running, go to the `https://<yourdomain>.daily.co/<room_url>` set in `DAILY_SAMPLE_ROOM_URL` and talk to studypal!