Update CHANGELOG

Add Whisper STT service using OpenAI API

CHANGELOG.md

@@ -5,6 +5,63 @@ 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
+
+- Added `OpenAISTTService` which brings OpenAI's hosted Whisper to Pipecat.
+
+### Changed
+
+- Updated foundational example `07g-interruptible-openai.py` to use
+  `OpenAISTTService`.
+
+- `RTVIObserver` doesn't handle `LLMSearchResponseFrame` frames anymore. For
+  now, to handle those frames you need to create a `GoogleRTVIObserver` instead.
+
+### Deprecated
+
+- `RTVI.observer()` is now deprecated, instantiate an `RTVIObserver` directly
+  instead.
+
+- All RTVI frame processors (e.g. `RTVISpeakingProcessor`,
+  `RTVIBotLLMProcessor`) are now deprecated, instantiate an `RTVIObserver`
+  instead.
+
+## [0.0.56] - 2025-02-06
+
+### Changed
+
+- Use `gemini-2.0-flash-001` as the default model for `GoogleLLMSerivce`.
+
+- Improved foundational examples 22b, 22c, and 22d to support function calling.
+  With these base examples, `FunctionCallInProgressFrame` and
+  `FunctionCallResultFrame` will no longer be blocked by the gates.
+
+### Fixed
+
+- Fixed a `TkLocalTransport` and `LocalAudioTransport` issues that was causing
+  errors on cleanup.
+
+- Fixed an issue that was causing `tests.utils` import to fail because of
+  logging setup.
+
+- Fixed a `SentryMetrics` issue that was preventing any metrics to be sent to
+  Sentry and also was preventing from metrics frames to be pushed to the pipeline.
+
+- Fixed an issue in `BaseOutputTransport` where incoming audio would not be
+  resampled to the desired output sample rate.
+
+- Fixed an issue with the `TwilioFrameSerializer` and `TelnyxFrameSerializer`
+  where `twilio_sample_rate` and `telnyx_sample_rate` were incorrectly
+  initialized to `audio_in_sample_rate`. Those values currently default to 8000
+  and should be set manually from the serializer constructor if a different
+  value is needed.
+
+### Other
+
+- Added a new `sentry-metrics` example.
+
 ## [0.0.55] - 2025-02-05
 
 ### Added
@@ -83,7 +140,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `AudioBufferProcessor.reset_audio_buffers()` has been removed, use
   `AudioBufferProcessor.start_recording()` and
-  ``AudioBufferProcessor.stop_recording()` instead.
+  `AudioBufferProcessor.stop_recording()` instead.
 
 ### Fixed
 

README.md

@@ -55,17 +55,17 @@ pip install "pipecat-ai[option,...]"
 
 ### Available services
 
-| Category            | Services                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    | Install Command Example                 |
-| ------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------- |
-| Speech-to-Text      | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper)                                                                                                                                                                                                                                                                                                                                                                                                                                                            | `pip install "pipecat-ai[deepgram]"`    |
+| Category            | Services                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          | Install Command Example                 |
+| ------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------- |
+| Speech-to-Text      | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper)                                                                                                                                                                                                                                                                                                                                                                                                                                                          | `pip install "pipecat-ai[deepgram]"`    |
 | LLMs                | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Together AI](https://docs.pipecat.ai/server/services/llm/together) | `pip install "pipecat-ai[openai]"`      |
-| Text-to-Speech      | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts)                              | `pip install "pipecat-ai[cartesia]"`    |
-| Speech-to-Speech    | [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         | `pip install "pipecat-ai[openai]"`      |
-| Transport           | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          | `pip install "pipecat-ai[daily]"`       |
-| Video               | [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  | `pip install "pipecat-ai[tavus,simli]"` |
-| Vision & Image      | [Moondream](https://docs.pipecat.ai/server/services/vision/moondream), [fal](https://docs.pipecat.ai/server/services/image-generation/fal)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  | `pip install "pipecat-ai[moondream]"`   |
-| Audio Processing    | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [Noisereduce](https://docs.pipecat.ai/server/utilities/audio/noisereduce-filter)                                                                                                                                                                                                                                                                                                                                                                                                                                                                              | `pip install "pipecat-ai[silero]"`      |
-| Analytics & Metrics | [Canonical AI](https://docs.pipecat.ai/server/services/analytics/canonical), [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             | `pip install "pipecat-ai[canonical]"`   |
+| Text-to-Speech      | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts)                                                                                                    | `pip install "pipecat-ai[cartesia]"`    |
+| Speech-to-Speech    | [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               | `pip install "pipecat-ai[openai]"`      |
+| Transport           | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                | `pip install "pipecat-ai[daily]"`       |
+| Video               | [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        | `pip install "pipecat-ai[tavus,simli]"` |
+| Vision & Image      | [Moondream](https://docs.pipecat.ai/server/services/vision/moondream), [fal](https://docs.pipecat.ai/server/services/image-generation/fal)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        | `pip install "pipecat-ai[moondream]"`   |
+| Audio Processing    | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [Noisereduce](https://docs.pipecat.ai/server/utilities/audio/noisereduce-filter)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    | `pip install "pipecat-ai[silero]"`      |
+| Analytics & Metrics | [Canonical AI](https://docs.pipecat.ai/server/services/analytics/canonical), [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   | `pip install "pipecat-ai[canonical]"`   |
 
 📚 [View full services documentation →](https://docs.pipecat.ai/server/services/supported-services)
 

dev-requirements.txt

@@ -2,10 +2,10 @@ build~=1.2.2
 grpcio-tools~=1.67.1
 pip-tools~=7.4.1
 pre-commit~=4.0.1
-pyright~=1.1.392
+pyright~=1.1.393
 pytest~=8.3.4
 pytest-asyncio~=0.25.2
-ruff~=0.9.1
+ruff~=0.9.5
 setuptools~=70.0.0
 setuptools_scm~=8.1.0
 python-dotenv~=1.0.1

examples/bot-ready-signalling/server/runner.py

@@ -6,6 +6,7 @@
 
 import argparse
 import os
+from typing import Optional
 
 import aiohttp
 
@@ -18,7 +19,7 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
 
 
 async def configure_with_args(
-    aiohttp_session: aiohttp.ClientSession, parser: argparse.ArgumentParser | None = None
+    aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
 ):
     if not parser:
         parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")

examples/canonical-metrics/bot.py

@@ -65,7 +65,6 @@ async def main():
             # English
             #
             voice_id="cgSgspJ2msm6clMCkdW9",
-            aiohttp_session=session,
             #
             # Spanish
             #

examples/chatbot-audio-recording/bot.py

@@ -82,7 +82,6 @@ async def main():
             # English
             #
             voice_id="cgSgspJ2msm6clMCkdW9",
-            aiohttp_session=session,
             #
             # Spanish
             #

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

@@ -51,7 +51,6 @@ async def main():
         )
 
         elevenlabs_tts = ElevenLabsTTSService(
-            aiohttp_session=session,
             api_key=os.getenv("ELEVENLABS_API_KEY"),
             voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
         )

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

@@ -18,7 +18,7 @@ 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.openai import OpenAILLMService, OpenAITTSService
+from pipecat.services.openai import OpenAILLMService, OpenAISTTService, OpenAITTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
 load_dotenv(override=True)
@@ -37,12 +37,22 @@ async def main():
             "Respond bot",
             DailyParams(
                 audio_out_enabled=True,
-                transcription_enabled=True,
+                audio_out_sample_rate=24000,
+                transcription_enabled=False,
                 vad_enabled=True,
                 vad_analyzer=SileroVADAnalyzer(),
+                vad_audio_passthrough=True,
             ),
         )
 
+        # You can use the OpenAI compatible API like Groq.
+        # stt = OpenAISTTService(
+        #     base_url="https://api.groq.com/openai/v1",
+        #     api_key="gsk_***",
+        #     model="whisper-large-v3",
+        # )
+        stt = OpenAISTTService(api_key=os.getenv("OPENAI_API_KEY"), model="whisper-1")
+
         tts = OpenAITTSService(api_key=os.getenv("OPENAI_API_KEY"), voice="alloy")
 
         llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
@@ -60,6 +70,7 @@ async def main():
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
+                stt,  # STT
                 context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS

examples/foundational/07s-interruptible-google-audio-in.py

@@ -216,11 +216,7 @@ async def main():
             voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
         )
 
-        llm = GoogleLLMService(
-            model="gemini-1.5-flash-latest",
-            # model="gemini-exp-1114",
-            api_key=os.getenv("GOOGLE_API_KEY"),
-        )
+        llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001")
 
         messages = [
             {

examples/foundational/08-bots-arguing.py

@@ -48,7 +48,6 @@ async def main():
             region=os.getenv("AZURE_SPEECH_REGION"),
         )
         tts2 = ElevenLabsTTSService(
-            aiohttp_session=session,
             api_key=os.getenv("ELEVENLABS_API_KEY"),
             voice_id="jBpfuIE2acCO8z3wKNLl",
         )

examples/foundational/12-describe-video.py

@@ -7,6 +7,7 @@
 import asyncio
 import os
 import sys
+from typing import Optional
 
 import aiohttp
 from dotenv import load_dotenv
@@ -32,7 +33,7 @@ logger.add(sys.stderr, level="DEBUG")
 
 
 class UserImageRequester(FrameProcessor):
-    def __init__(self, participant_id: str | None = None):
+    def __init__(self, participant_id: Optional[str] = None):
         super().__init__()
         self._participant_id = participant_id
 

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

@@ -7,6 +7,7 @@
 import asyncio
 import os
 import sys
+from typing import Optional
 
 import aiohttp
 from dotenv import load_dotenv
@@ -32,7 +33,7 @@ logger.add(sys.stderr, level="DEBUG")
 
 
 class UserImageRequester(FrameProcessor):
-    def __init__(self, participant_id: str | None = None):
+    def __init__(self, participant_id: Optional[str] = None):
         super().__init__()
         self._participant_id = participant_id
 
@@ -72,9 +73,7 @@ async def main():
 
         vision_aggregator = VisionImageFrameAggregator()
 
-        google = GoogleLLMService(
-            model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY")
-        )
+        google = GoogleLLMService(model="gemini-2.0-flash-001", api_key=os.getenv("GOOGLE_API_KEY"))
 
         tts = CartesiaTTSService(
             api_key=os.getenv("CARTESIA_API_KEY"),

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

@@ -7,6 +7,7 @@
 import asyncio
 import os
 import sys
+from typing import Optional
 
 import aiohttp
 from dotenv import load_dotenv
@@ -32,7 +33,7 @@ logger.add(sys.stderr, level="DEBUG")
 
 
 class UserImageRequester(FrameProcessor):
-    def __init__(self, participant_id: str | None = None):
+    def __init__(self, participant_id: Optional[str] = None):
         super().__init__()
         self._participant_id = participant_id
 

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

@@ -7,6 +7,7 @@
 import asyncio
 import os
 import sys
+from typing import Optional
 
 import aiohttp
 from dotenv import load_dotenv
@@ -32,7 +33,7 @@ logger.add(sys.stderr, level="DEBUG")
 
 
 class UserImageRequester(FrameProcessor):
-    def __init__(self, participant_id: str | None = None):
+    def __init__(self, participant_id: Optional[str] = None):
         super().__init__()
         self._participant_id = participant_id
 

examples/foundational/14e-function-calling-gemini.py

@@ -62,11 +62,7 @@ async def main():
             voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
         )
 
-        llm = GoogleLLMService(
-            model="gemini-1.5-flash-latest",
-            # model="gemini-exp-1114",
-            api_key=os.getenv("GOOGLE_API_KEY"),
-        )
+        llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001")
         llm.register_function("get_weather", get_weather)
         llm.register_function("get_image", get_image)
 

examples/foundational/20d-persistent-context-gemini.py

@@ -237,7 +237,7 @@ async def main():
             voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
         )
 
-        llm = GoogleLLMService(model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY"))
+        llm = GoogleLLMService(model="gemini-2.0-flash-001", api_key=os.getenv("GOOGLE_API_KEY"))
 
         # you can either register a single function for all function calls, or specific functions
         # llm.register_function(None, fetch_weather_from_api)

examples/foundational/22b-natural-conversation-proposal.py

@@ -12,6 +12,7 @@ import time
 import aiohttp
 from dotenv import load_dotenv
 from loguru import logger
+from openai.types.chat import ChatCompletionToolParam
 from runner import configure
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
@@ -19,6 +20,8 @@ from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
     Frame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
     LLMMessagesFrame,
     StartFrame,
     StartInterruptionFrame,
@@ -26,6 +29,7 @@ from pipecat.frames.frames import (
     SystemFrame,
     TextFrame,
     TranscriptionFrame,
+    TTSSpeakFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -156,6 +160,11 @@ class OutputGate(FrameProcessor):
             await self.push_frame(frame, direction)
             return
 
+        # Don't block function call frames
+        if isinstance(frame, (FunctionCallInProgressFrame, FunctionCallResultFrame)):
+            await self.push_frame(frame, direction)
+            return
+
         # Ignore frames that are not following the direction of this gate.
         if direction != FrameDirection.DOWNSTREAM:
             await self.push_frame(frame, direction)
@@ -186,6 +195,16 @@ class OutputGate(FrameProcessor):
                 break
 
 
+async def start_fetch_weather(function_name, llm, context):
+    """Push a frame to the LLM; this is handy when the LLM response might take a while."""
+    await llm.push_frame(TTSSpeakFrame("Let me check on that."))
+    logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
+
+
+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 main():
     async with aiohttp.ClientSession() as session:
         (room_url, _) = await configure(session)
@@ -216,6 +235,34 @@ async def main():
 
         # This is the regular LLM.
         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, fetch_weather_from_api, start_callback=start_fetch_weather)
+
+        tools = [
+            ChatCompletionToolParam(
+                type="function",
+                function={
+                    "name": "get_current_weather",
+                    "description": "Get the current weather",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "location": {
+                                "type": "string",
+                                "description": "The city and state, e.g. San Francisco, CA",
+                            },
+                            "format": {
+                                "type": "string",
+                                "enum": ["celsius", "fahrenheit"],
+                                "description": "The temperature unit to use. Infer this from the users location.",
+                            },
+                        },
+                        "required": ["location", "format"],
+                    },
+                },
+            )
+        ]
 
         messages = [
             {
@@ -224,7 +271,7 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
+        context = OpenAILLMContext(messages, tools)
         context_aggregator = llm.create_context_aggregator(context)
 
         # We have instructed the LLM to return 'YES' if it thinks the user
@@ -265,6 +312,8 @@ async def main():
                 or isinstance(frame, LLMMessagesFrame)
                 or isinstance(frame, StartInterruptionFrame)
                 or isinstance(frame, StopInterruptionFrame)
+                or isinstance(frame, FunctionCallInProgressFrame)
+                or isinstance(frame, FunctionCallResultFrame)
             )
 
         pipeline = Pipeline(

examples/foundational/22c-natural-conversation-mixed-llms.py

@@ -12,6 +12,7 @@ import time
 import aiohttp
 from dotenv import load_dotenv
 from loguru import logger
+from openai.types.chat import ChatCompletionToolParam
 from runner import configure
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
@@ -19,6 +20,8 @@ from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
     Frame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
     LLMMessagesFrame,
     StartFrame,
     StartInterruptionFrame,
@@ -26,6 +29,7 @@ from pipecat.frames.frames import (
     SystemFrame,
     TextFrame,
     TranscriptionFrame,
+    TTSSpeakFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -360,6 +364,11 @@ class OutputGate(FrameProcessor):
             await self.push_frame(frame, direction)
             return
 
+        # Don't block function call frames
+        if isinstance(frame, (FunctionCallInProgressFrame, FunctionCallResultFrame)):
+            await self.push_frame(frame, direction)
+            return
+
         # Ignore frames that are not following the direction of this gate.
         if direction != FrameDirection.DOWNSTREAM:
             await self.push_frame(frame, direction)
@@ -390,6 +399,16 @@ class OutputGate(FrameProcessor):
                 break
 
 
+async def start_fetch_weather(function_name, llm, context):
+    """Push a frame to the LLM; this is handy when the LLM response might take a while."""
+    await llm.push_frame(TTSSpeakFrame("Let me check on that."))
+    logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
+
+
+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 main():
     async with aiohttp.ClientSession() as session:
         (room_url, _) = await configure(session)
@@ -426,6 +445,34 @@ async def main():
             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, fetch_weather_from_api, start_callback=start_fetch_weather)
+
+        tools = [
+            ChatCompletionToolParam(
+                type="function",
+                function={
+                    "name": "get_current_weather",
+                    "description": "Get the current weather",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "location": {
+                                "type": "string",
+                                "description": "The city and state, e.g. San Francisco, CA",
+                            },
+                            "format": {
+                                "type": "string",
+                                "enum": ["celsius", "fahrenheit"],
+                                "description": "The temperature unit to use. Infer this from the users location.",
+                            },
+                        },
+                        "required": ["location", "format"],
+                    },
+                },
+            )
+        ]
 
         messages = [
             {
@@ -434,7 +481,7 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
+        context = OpenAILLMContext(messages, tools)
         context_aggregator = llm.create_context_aggregator(context)
 
         # We have instructed the LLM to return 'YES' if it thinks the user
@@ -474,6 +521,8 @@ async def main():
                 or isinstance(frame, LLMMessagesFrame)
                 or isinstance(frame, StartInterruptionFrame)
                 or isinstance(frame, StopInterruptionFrame)
+                or isinstance(frame, FunctionCallInProgressFrame)
+                or isinstance(frame, FunctionCallResultFrame)
             )
 
         pipeline = Pipeline(

examples/foundational/22d-natural-conversation-gemini-audio.py

@@ -20,6 +20,8 @@ from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
     Frame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
     InputAudioRawFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
@@ -55,13 +57,9 @@ load_dotenv(override=True)
 logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
-# TRANSCRIBER_MODEL = "gemini-1.5-flash-latest"
-# CLASSIFIER_MODEL = "gemini-1.5-flash-latest"
-# CONVERSATION_MODEL = "gemini-1.5-flash-latest"
-
-TRANSCRIBER_MODEL = "gemini-2.0-flash-exp"
-CLASSIFIER_MODEL = "gemini-2.0-flash-exp"
-CONVERSATION_MODEL = "gemini-2.0-flash-exp"
+TRANSCRIBER_MODEL = "gemini-2.0-flash-001"
+CLASSIFIER_MODEL = "gemini-2.0-flash-001"
+CONVERSATION_MODEL = "gemini-2.0-flash-001"
 
 transcriber_system_instruction = """You are an audio transcriber. You are receiving audio from a user. Your job is to
 transcribe the input audio to text exactly as it was said by the user.
@@ -579,6 +577,11 @@ class OutputGate(FrameProcessor):
             await self.push_frame(frame, direction)
             return
 
+        # Don't block function call frames
+        if isinstance(frame, (FunctionCallInProgressFrame, FunctionCallResultFrame)):
+            await self.push_frame(frame, direction)
+            return
+
         # Ignore frames that are not following the direction of this gate.
         if direction != FrameDirection.DOWNSTREAM:
             await self.push_frame(frame, direction)
@@ -676,12 +679,6 @@ async def main():
         context = OpenAILLMContext()
         context_aggregator = conversation_llm.create_context_aggregator(context)
 
-        # We have instructed the LLM to return 'True' if it thinks the user
-        # completed a sentence. So, if it's 'True' we will return true in this
-        # predicate which will wake up the notifier.
-        async def wake_check_filter(frame):
-            return frame.text == "True"
-
         # This is a notifier that we use to synchronize the two LLMs.
         notifier = EventNotifier()
 
@@ -698,14 +695,6 @@ async def main():
         async def block_user_stopped_speaking(frame):
             return not isinstance(frame, UserStoppedSpeakingFrame)
 
-        async def pass_only_llm_trigger_frames(frame):
-            return (
-                isinstance(frame, OpenAILLMContextFrame)
-                or isinstance(frame, LLMMessagesFrame)
-                or isinstance(frame, StartInterruptionFrame)
-                or isinstance(frame, StopInterruptionFrame)
-            )
-
         conversation_audio_context_assembler = ConversationAudioContextAssembler(context=context)
 
         user_aggregator_buffer = UserAggregatorBuffer()

examples/foundational/25-google-audio-in.py

@@ -292,7 +292,7 @@ async def main():
 
         conversation_llm = GoogleLLMService(
             name="Conversation",
-            model="gemini-1.5-flash-latest",
+            model="gemini-2.0-flash-001",
             # model="gemini-exp-1121",
             api_key=os.getenv("GOOGLE_API_KEY"),
             # we can give the GoogleLLMService a system instruction to use directly
@@ -303,7 +303,7 @@ async def main():
 
         input_transcription_llm = GoogleLLMService(
             name="Transcription",
-            model="gemini-1.5-flash-latest",
+            model="gemini-2.0-flash-001",
             # model="gemini-exp-1121",
             api_key=os.getenv("GOOGLE_API_KEY"),
             system_instruction=transcriber_system_message,

examples/foundational/32-gemini-grounding-metadata.py

@@ -89,6 +89,7 @@ async def main():
             api_key=os.getenv("GOOGLE_API_KEY"),
             system_instruction=system_instruction,
             tools=tools,
+            model="gemini-1.5-flash-002",
         )
 
         context = OpenAILLMContext(

examples/foundational/runner.py

@@ -6,6 +6,7 @@
 
 import argparse
 import os
+from typing import Optional
 
 import aiohttp
 
@@ -18,7 +19,7 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
 
 
 async def configure_with_args(
-    aiohttp_session: aiohttp.ClientSession, parser: argparse.ArgumentParser | None = None
+    aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
 ):
     if not parser:
         parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")

examples/news-chatbot/server/news_bot.py

@@ -23,7 +23,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.deepgram import DeepgramSTTService
-from pipecat.services.google import GoogleLLMService, LLMSearchResponseFrame
+from pipecat.services.google import GoogleLLMService, GoogleRTVIObserver, LLMSearchResponseFrame
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.utils.text.markdown_text_filter import MarkdownTextFilter
 
@@ -102,6 +102,7 @@ async def main():
 
         llm = GoogleLLMService(
             api_key=os.getenv("GOOGLE_API_KEY"),
+            model="gemini-1.5-flash-002",
             system_instruction=system_instruction,
             tools=tools,
         )
@@ -141,7 +142,7 @@ async def main():
             pipeline,
             PipelineParams(
                 allow_interruptions=True,
-                observers=[rtvi.observer()],
+                observers=[GoogleRTVIObserver(rtvi)],
             ),
         )
 

examples/news-chatbot/server/runner.py

@@ -6,6 +6,7 @@
 
 import argparse
 import os
+from typing import Optional
 
 import aiohttp
 
@@ -18,7 +19,7 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
 
 
 async def configure_with_args(
-    aiohttp_session: aiohttp.ClientSession, parser: argparse.ArgumentParser | None = None
+    aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
 ):
     if not parser:
         parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")

examples/phone-chatbot/bot_daily.py

@@ -2,13 +2,14 @@ import argparse
 import asyncio
 import os
 import sys
+from typing import Optional
 
 from dotenv import load_dotenv
 from loguru import logger
 from openai.types.chat import ChatCompletionToolParam
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import EndFrame, EndTaskFrame
+from pipecat.frames.frames import EndTaskFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -42,7 +43,7 @@ async def main(
     callId: str,
     callDomain: str,
     detect_voicemail: bool,
-    dialout_number: str | None,
+    dialout_number: Optional[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
@@ -99,14 +100,14 @@ async def main(
             - **ASSUME IT IS A VOICEMAIL. DO NOT WAIT FOR MORE CONFIRMATION.**
 
             #### **Step 2: Leave a Voicemail Message**
-            - Immediately say:  
+            - Immediately say:
             *"Hello, this is a message for Pipecat example user. This is Chatbot. Please call back on 123-456-7891. Thank you."*
             - **IMMEDIATELY AFTER LEAVING THE MESSAGE, CALL `terminate_call`.**
             - **DO NOT SPEAK AFTER CALLING `terminate_call`.**
             - **FAILURE TO CALL `terminate_call` IMMEDIATELY IS A MISTAKE.**
 
             #### **Step 3: If Speaking to a Human**
-            - If the call is answered by a human, say:  
+            - If the call is answered by a human, say:
             *"Oh, hello! I'm a friendly chatbot. Is there anything I can help you with?"*
             - Keep responses **brief and helpful**.
             - If the user no longer needs assistance, **call `terminate_call` immediately.**

examples/sentry-metrics/.gitignore

@@ -0,0 +1,161 @@
+# 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

examples/sentry-metrics/Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.10-bullseye
+
+RUN mkdir /app
+RUN mkdir /app/assets
+RUN mkdir /app/utils
+COPY *.py /app/
+COPY requirements.txt /app/
+
+
+WORKDIR /app
+RUN pip3 install -r requirements.txt
+
+EXPOSE 7860
+
+CMD ["python3", "server.py"]

examples/sentry-metrics/README.md

@@ -0,0 +1,29 @@
+# Sentry Metrics
+
+This app connects you to a chatbot powered by GPT-4. It provides TTFB (Time-To-First-Byte) and processing metrics to Sentry.
+
+## Get started
+
+```python
+python3 -m venv venv
+source venv/bin/activate
+pip install -r requirements.txt
+
+cp env.example .env # and add your credentials
+
+```
+
+## Run the server
+
+```bash
+python server.py
+```
+
+Then, visit `http://localhost:7860/` in your browser to start a chatbot session.
+
+## Build and test the Docker image
+
+```
+docker build -t chatbot .
+docker run --env-file .env -p 7860:7860 chatbot
+```

examples/sentry-metrics/bot.py

@@ -0,0 +1,112 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+import sentry_sdk
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.metrics.sentry import SentryMetrics
+from pipecat.services.elevenlabs import ElevenLabsTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+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,
+            "Chatbot",
+            DailyParams(
+                audio_out_enabled=True,
+                audio_in_enabled=True,
+                camera_out_enabled=False,
+                vad_enabled=True,
+                vad_audio_passthrough=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                transcription_enabled=True,
+            ),
+        )
+
+        # Initialize Sentry
+        sentry_sdk.init(
+            dsn="your-project-dsn",
+            traces_sample_rate=1.0,
+        )
+
+        tts = ElevenLabsTTSService(
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id="cgSgspJ2msm6clMCkdW9",
+            metrics=SentryMetrics(),
+        )
+
+        llm = OpenAILLMService(
+            api_key=os.getenv("OPENAI_API_KEY"),
+            model="gpt-4o",
+            metrics=SentryMetrics(),
+        )
+
+        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 introducing yourself.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # microphone
+                context_aggregator.user(),
+                llm,
+                tts,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        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):
+            await transport.capture_participant_transcription(participant["id"])
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        @transport.event_handler("on_participant_left")
+        async def on_participant_left(transport, participant, reason):
+            print(f"Participant left: {participant}")
+            await task.cancel()
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/sentry-metrics/env.example

@@ -0,0 +1,4 @@
+DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev)
+DAILY_API_KEY=7df...
+OPENAI_API_KEY=sk-PL...
+ELEVENLABS_API_KEY=aeb...

examples/sentry-metrics/requirements.txt

@@ -0,0 +1,4 @@
+python-dotenv
+fastapi[all]
+uvicorn
+pipecat-ai[daily,openai,sentry,silero,elevenlabs]

examples/sentry-metrics/runner.py

@@ -0,0 +1,56 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+
+import aiohttp
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+
+
+async def configure(aiohttp_session: aiohttp.ClientSession):
+    parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
+    parser.add_argument(
+        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
+    )
+    parser.add_argument(
+        "-k",
+        "--apikey",
+        type=str,
+        required=False,
+        help="Daily API Key (needed to create an owner token for the room)",
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
+    key = args.apikey or os.getenv("DAILY_API_KEY")
+
+    if not url:
+        raise Exception(
+            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
+        )
+
+    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
+
+    token = await daily_rest_helper.get_token(url, expiry_time)
+
+    return (url, token)
+    return (url, token)

examples/sentry-metrics/server.py

@@ -0,0 +1,139 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+import subprocess
+from contextlib import asynccontextmanager
+
+import aiohttp
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, RedirectResponse
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
+
+MAX_BOTS_PER_ROOM = 1
+
+# Bot sub-process dict for status reporting and concurrency control
+bot_procs = {}
+
+daily_helpers = {}
+
+load_dotenv(override=True)
+
+
+def cleanup():
+    # Clean up function, just to be extra safe
+    for entry in bot_procs.values():
+        proc = entry[0]
+        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()
+
+
+app = FastAPI(lifespan=lifespan)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.get("/")
+async def start_agent(request: Request):
+    print(f"!!! Creating room")
+    room = await daily_helpers["rest"].create_room(DailyRoomParams())
+    print(f"!!! Room URL: {room.url}")
+    # Ensure the room property is present
+    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 = await daily_helpers["rest"].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 RedirectResponse(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})
+
+
+if __name__ == "__main__":
+    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/simple-chatbot/server/bot-gemini.py

@@ -40,7 +40,7 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor
+from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
 from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
@@ -176,7 +176,7 @@ async def main():
                 allow_interruptions=True,
                 enable_metrics=True,
                 enable_usage_metrics=True,
-                observers=[rtvi.observer()],
+                observers=[RTVIObserver(rtvi)],
             ),
         )
         await task.queue_frame(quiet_frame)

examples/simple-chatbot/server/bot-openai.py

@@ -40,7 +40,7 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor
+from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -202,7 +202,7 @@ async def main():
                 allow_interruptions=True,
                 enable_metrics=True,
                 enable_usage_metrics=True,
-                observers=[rtvi.observer()],
+                observers=[RTVIObserver(rtvi)],
             ),
         )
         await task.queue_frame(quiet_frame)

examples/storytelling-chatbot/README.md

@@ -74,6 +74,8 @@ If you'd like to run a custom domain or port:
 
 ➡️ Open the host URL in your browser `http://localhost:7860`
 
+If you've run previous versions of the demo, make sure to set `ENV=dev`, and remove the `RUN_AS_VM` line from the .env file.
+
 ---
 
 ## Improvements to make

examples/storytelling-chatbot/env.example

@@ -3,6 +3,4 @@ DAILY_SAMPLE_ROOM_URL=
 ELEVENLABS_API_KEY=
 ELEVENLABS_VOICE_ID=
 GOOGLE_API_KEY=
-
-ENV= # dev | production
-RUN_AS_VM= # Set this if you want to run bots on process (not launch a new VM)
+ENV=dev

examples/storytelling-chatbot/requirements.txt

@@ -2,5 +2,4 @@ async_timeout
 fastapi
 uvicorn
 python-dotenv
--e "../..[daily,silero,openai,fal,cartesia,google]"
--e "../../../python-genai"
+pipecat-ai[daily,silero,openai,cartesia,google]

examples/storytelling-chatbot/src/bot.py

@@ -23,8 +23,7 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.elevenlabs import ElevenLabsTTSService
-from pipecat.services.fal import FalImageGenService
-from pipecat.services.google import GoogleLLMService
+from pipecat.services.google import GoogleImageGenService, GoogleLLMService
 from pipecat.transports.services.daily import (
     DailyParams,
     DailyTransport,

examples/studypal/runner.py

@@ -6,6 +6,7 @@
 
 import argparse
 import os
+from typing import Optional
 
 import aiohttp
 
@@ -18,7 +19,7 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
 
 
 async def configure_with_args(
-    aiohttp_session: aiohttp.ClientSession, parser: argparse.ArgumentParser | None = None
+    aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
 ):
     if not parser:
         parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")

pyproject.toml

@@ -55,7 +55,7 @@ elevenlabs = [ "websockets~=13.1" ]
 fal = [ "fal-client~=0.5.6" ]
 fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ]
 gladia = [ "websockets~=13.1" ]
-google = [ "google-generativeai~=0.8.3", "google-cloud-texttospeech~=2.24.0", "google-genai~=0.7.0" ]
+google = [ "google-generativeai~=0.8.3", "google-cloud-texttospeech~=2.24.0", "google-genai~=1.0.0" ]
 grok = [ "openai~=1.59.6" ]
 groq = [ "openai~=1.59.6" ]
 gstreamer = [ "pygobject~=3.50.0" ]
@@ -73,6 +73,7 @@ openai = [ "openai~=1.59.6", "websockets~=13.1", "python-deepcompare~=2.1.0" ]
 openpipe = [ "openpipe~=4.45.0" ]
 playht = [ "pyht~=0.1.6", "websockets~=13.1" ]
 riva = [ "nvidia-riva-client~=2.18.0" ]
+sentry = [ "sentry-sdk~=2.20.0" ]
 silero = [ "onnxruntime~=1.20.1" ]
 simli = [ "simli-ai~=0.1.10"]
 soundfile = [ "soundfile~=0.13.0" ]

src/pipecat/frames/frames.py

@@ -48,7 +48,7 @@ class KeypadEntry(str, Enum):
     STAR = "*"
 
 
-def format_pts(pts: int | None):
+def format_pts(pts: Optional[int]):
     return nanoseconds_to_str(pts) if pts else None
 
 
@@ -126,7 +126,7 @@ class ImageRawFrame:
 
     image: bytes
     size: Tuple[int, int]
-    format: str | None
+    format: Optional[str]
 
 
 #
@@ -176,7 +176,7 @@ class URLImageRawFrame(OutputImageRawFrame):
 
     """
 
-    url: str | None
+    url: Optional[str]
 
     def __str__(self):
         pts = format_pts(self.pts)
@@ -235,7 +235,7 @@ class TranscriptionFrame(TextFrame):
 
     user_id: str
     timestamp: str
-    language: Language | None = None
+    language: Optional[Language] = None
 
     def __str__(self):
         return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})"
@@ -250,7 +250,7 @@ class InterimTranscriptionFrame(TextFrame):
     text: str
     user_id: str
     timestamp: str
-    language: Language | None = None
+    language: Optional[Language] = None
 
     def __str__(self):
         return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})"
@@ -272,7 +272,7 @@ class TranscriptionMessage:
 
     role: Literal["user", "assistant"]
     content: str
-    timestamp: str | None = None
+    timestamp: Optional[str] = None
 
 
 @dataclass
@@ -674,7 +674,7 @@ class UserImageRawFrame(InputImageRawFrame):
 class VisionImageRawFrame(InputImageRawFrame):
     """An image with an associated text to ask for a description of it."""
 
-    text: str | None
+    text: Optional[str]
 
     def __str__(self):
         pts = format_pts(self.pts)

src/pipecat/pipeline/runner.py

@@ -19,7 +19,7 @@ class PipelineRunner:
     def __init__(
         self,
         *,
-        name: str | None = None,
+        name: Optional[str] = None,
         handle_sigint: bool = True,
         force_gc: bool = False,
         loop: Optional[asyncio.AbstractEventLoop] = None,

src/pipecat/pipeline/task.py

@@ -38,6 +38,22 @@ HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5
 
 
 class PipelineParams(BaseModel):
+    """Configuration parameters for pipeline execution.
+
+    Attributes:
+        allow_interruptions: Whether to allow pipeline interruptions.
+        audio_in_sample_rate: Input audio sample rate in Hz.
+        audio_out_sample_rate: Output audio sample rate in Hz.
+        enable_heartbeats: Whether to enable heartbeat monitoring.
+        enable_metrics: Whether to enable metrics collection.
+        enable_usage_metrics: Whether to enable usage metrics.
+        heartbeats_period_secs: Period between heartbeats in seconds.
+        observers: List of observers for monitoring pipeline execution.
+        report_only_initial_ttfb: Whether to report only initial time to first byte.
+        send_initial_empty_metrics: Whether to send initial empty metrics.
+        start_metadata: Additional metadata for pipeline start.
+    """
+
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     allow_interruptions: bool = False
@@ -54,11 +70,16 @@ class PipelineParams(BaseModel):
 
 
 class PipelineTaskSource(FrameProcessor):
-    """This is the source processor that is linked at the beginning of the
+    """Source processor for pipeline tasks that handles frame routing.
+
+    This is the source processor that is linked at the beginning of the
     pipeline given to the pipeline task. It allows us to easily push frames
     downstream to the pipeline and also receive upstream frames coming from the
     pipeline.
 
+    Args:
+        up_queue: Queue for upstream frame processing.
+
     """
 
     def __init__(self, up_queue: asyncio.Queue, **kwargs):
@@ -76,10 +97,14 @@ class PipelineTaskSource(FrameProcessor):
 
 
 class PipelineTaskSink(FrameProcessor):
-    """This is the sink processor that is linked at the end of the pipeline
+    """Sink processor for pipeline tasks that handles final frame processing.
+
+    This is the sink processor that is linked at the end of the pipeline
     given to the pipeline task. It allows us to receive downstream frames and
     act on them, for example, waiting to receive an EndFrame.
 
+    Args:
+        down_queue: Queue for downstream frame processing.
     """
 
     def __init__(self, down_queue: asyncio.Queue, **kwargs):
@@ -92,6 +117,14 @@ class PipelineTaskSink(FrameProcessor):
 
 
 class PipelineTask(BaseTask):
+    """Manages the execution of a pipeline, handling frame processing and task lifecycle.
+
+    Args:
+        pipeline: The pipeline to execute.
+        params: Configuration parameters for the pipeline.
+        clock: Clock implementation for timing operations.
+    """
+
     def __init__(
         self,
         pipeline: BasePipeline,
@@ -163,9 +196,7 @@ class PipelineTask(BaseTask):
         await self.queue_frame(EndFrame())
 
     async def cancel(self):
-        """
-        Stops the running pipeline immediately.
-        """
+        """Stops the running pipeline immediately."""
         logger.debug(f"Canceling pipeline task {self}")
         # Make sure everything is cleaned up downstream. This is sent
         # out-of-band from the main streaming task which is what we want since
@@ -175,9 +206,7 @@ class PipelineTask(BaseTask):
         await self._task_manager.cancel_task(self._process_push_task)
 
     async def run(self):
-        """
-        Starts running the given pipeline.
-        """
+        """Starts and manages the pipeline execution until completion or cancellation."""
         if self.has_finished():
             return
         try:
@@ -195,14 +224,18 @@ class PipelineTask(BaseTask):
         self._finished = True
 
     async def queue_frame(self, frame: Frame):
-        """
-        Queue a frame to be pushed down the pipeline.
+        """Queue a single frame to be pushed down the pipeline.
+
+        Args:
+            frame: The frame to be processed.
         """
         await self._push_queue.put(frame)
 
     async def queue_frames(self, frames: Iterable[Frame] | AsyncIterable[Frame]):
-        """
-        Queues multiple frames to be pushed down the pipeline.
+        """Queues multiple frames to be pushed down the pipeline.
+
+        Args:
+            frames: An iterable or async iterable of frames to be processed.
         """
         if isinstance(frames, AsyncIterable):
             async for frame in frames:
@@ -348,9 +381,7 @@ class PipelineTask(BaseTask):
             self._down_queue.task_done()
 
     async def _heartbeat_push_handler(self):
-        """
-        This tasks pushes a heartbeat frame every heartbeat period.
-        """
+        """This tasks pushes a heartbeat frame every heartbeat period."""
         while True:
             # Don't use `queue_frame()` because if an EndFrame is queued the
             # task will just stop waiting for the pipeline to finish not

src/pipecat/processors/aggregators/llm_response.py

@@ -4,7 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from typing import List, Type
+from typing import List, Optional, Type
 
 from pipecat.frames.frames import (
     Frame,
@@ -37,7 +37,7 @@ class LLMResponseAggregator(FrameProcessor):
         start_frame,
         end_frame,
         accumulator_frame: Type[TextFrame],
-        interim_accumulator_frame: Type[TextFrame] | None = None,
+        interim_accumulator_frame: Optional[Type[TextFrame]] = None,
         handle_interruptions: bool = False,
         expect_stripped_words: bool = True,  # if True, need to add spaces between words
     ):

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -51,7 +51,7 @@ class CustomEncoder(json.JSONEncoder):
 class OpenAILLMContext:
     def __init__(
         self,
-        messages: List[ChatCompletionMessageParam] | None = None,
+        messages: Optional[List[ChatCompletionMessageParam]] = None,
         tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN,
         tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = NOT_GIVEN,
     ):

src/pipecat/processors/aggregators/user_response.py

@@ -4,6 +4,8 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+from typing import Optional
+
 from pipecat.frames.frames import (
     Frame,
     InterimTranscriptionFrame,
@@ -50,7 +52,7 @@ class ResponseAggregator(FrameProcessor):
         start_frame,
         end_frame,
         accumulator_frame: TextFrame,
-        interim_accumulator_frame: TextFrame | None = None,
+        interim_accumulator_frame: Optional[TextFrame] = None,
     ):
         super().__init__()
 

src/pipecat/processors/frame_processor.py

@@ -245,6 +245,9 @@ class FrameProcessor:
         await self.push_frame(error, FrameDirection.UPSTREAM)
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
+        if not self._check_ready(frame):
+            return
+
         if isinstance(frame, SystemFrame):
             await self.__internal_push_frame(frame, direction)
         else:
@@ -319,6 +322,16 @@ class FrameProcessor:
             await self.push_error(ErrorFrame(str(e)))
             raise
 
+    def _check_ready(self, frame: Frame):
+        # If we are trying to push a frame but we still have no clock, it means
+        # we didn't process a StartFrame.
+        if not self._clock:
+            logger.error(
+                f"{self} not properly initialized, missing super().process_frame(frame, direction)?"
+            )
+            return False
+        return True
+
     def __create_input_task(self):
         if not self.__input_frame_task:
             self.__should_block_frames = False

src/pipecat/processors/frameworks/langchain.py

@@ -4,7 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from typing import Union
+from typing import Optional, Union
 
 from loguru import logger
 
@@ -30,7 +30,7 @@ class LangchainProcessor(FrameProcessor):
         super().__init__()
         self._chain = chain
         self._transcript_key = transcript_key
-        self._participant_id: str | None = None
+        self._participant_id: Optional[str] = None
 
     def set_participant_id(self, participant_id: str):
         self._participant_id = participant_id

src/pipecat/processors/frameworks/rtvi.py

@@ -58,7 +58,6 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame
 from pipecat.utils.string import match_endofsentence
 
 RTVI_PROTOCOL_VERSION = "0.3.0"
@@ -296,12 +295,6 @@ class RTVITextMessageData(BaseModel):
     text: str
 
 
-class RTVISearchResponseMessageData(BaseModel):
-    search_result: Optional[str]
-    rendered_content: Optional[str]
-    origins: List[LLMSearchOrigin]
-
-
 class RTVIBotTranscriptionMessage(BaseModel):
     label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
     type: Literal["bot-transcription"] = "bot-transcription"
@@ -314,12 +307,6 @@ class RTVIBotLLMTextMessage(BaseModel):
     data: RTVITextMessageData
 
 
-class RTVIBotLLMSearchResponseMessage(BaseModel):
-    label: Literal["rtvi-ai"] = "rtvi-ai"
-    type: Literal["bot-llm-search-response"] = "bot-llm-search-response"
-    data: RTVISearchResponseMessageData
-
-
 class RTVIBotTTSTextMessage(BaseModel):
     label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
     type: Literal["bot-tts-text"] = "bot-tts-text"
@@ -397,6 +384,15 @@ class RTVISpeakingProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVISpeakingProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -432,6 +428,15 @@ class RTVIUserTranscriptionProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIUserTranscriptionProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -463,6 +468,15 @@ class RTVIUserLLMTextProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIUserLLMTextProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -490,6 +504,15 @@ class RTVIBotTranscriptionProcessor(RTVIFrameProcessor):
         super().__init__()
         self._aggregation = ""
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIBotTranscriptionProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -513,6 +536,15 @@ class RTVIBotLLMProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIBotLLMProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -531,6 +563,15 @@ class RTVIBotTTSProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIBotTTSProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -549,6 +590,15 @@ class RTVIMetricsProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIMetricsProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -618,24 +668,22 @@ class RTVIObserver(BaseObserver):
         elif isinstance(frame, UserStartedSpeakingFrame):
             await self._push_bot_transcription()
         elif isinstance(frame, LLMFullResponseStartFrame):
-            await self._push_transport_message_urgent(RTVIBotLLMStartedMessage())
+            await self.push_transport_message_urgent(RTVIBotLLMStartedMessage())
         elif isinstance(frame, LLMFullResponseEndFrame):
-            await self._push_transport_message_urgent(RTVIBotLLMStoppedMessage())
+            await self.push_transport_message_urgent(RTVIBotLLMStoppedMessage())
         elif isinstance(frame, LLMTextFrame):
             await self._handle_llm_text_frame(frame)
-        elif isinstance(frame, LLMSearchResponseFrame):
-            await self._handle_llm_search_response_frame(frame)
         elif isinstance(frame, TTSStartedFrame):
-            await self._push_transport_message_urgent(RTVIBotTTSStartedMessage())
+            await self.push_transport_message_urgent(RTVIBotTTSStartedMessage())
         elif isinstance(frame, TTSStoppedFrame):
-            await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage())
+            await self.push_transport_message_urgent(RTVIBotTTSStoppedMessage())
         elif isinstance(frame, TTSTextFrame):
             message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text))
-            await self._push_transport_message_urgent(message)
+            await self.push_transport_message_urgent(message)
         elif isinstance(frame, MetricsFrame):
             await self._handle_metrics(frame)
 
-    async def _push_transport_message_urgent(self, model: BaseModel, exclude_none: bool = True):
+    async def push_transport_message_urgent(self, model: BaseModel, exclude_none: bool = True):
         frame = TransportMessageUrgentFrame(message=model.model_dump(exclude_none=exclude_none))
         await self._rtvi.push_frame(frame)
 
@@ -644,7 +692,7 @@ class RTVIObserver(BaseObserver):
             message = RTVIBotTranscriptionMessage(
                 data=RTVITextMessageData(text=self._bot_transcription)
             )
-            await self._push_transport_message_urgent(message)
+            await self.push_transport_message_urgent(message)
             self._bot_transcription = ""
 
     async def _handle_interruptions(self, frame: Frame):
@@ -655,7 +703,7 @@ class RTVIObserver(BaseObserver):
             message = RTVIUserStoppedSpeakingMessage()
 
         if message:
-            await self._push_transport_message_urgent(message)
+            await self.push_transport_message_urgent(message)
 
     async def _handle_bot_speaking(self, frame: Frame):
         message = None
@@ -665,26 +713,16 @@ class RTVIObserver(BaseObserver):
             message = RTVIBotStoppedSpeakingMessage()
 
         if message:
-            await self._push_transport_message_urgent(message)
+            await self.push_transport_message_urgent(message)
 
     async def _handle_llm_text_frame(self, frame: LLMTextFrame):
         message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text))
-        await self._push_transport_message_urgent(message)
+        await self.push_transport_message_urgent(message)
 
         self._bot_transcription += frame.text
         if match_endofsentence(self._bot_transcription):
             await self._push_bot_transcription()
 
-    async def _handle_llm_search_response_frame(self, frame: LLMSearchResponseFrame):
-        message = RTVIBotLLMSearchResponseMessage(
-            data=RTVISearchResponseMessageData(
-                search_result=frame.search_result,
-                origins=frame.origins,
-                rendered_content=frame.rendered_content,
-            )
-        )
-        await self._push_transport_message_urgent(message)
-
     async def _handle_user_transcriptions(self, frame: Frame):
         message = None
         if isinstance(frame, TranscriptionFrame):
@@ -701,7 +739,7 @@ class RTVIObserver(BaseObserver):
             )
 
         if message:
-            await self._push_transport_message_urgent(message)
+            await self.push_transport_message_urgent(message)
 
     async def _handle_context(self, frame: OpenAILLMContextFrame):
         try:
@@ -715,7 +753,7 @@ class RTVIObserver(BaseObserver):
                     else:
                         text = content
                     rtvi_message = RTVIUserLLMTextMessage(data=RTVITextMessageData(text=text))
-                    await self._push_transport_message_urgent(rtvi_message)
+                    await self.push_transport_message_urgent(rtvi_message)
         except TypeError as e:
             logger.warning(f"Caught an error while trying to handle context: {e}")
 
@@ -740,7 +778,7 @@ class RTVIObserver(BaseObserver):
                 metrics["characters"].append(d.model_dump(exclude_none=True))
 
         message = RTVIMetricsMessage(data=metrics)
-        await self._push_transport_message_urgent(message)
+        await self.push_transport_message_urgent(message)
 
 
 class RTVIProcessor(FrameProcessor):
@@ -753,7 +791,7 @@ class RTVIProcessor(FrameProcessor):
         super().__init__(**kwargs)
         self._config = config
 
-        self._pipeline: FrameProcessor | None = None
+        self._pipeline: Optional[FrameProcessor] = None
 
         self._bot_ready = False
         self._client_ready = False
@@ -774,6 +812,15 @@ class RTVIProcessor(FrameProcessor):
         self._register_event_handler("on_client_ready")
 
     def observer(self) -> RTVIObserver:
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVI.observer()' is deprecated, instantiate an 'RTVIObserver' directly instead.",
+                DeprecationWarning,
+            )
+
         return RTVIObserver(self)
 
     def register_action(self, action: RTVIAction):
@@ -999,7 +1046,7 @@ class RTVIProcessor(FrameProcessor):
         )
         await self.push_frame(frame)
 
-    async def _handle_action(self, request_id: str | None, data: RTVIActionRun):
+    async def _handle_action(self, request_id: Optional[str], data: RTVIActionRun):
         action_id = self._action_id(data.service, data.action)
         if action_id not in self._registered_actions:
             await self._send_error_response(request_id, f"Action {action_id} not registered")

src/pipecat/processors/metrics/sentry.py

@@ -4,19 +4,14 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import time
-
 from loguru import logger
 
 try:
     import sentry_sdk
-
-    sentry_available = sentry_sdk.is_initialized()
-    if not sentry_available:
-        logger.warning("Sentry SDK not initialized. Sentry features will be disabled.")
-except ImportError:
-    sentry_available = False
-    logger.warning("Sentry SDK not installed. Sentry features will be disabled.")
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Sentry, you need to `pip install pipecat-ai[sentry]`.")
+    raise Exception(f"Missing module: {e}")
 
 from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
 
@@ -24,41 +19,44 @@ from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMet
 class SentryMetrics(FrameProcessorMetrics):
     def __init__(self):
         super().__init__()
-        self._ttfb_metrics_span = None
-        self._processing_metrics_span = None
+        self._ttfb_metrics_tx = None
+        self._processing_metrics_tx = None
+        self._sentry_available = sentry_sdk.is_initialized()
+        if not self._sentry_available:
+            logger.warning("Sentry SDK not initialized. Sentry features will be disabled.")
 
     async def start_ttfb_metrics(self, report_only_initial_ttfb):
-        if self._should_report_ttfb:
-            self._start_ttfb_time = time.time()
-            if sentry_available:
-                self._ttfb_metrics_span = sentry_sdk.start_span(
-                    op="ttfb",
-                    description=f"TTFB for {self._processor_name()}",
-                    start_timestamp=self._start_ttfb_time,
-                )
-                logger.debug(
-                    f"Sentry Span ID: {self._ttfb_metrics_span.span_id} Description: {self._ttfb_metrics_span.description} started."
-                )
-            self._should_report_ttfb = not report_only_initial_ttfb
+        await super().start_ttfb_metrics(report_only_initial_ttfb)
 
-    async def stop_ttfb_metrics(self):
-        stop_time = time.time()
-        if sentry_available:
-            self._ttfb_metrics_span.finish(end_timestamp=stop_time)
-
-    async def start_processing_metrics(self):
-        self._start_processing_time = time.time()
-        if sentry_available:
-            self._processing_metrics_span = sentry_sdk.start_span(
-                op="processing",
-                description=f"Processing for {self._processor_name()}",
-                start_timestamp=self._start_processing_time,
+        if self._should_report_ttfb and self._sentry_available:
+            self._ttfb_metrics_tx = sentry_sdk.start_transaction(
+                op="ttfb",
+                name=f"TTFB for {self._processor_name()}",
             )
             logger.debug(
-                f"Sentry Span ID: {self._processing_metrics_span.span_id} Description: {self._processing_metrics_span.description} started."
+                f"Sentry transaction started (ID: {self._ttfb_metrics_tx.span_id} Name: {self._ttfb_metrics_tx.name})"
+            )
+
+    async def stop_ttfb_metrics(self):
+        await super().stop_ttfb_metrics()
+
+        if self._sentry_available and self._ttfb_metrics_tx:
+            self._ttfb_metrics_tx.finish()
+
+    async def start_processing_metrics(self):
+        await super().start_processing_metrics()
+
+        if self._sentry_available:
+            self._processing_metrics_tx = sentry_sdk.start_transaction(
+                op="processing",
+                name=f"Processing for {self._processor_name()}",
+            )
+            logger.debug(
+                f"Sentry transaction started (ID: {self._processing_metrics_tx.span_id} Name: {self._processing_metrics_tx.name})"
             )
 
     async def stop_processing_metrics(self):
-        stop_time = time.time()
-        if sentry_available:
-            self._processing_metrics_span.finish(end_timestamp=stop_time)
+        await super().stop_processing_metrics()
+
+        if self._sentry_available and self._processing_metrics_tx:
+            self._processing_metrics_tx.finish()

src/pipecat/processors/transcript_processor.py

@@ -87,7 +87,7 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
         """Initialize processor with aggregation state."""
         super().__init__(**kwargs)
         self._current_text_parts: List[str] = []
-        self._aggregation_start_time: Optional[str] | None = None
+        self._aggregation_start_time: Optional[str] = None
 
     async def _emit_aggregated_text(self):
         """Emit aggregated text as a transcript message."""

src/pipecat/serializers/telnyx.py

@@ -31,8 +31,8 @@ from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializer
 
 class TelnyxFrameSerializer(FrameSerializer):
     class InputParams(BaseModel):
-        telnyx_sample_rate: Optional[int] = None
-        sample_rate: Optional[int] = None
+        telnyx_sample_rate: int = 8000  # Default Telnyx rate (8kHz)
+        sample_rate: Optional[int] = None  # Pipeline input rate
         inbound_encoding: str = "PCMU"
         outbound_encoding: str = "PCMU"
 
@@ -48,6 +48,9 @@ class TelnyxFrameSerializer(FrameSerializer):
         params.inbound_encoding = inbound_encoding
         self._params = params
 
+        self._telnyx_sample_rate = self._params.telnyx_sample_rate
+        self._sample_rate = 0  # Pipeline input rate
+
         self._resampler = create_default_resampler()
 
     @property
@@ -55,13 +58,13 @@ class TelnyxFrameSerializer(FrameSerializer):
         return FrameSerializerType.TEXT
 
     async def setup(self, frame: StartFrame):
-        self._telnyx_sample_rate = self._params.telnyx_sample_rate or frame.audio_in_sample_rate
-        self._sample_rate = self._params.sample_rate or frame.audio_out_sample_rate
+        self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate
 
     async def serialize(self, frame: Frame) -> str | bytes | None:
         if isinstance(frame, AudioRawFrame):
             data = frame.audio
 
+            # Output: Convert PCM at frame's rate to 8kHz encoded for Telnyx
             if self._params.inbound_encoding == "PCMU":
                 serialized_data = await pcm_to_ulaw(
                     data, frame.sample_rate, self._telnyx_sample_rate, self._resampler
@@ -92,6 +95,7 @@ class TelnyxFrameSerializer(FrameSerializer):
             payload_base64 = message["media"]["payload"]
             payload = base64.b64decode(payload_base64)
 
+            # Input: Convert Telnyx's 8kHz encoded audio to PCM at pipeline input rate
             if self._params.outbound_encoding == "PCMU":
                 deserialized_data = await ulaw_to_pcm(
                     payload,

src/pipecat/serializers/twilio.py

@@ -27,15 +27,15 @@ from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializer
 
 class TwilioFrameSerializer(FrameSerializer):
     class InputParams(BaseModel):
-        twilio_sample_rate: Optional[int] = None
-        sample_rate: Optional[int] = None
+        twilio_sample_rate: int = 8000  # Default Twilio rate (8kHz)
+        sample_rate: Optional[int] = None  # Pipeline input rate
 
     def __init__(self, stream_sid: str, params: InputParams = InputParams()):
         self._stream_sid = stream_sid
         self._params = params
 
-        self._twilio_sample_rate = 0
-        self._sample_rate = 0
+        self._twilio_sample_rate = self._params.twilio_sample_rate
+        self._sample_rate = 0  # Pipeline input rate
 
         self._resampler = create_default_resampler()
 
@@ -44,8 +44,7 @@ class TwilioFrameSerializer(FrameSerializer):
         return FrameSerializerType.TEXT
 
     async def setup(self, frame: StartFrame):
-        self._twilio_sample_rate = self._params.twilio_sample_rate or frame.audio_in_sample_rate
-        self._sample_rate = self._params.sample_rate or frame.audio_out_sample_rate
+        self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate
 
     async def serialize(self, frame: Frame) -> str | bytes | None:
         if isinstance(frame, StartInterruptionFrame):
@@ -54,6 +53,7 @@ class TwilioFrameSerializer(FrameSerializer):
         elif isinstance(frame, AudioRawFrame):
             data = frame.audio
 
+            # Output: Convert PCM at frame's rate to 8kHz μ-law for Twilio
             serialized_data = await pcm_to_ulaw(
                 data, frame.sample_rate, self._twilio_sample_rate, self._resampler
             )
@@ -75,6 +75,7 @@ class TwilioFrameSerializer(FrameSerializer):
             payload_base64 = message["media"]["payload"]
             payload = base64.b64decode(payload_base64)
 
+            # Input: Convert Twilio's 8kHz μ-law to PCM at pipeline input rate
             deserialized_data = await ulaw_to_pcm(
                 payload, self._twilio_sample_rate, self._sample_rate, self._resampler
             )

src/pipecat/services/ai_services.py

@@ -140,7 +140,7 @@ class LLMService(AIService):
         self._start_callbacks = {}
 
     # TODO-CB: callback function type
-    def register_function(self, function_name: str | None, callback, start_callback=None):
+    def register_function(self, function_name: Optional[str], callback, start_callback=None):
         # Registering a function with the function_name set to None will run that callback
         # for all functions
         self._callbacks[function_name] = callback
@@ -148,7 +148,7 @@ class LLMService(AIService):
         if start_callback:
             self._start_callbacks[function_name] = start_callback
 
-    def unregister_function(self, function_name: str | None):
+    def unregister_function(self, function_name: Optional[str]):
         del self._callbacks[function_name]
         if self._start_callbacks[function_name]:
             del self._start_callbacks[function_name]
@@ -190,7 +190,7 @@ class LLMService(AIService):
         elif None in self._start_callbacks.keys():
             return await self._start_callbacks[None](function_name, self, context)
 
-    async def request_image_frame(self, user_id: str, *, text_content: str | None = None):
+    async def request_image_frame(self, user_id: str, *, text_content: Optional[str] = None):
         await self.push_frame(
             UserImageRequestFrame(user_id=user_id, context=text_content), FrameDirection.UPSTREAM
         )
@@ -254,7 +254,7 @@ class TTSService(AIService):
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         pass
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return Language(language)
 
     async def update_setting(self, key: str, value: Any):
@@ -352,7 +352,7 @@ class TTSService(AIService):
         await self.push_frame(frame, direction)
 
     async def _process_text_frame(self, frame: TextFrame):
-        text: str | None = None
+        text: Optional[str] = None
         if not self._aggregate_sentences:
             text = frame.text
         else:

src/pipecat/services/anthropic.py

@@ -326,9 +326,9 @@ class AnthropicLLMService(LLMService):
 class AnthropicLLMContext(OpenAILLMContext):
     def __init__(
         self,
-        messages: list[dict] | None = None,
-        tools: list[dict] | None = None,
-        tool_choice: dict | None = None,
+        messages: Optional[List[dict]] = None,
+        tools: Optional[List[dict]] = None,
+        tool_choice: Optional[dict] = None,
         *,
         system: Union[str, NotGiven] = NOT_GIVEN,
     ):

src/pipecat/services/assemblyai.py

@@ -46,7 +46,7 @@ class AssemblyAISTTService(STTService):
         super().__init__(sample_rate=sample_rate, **kwargs)
 
         aai.settings.api_key = api_key
-        self._transcriber: aai.RealtimeTranscriber | None = None
+        self._transcriber: Optional[aai.RealtimeTranscriber] = None
 
         self._settings = {
             "encoding": encoding,

src/pipecat/services/aws.py

@@ -32,7 +32,7 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
-def language_to_aws_language(language: Language) -> str | None:
+def language_to_aws_language(language: Language) -> Optional[str]:
     language_map = {
         # Arabic
         Language.AR: "arb",
@@ -154,7 +154,7 @@ class PollyTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_aws_language(language)
 
     def _construct_ssml(self, text: str) -> str:

src/pipecat/services/azure.py

@@ -57,7 +57,7 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
-def language_to_azure_language(language: Language) -> str | None:
+def language_to_azure_language(language: Language) -> Optional[str]:
     language_map = {
         # Afrikaans
         Language.AF: "af-ZA",
@@ -477,7 +477,7 @@ class AzureBaseTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_azure_language(language)
 
     def _construct_ssml(self, text: str) -> str:

src/pipecat/services/canonical.py

@@ -9,7 +9,7 @@ import os
 import uuid
 import wave
 from datetime import datetime
-from typing import Dict, List, Tuple
+from typing import Dict, List, Optional, Tuple
 
 import aiohttp
 from loguru import logger
@@ -69,7 +69,7 @@ class CanonicalMetricsService(AIService):
         api_url: str = "https://voiceapp.canonical.chat/api/v1",
         assistant_speaks_first: bool = True,
         output_dir: str = "recordings",
-        context: OpenAILLMContext | None = None,
+        context: Optional[OpenAILLMContext] = None,
         **kwargs,
     ):
         super().__init__(**kwargs)

src/pipecat/services/cartesia.py

@@ -43,7 +43,7 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
-def language_to_cartesia_language(language: Language) -> str | None:
+def language_to_cartesia_language(language: Language) -> Optional[str]:
     BASE_LANGUAGES = {
         Language.DE: "de",
         Language.EN: "en",
@@ -143,7 +143,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService):
         await super().set_model(model)
         logger.info(f"Switching TTS model to: [{model}]")
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_cartesia_language(language)
 
     def _build_msg(
@@ -358,7 +358,7 @@ class CartesiaHttpTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_cartesia_language(language)
 
     async def start(self, frame: StartFrame):

src/pipecat/services/elevenlabs.py

@@ -55,7 +55,7 @@ ELEVENLABS_MULTILINGUAL_MODELS = {
 }
 
 
-def language_to_elevenlabs_language(language: Language) -> str | None:
+def language_to_elevenlabs_language(language: Language) -> Optional[str]:
     BASE_LANGUAGES = {
         Language.AR: "ar",
         Language.BG: "bg",
@@ -223,7 +223,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
     def can_generate_metrics(self) -> bool:
         return True
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_elevenlabs_language(language)
 
     def _set_voice_settings(self):

src/pipecat/services/fal.py

@@ -42,7 +42,7 @@ class FalImageGenService(ImageGenService):
         params: InputParams,
         aiohttp_session: aiohttp.ClientSession,
         model: str = "fal-ai/fast-sdxl",
-        key: str | None = None,
+        key: Optional[str] = None,
         **kwargs,
     ):
         super().__init__(**kwargs)

src/pipecat/services/gladia.py

@@ -34,7 +34,7 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
-def language_to_gladia_language(language: Language) -> str | None:
+def language_to_gladia_language(language: Language) -> Optional[str]:
     BASE_LANGUAGES = {
         Language.AF: "af",
         Language.AM: "am",
@@ -173,7 +173,7 @@ class GladiaSTTService(STTService):
         }
         self._confidence = confidence
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_gladia_language(language)
 
     async def start(self, frame: StartFrame):

src/pipecat/services/google/__init__.py

@@ -1,2 +1,3 @@
 from .frames import LLMSearchResponseFrame
 from .google import *
+from .rtvi import *

src/pipecat/services/google/google.py

@@ -63,7 +63,7 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
-def language_to_google_language(language: Language) -> str | None:
+def language_to_google_language(language: Language) -> Optional[str]:
     language_map = {
         # Afrikaans
         Language.AF: "af-ZA",
@@ -346,9 +346,9 @@ class GoogleContextAggregatorPair:
 class GoogleLLMContext(OpenAILLMContext):
     def __init__(
         self,
-        messages: list[dict] | None = None,
-        tools: list[dict] | None = None,
-        tool_choice: dict | None = None,
+        messages: Optional[List[dict]] = None,
+        tools: Optional[List[dict]] = None,
+        tool_choice: Optional[dict] = None,
     ):
         super().__init__(messages=messages, tools=tools, tool_choice=tool_choice)
         self.system_message = None
@@ -639,7 +639,7 @@ class GoogleLLMService(LLMService):
         self,
         *,
         api_key: str,
-        model: str = "gemini-1.5-flash-latest",
+        model: str = "gemini-2.0-flash-001",
         params: InputParams = InputParams(),
         system_instruction: Optional[str] = None,
         tools: Optional[List[Dict[str, Any]]] = None,
@@ -926,7 +926,7 @@ class GoogleTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_google_language(language)
 
     def _construct_ssml(self, text: str) -> str:

src/pipecat/services/google/rtvi.py

@@ -0,0 +1,54 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from typing import List, Literal, Optional
+
+from pydantic import BaseModel
+
+from pipecat.frames.frames import Frame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.processors.frameworks.rtvi import RTVIObserver
+from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame
+
+
+class RTVISearchResponseMessageData(BaseModel):
+    search_result: Optional[str]
+    rendered_content: Optional[str]
+    origins: List[LLMSearchOrigin]
+
+
+class RTVIBotLLMSearchResponseMessage(BaseModel):
+    label: Literal["rtvi-ai"] = "rtvi-ai"
+    type: Literal["bot-llm-search-response"] = "bot-llm-search-response"
+    data: RTVISearchResponseMessageData
+
+
+class GoogleRTVIObserver(RTVIObserver):
+    def __init__(self, rtvi: FrameProcessor):
+        super().__init__(rtvi)
+
+    async def on_push_frame(
+        self,
+        src: FrameProcessor,
+        dst: FrameProcessor,
+        frame: Frame,
+        direction: FrameDirection,
+        timestamp: int,
+    ):
+        await super().on_push_frame(src, dst, frame, direction, timestamp)
+
+        if isinstance(frame, LLMSearchResponseFrame):
+            await self._handle_llm_search_response_frame(frame)
+
+    async def _handle_llm_search_response_frame(self, frame: LLMSearchResponseFrame):
+        message = RTVIBotLLMSearchResponseMessage(
+            data=RTVISearchResponseMessageData(
+                search_result=frame.search_result,
+                origins=frame.origins,
+                rendered_content=frame.rendered_content,
+            )
+        )
+        await self.push_transport_message_urgent(message)

src/pipecat/services/lmnt.py

@@ -36,7 +36,7 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
-def language_to_lmnt_language(language: Language) -> str | None:
+def language_to_lmnt_language(language: Language) -> Optional[str]:
     BASE_LANGUAGES = {
         Language.DE: "de",
         Language.EN: "en",
@@ -89,7 +89,7 @@ class LmntTTSService(TTSService, WebsocketService):
     def can_generate_metrics(self) -> bool:
         return True
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_lmnt_language(language)
 
     async def start(self, frame: StartFrame):

src/pipecat/services/openai.py

@@ -30,6 +30,7 @@ from pipecat.frames.frames import (
     OpenAILLMContextAssistantTimestampFrame,
     StartFrame,
     StartInterruptionFrame,
+    TranscriptionFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
@@ -48,7 +49,12 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import ImageGenService, LLMService, TTSService
+from pipecat.services.ai_services import (
+    ImageGenService,
+    LLMService,
+    SegmentedSTTService,
+    TTSService,
+)
 from pipecat.utils.time import time_now_iso8601
 
 try:
@@ -59,6 +65,7 @@ try:
         BadRequestError,
         DefaultAsyncHttpxClient,
     )
+    from openai.types.audio import Transcription
     from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
@@ -391,6 +398,61 @@ class OpenAIImageGenService(ImageGenService):
             yield frame
 
 
+class OpenAISTTService(SegmentedSTTService):
+    """OpenAI Speech-to-Text (STT) service.
+
+    This service uses OpenAI's Whisper API to convert audio to text.
+
+    Args:
+        model: Whisper model to use. Defaults to "whisper-1".
+        api_key: OpenAI API key. Defaults to None.
+        base_url: API base URL. Defaults to None.
+        **kwargs: Additional arguments passed to SegmentedSTTService.
+    """
+
+    def __init__(
+        self,
+        *,
+        model: str = "whisper-1",
+        api_key: Optional[str] = None,
+        base_url: Optional[str] = None,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.set_model_name(model)
+        self._client = AsyncOpenAI(api_key=api_key, base_url=base_url)
+
+    async def set_model(self, model: str):
+        self.set_model_name(model)
+
+    def can_generate_metrics(self) -> bool:
+        return True
+
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
+        try:
+            await self.start_processing_metrics()
+            await self.start_ttfb_metrics()
+
+            response: Transcription = await self._client.audio.transcriptions.create(
+                file=("audio.wav", audio, "audio/wav"), model=self.model_name
+            )
+
+            await self.stop_ttfb_metrics()
+            await self.stop_processing_metrics()
+
+            text = response.text.strip()
+
+            if text:
+                logger.debug(f"Transcription: [{text}]")
+                yield TranscriptionFrame(text, "", time_now_iso8601())
+            else:
+                logger.warning("Received empty transcription from API")
+
+        except Exception as e:
+            logger.exception(f"Exception during transcription: {e}")
+            yield ErrorFrame(f"Error during transcription: {str(e)}")
+
+
 class OpenAITTSService(TTSService):
     """OpenAI Text-to-Speech service that generates audio from text.
 

src/pipecat/services/openpipe.py

@@ -4,7 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 from loguru import logger
 
@@ -28,11 +28,11 @@ class OpenPipeLLMService(OpenAILLMService):
         self,
         *,
         model: str = "gpt-4o",
-        api_key: str | None = None,
-        base_url: str | None = None,
-        openpipe_api_key: str | None = None,
+        api_key: Optional[str] = None,
+        base_url: Optional[str] = None,
+        openpipe_api_key: Optional[str] = None,
         openpipe_base_url: str = "https://app.openpipe.ai/api/v1",
-        tags: Dict[str, str] | None = None,
+        tags: Optional[Dict[str, str]] = None,
         **kwargs,
     ):
         super().__init__(

src/pipecat/services/openrouter.py

@@ -4,23 +4,12 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from typing import Dict, List
+from typing import Optional
 
 from loguru import logger
 
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai import OpenAILLMService
 
-try:
-    from openai import AsyncStream, OpenAI
-    from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
-except ModuleNotFoundError as e:
-    logger.error(f"Exception: {e}")
-    logger.error(
-        "In order to use OpenRouter, you need to `pip install pipecat-ai[openrouter]`. Also, set `OPENROUTER_API_KEY` environment variable."
-    )
-    raise Exception(f"Missing module: {e}")
-
 
 class OpenRouterLLMService(OpenAILLMService):
     """A service for interacting with OpenRouter's API using the OpenAI-compatible interface.
@@ -38,7 +27,7 @@ class OpenRouterLLMService(OpenAILLMService):
     def __init__(
         self,
         *,
-        api_key: str | None = None,
+        api_key: Optional[str] = None,
         model: str = "openai/gpt-4o-2024-11-20",
         base_url: str = "https://openrouter.ai/api/v1",
         **kwargs,

src/pipecat/services/playht.py

@@ -46,7 +46,7 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
-def language_to_playht_language(language: Language) -> str | None:
+def language_to_playht_language(language: Language) -> Optional[str]:
     BASE_LANGUAGES = {
         Language.AF: "afrikans",
         Language.AM: "amharic",
@@ -146,7 +146,7 @@ class PlayHTTTSService(TTSService, WebsocketService):
     def can_generate_metrics(self) -> bool:
         return True
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_playht_language(language)
 
     async def start(self, frame: StartFrame):
@@ -389,7 +389,7 @@ class PlayHTHttpTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_playht_language(language)
 
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:

src/pipecat/services/whisper.py

@@ -8,7 +8,7 @@
 
 import asyncio
 from enum import Enum
-from typing import AsyncGenerator
+from typing import AsyncGenerator, Optional
 
 import numpy as np
 from loguru import logger
@@ -53,7 +53,7 @@ class WhisperSTTService(SegmentedSTTService):
         self._compute_type = compute_type
         self.set_model_name(model if isinstance(model, str) else model.value)
         self._no_speech_prob = no_speech_prob
-        self._model: WhisperModel | None = None
+        self._model: Optional[WhisperModel] = None
         self._load()
 
     def can_generate_metrics(self) -> bool:

src/pipecat/services/xtts.py

@@ -29,7 +29,7 @@ from pipecat.transcriptions.language import Language
 # https://github.com/coqui-ai/xtts-streaming-server
 
 
-def language_to_xtts_language(language: Language) -> str | None:
+def language_to_xtts_language(language: Language) -> Optional[str]:
     BASE_LANGUAGES = {
         Language.CS: "cs",
         Language.DE: "de",
@@ -86,7 +86,7 @@ class XTTSService(TTSService):
             "base_url": base_url,
         }
         self.set_voice(voice_id)
-        self._studio_speakers: Dict[str, Any] | None = None
+        self._studio_speakers: Optional[Dict[str, Any]] = None
         self._aiohttp_session = aiohttp_session
 
         self._resampler = create_default_resampler()
@@ -94,7 +94,7 @@ class XTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
-    def language_to_service_language(self, language: Language) -> str | None:
+    def language_to_service_language(self, language: Language) -> Optional[str]:
         return language_to_xtts_language(language)
 
     async def start(self, frame: StartFrame):

src/pipecat/tests/utils.py

@@ -5,11 +5,8 @@
 #
 
 import asyncio
-import sys
 from typing import Any, Awaitable, Callable, Dict, Sequence, Tuple
 
-from loguru import logger
-
 from pipecat.frames.frames import (
     EndFrame,
     Frame,
@@ -22,9 +19,6 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
-logger.remove(0)
-logger.add(sys.stderr, level="TRACE")
-
 
 class HeartbeatsObserver(BaseObserver):
     def __init__(

src/pipecat/transports/base_input.py

@@ -6,6 +6,7 @@
 
 import asyncio
 from concurrent.futures import ThreadPoolExecutor
+from typing import Optional
 
 from loguru import logger
 
@@ -51,7 +52,7 @@ class BaseInputTransport(FrameProcessor):
         return self._sample_rate
 
     @property
-    def vad_analyzer(self) -> VADAnalyzer | None:
+    def vad_analyzer(self) -> Optional[VADAnalyzer]:
         return self._params.vad_analyzer
 
     async def start(self, frame: StartFrame):
@@ -130,17 +131,18 @@ class BaseInputTransport(FrameProcessor):
     #
 
     async def _handle_interruptions(self, frame: Frame):
-        if self.interruptions_allowed:
+        if isinstance(frame, UserStartedSpeakingFrame):
+            logger.debug("User started speaking")
             # Make sure we notify about interruptions quickly out-of-band.
-            if isinstance(frame, UserStartedSpeakingFrame):
-                logger.debug("User started speaking")
+            if self.interruptions_allowed:
                 await self._start_interruption()
                 # Push an out-of-band frame (i.e. not using the ordered push
                 # frame task) to stop everything, specially at the output
                 # transport.
                 await self.push_frame(StartInterruptionFrame())
-            elif isinstance(frame, UserStoppedSpeakingFrame):
-                logger.debug("User stopped speaking")
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            logger.debug("User stopped speaking")
+            if self.interruptions_allowed:
                 await self._stop_interruption()
                 await self.push_frame(StopInterruptionFrame())
 

src/pipecat/transports/base_output.py

@@ -13,6 +13,7 @@ from typing import AsyncGenerator, List
 from loguru import logger
 from PIL import Image
 
+from pipecat.audio.utils import create_default_resampler
 from pipecat.audio.vad.vad_analyzer import VAD_STOP_SECS
 from pipecat.frames.frames import (
     BotSpeakingFrame,
@@ -59,6 +60,7 @@ class BaseOutputTransport(FrameProcessor):
 
         # Output sample rate. It will be initialized on StartFrame.
         self._sample_rate = 0
+        self._resampler = create_default_resampler()
 
         # Chunk size that will be written. It will be computed on StartFrame
         self._audio_chunk_size = 0
@@ -188,12 +190,18 @@ class BaseOutputTransport(FrameProcessor):
         if not self._params.audio_out_enabled:
             return
 
+        # We might need to resample if incoming audio doesn't match the
+        # transport sample rate.
+        resampled = await self._resampler.resample(
+            frame.audio, frame.sample_rate, self._sample_rate
+        )
+
         cls = type(frame)
-        self._audio_buffer.extend(frame.audio)
+        self._audio_buffer.extend(resampled)
         while len(self._audio_buffer) >= self._audio_chunk_size:
             chunk = cls(
                 bytes(self._audio_buffer[: self._audio_chunk_size]),
-                sample_rate=frame.sample_rate,
+                sample_rate=self._sample_rate,
                 num_channels=frame.num_channels,
             )
             await self._sink_queue.put(chunk)

src/pipecat/transports/base_transport.py

@@ -41,7 +41,7 @@ class TransportParams(BaseModel):
     audio_in_filter: Optional[BaseAudioFilter] = None
     vad_enabled: bool = False
     vad_audio_passthrough: bool = False
-    vad_analyzer: VADAnalyzer | None = None
+    vad_analyzer: Optional[VADAnalyzer] = None
 
 
 class BaseTransport(ABC):

src/pipecat/transports/local/audio.py

@@ -6,6 +6,7 @@
 
 import asyncio
 from concurrent.futures import ThreadPoolExecutor
+from typing import Optional
 
 from loguru import logger
 
@@ -25,8 +26,15 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
+class LocalTransportParams(TransportParams):
+    input_device_index: int = 0
+    output_device_index: int = 0
+
+
 class LocalAudioInputTransport(BaseInputTransport):
-    def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
+    _params: LocalTransportParams
+
+    def __init__(self, py_audio: pyaudio.PyAudio, params: LocalTransportParams):
         super().__init__(params)
         self._py_audio = py_audio
         self._in_stream = None
@@ -45,6 +53,7 @@ class LocalAudioInputTransport(BaseInputTransport):
             frames_per_buffer=num_frames,
             stream_callback=self._audio_in_callback,
             input=True,
+            input_device_index=self._params.input_device_index,
         )
         self._in_stream.start_stream()
 
@@ -52,9 +61,6 @@ class LocalAudioInputTransport(BaseInputTransport):
         await super().cleanup()
         if self._in_stream:
             self._in_stream.stop_stream()
-            # This is not very pretty (taken from PyAudio docs).
-            while self._in_stream.is_active():
-                await asyncio.sleep(0.1)
             self._in_stream.close()
             self._in_stream = None
 
@@ -71,6 +77,8 @@ class LocalAudioInputTransport(BaseInputTransport):
 
 
 class LocalAudioOutputTransport(BaseOutputTransport):
+    _params: LocalTransportParams
+
     def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
         super().__init__(params)
         self._py_audio = py_audio
@@ -91,6 +99,7 @@ class LocalAudioOutputTransport(BaseOutputTransport):
             channels=self._params.audio_out_channels,
             rate=self._sample_rate,
             output=True,
+            output_device_index=self._params.output_device_index,
         )
         self._out_stream.start_stream()
 
@@ -98,9 +107,6 @@ class LocalAudioOutputTransport(BaseOutputTransport):
         await super().cleanup()
         if self._out_stream:
             self._out_stream.stop_stream()
-            # This is not very pretty (taken from PyAudio docs).
-            while self._out_stream.is_active():
-                await asyncio.sleep(0.1)
             self._out_stream.close()
 
     async def write_raw_audio_frames(self, frames: bytes):
@@ -110,14 +116,14 @@ class LocalAudioOutputTransport(BaseOutputTransport):
             )
 
 
-class LocalAudioTransport(BaseTransport):
-    def __init__(self, params: TransportParams):
+class LocalAudioTransport(LocalTransportParams):
+    def __init__(self, params: LocalTransportParams):
         super().__init__()
         self._params = params
         self._pyaudio = pyaudio.PyAudio()
 
-        self._input: LocalAudioInputTransport | None = None
-        self._output: LocalAudioOutputTransport | None = None
+        self._input: Optional[LocalAudioInputTransport] = None
+        self._output: Optional[LocalAudioOutputTransport] = None
 
     #
     # BaseTransport

src/pipecat/transports/local/tk.py

@@ -7,6 +7,7 @@
 import asyncio
 import tkinter as tk
 from concurrent.futures import ThreadPoolExecutor
+from typing import Optional
 
 import numpy as np
 from loguru import logger
@@ -60,9 +61,6 @@ class TkInputTransport(BaseInputTransport):
         await super().cleanup()
         if self._in_stream:
             self._in_stream.stop_stream()
-            # This is not very pretty (taken from PyAudio docs).
-            while self._in_stream.is_active():
-                await asyncio.sleep(0.1)
             self._in_stream.close()
 
     def _audio_in_callback(self, in_data, frame_count, time_info, status):
@@ -112,9 +110,6 @@ class TkOutputTransport(BaseOutputTransport):
         await super().cleanup()
         if self._out_stream:
             self._out_stream.stop_stream()
-            # This is not very pretty (taken from PyAudio docs).
-            while self._out_stream.is_active():
-                await asyncio.sleep(0.1)
             self._out_stream.close()
 
     async def write_raw_audio_frames(self, frames: bytes):
@@ -145,8 +140,8 @@ class TkLocalTransport(BaseTransport):
         self._params = params
         self._pyaudio = pyaudio.PyAudio()
 
-        self._input: TkInputTransport | None = None
-        self._output: TkOutputTransport | None = None
+        self._input: Optional[TkInputTransport] = None
+        self._output: Optional[TkOutputTransport] = None
 
     #
     # BaseTransport

src/pipecat/transports/network/fastapi_websocket.py

@@ -10,7 +10,7 @@ import io
 import time
 import typing
 import wave
-from typing import Awaitable, Callable
+from typing import Awaitable, Callable, Optional
 
 from loguru import logger
 from pydantic import BaseModel
@@ -44,7 +44,7 @@ except ModuleNotFoundError as e:
 class FastAPIWebsocketParams(TransportParams):
     add_wav_header: bool = False
     serializer: FrameSerializer
-    session_timeout: int | None = None
+    session_timeout: Optional[int] = None
 
 
 class FastAPIWebsocketCallbacks(BaseModel):
@@ -202,8 +202,8 @@ class FastAPIWebsocketTransport(BaseTransport):
         self,
         websocket: WebSocket,
         params: FastAPIWebsocketParams,
-        input_name: str | None = None,
-        output_name: str | None = None,
+        input_name: Optional[str] = None,
+        output_name: Optional[str] = None,
     ):
         super().__init__(input_name=input_name, output_name=output_name)
         self._params = params

src/pipecat/transports/network/websocket_client.py

@@ -59,7 +59,7 @@ class WebsocketClientSession:
 
         self._task_manager: Optional[TaskManager] = None
 
-        self._websocket: websockets.WebSocketClientProtocol | None = None
+        self._websocket: Optional[websockets.WebSocketClientProtocol] = None
 
     @property
     def task_manager(self) -> TaskManager:
@@ -240,8 +240,8 @@ class WebsocketClientTransport(BaseTransport):
         )
 
         self._session = WebsocketClientSession(uri, params, callbacks, self.name)
-        self._input: WebsocketClientInputTransport | None = None
-        self._output: WebsocketClientOutputTransport | None = None
+        self._input: Optional[WebsocketClientInputTransport] = None
+        self._output: Optional[WebsocketClientOutputTransport] = None
 
         # Register supported handlers. The user will only be able to register
         # these handlers.

src/pipecat/transports/network/websocket_server.py

@@ -8,7 +8,7 @@ import asyncio
 import io
 import time
 import wave
-from typing import Awaitable, Callable
+from typing import Awaitable, Callable, Optional
 
 from loguru import logger
 from pydantic import BaseModel
@@ -39,7 +39,7 @@ except ModuleNotFoundError as e:
 class WebsocketServerParams(TransportParams):
     add_wav_header: bool = False
     serializer: FrameSerializer
-    session_timeout: int | None = None
+    session_timeout: Optional[int] = None
 
 
 class WebsocketServerCallbacks(BaseModel):
@@ -64,7 +64,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
         self._params = params
         self._callbacks = callbacks
 
-        self._websocket: websockets.WebSocketServerProtocol | None = None
+        self._websocket: Optional[websockets.WebSocketServerProtocol] = None
 
         self._server_task = None
 
@@ -158,7 +158,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
 
         self._params = params
 
-        self._websocket: websockets.WebSocketServerProtocol | None = None
+        self._websocket: Optional[websockets.WebSocketServerProtocol] = None
 
         # write_raw_audio_frames() is called quickly, as soon as we get audio
         # (e.g. from the TTS), and since this is just a network connection we
@@ -168,7 +168,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
         self._send_interval = 0
         self._next_send_time = 0
 
-    async def set_client_connection(self, websocket: websockets.WebSocketServerProtocol | None):
+    async def set_client_connection(self, websocket: Optional[websockets.WebSocketServerProtocol]):
         if self._websocket:
             await self._websocket.close()
             logger.warning("Only one client allowed, using new connection")
@@ -242,8 +242,8 @@ class WebsocketServerTransport(BaseTransport):
         params: WebsocketServerParams,
         host: str = "localhost",
         port: int = 8765,
-        input_name: str | None = None,
-        output_name: str | None = None,
+        input_name: Optional[str] = None,
+        output_name: Optional[str] = None,
     ):
         super().__init__(input_name=input_name, output_name=output_name)
         self._host = host
@@ -255,9 +255,9 @@ class WebsocketServerTransport(BaseTransport):
             on_client_disconnected=self._on_client_disconnected,
             on_session_timeout=self._on_session_timeout,
         )
-        self._input: WebsocketServerInputTransport | None = None
-        self._output: WebsocketServerOutputTransport | None = None
-        self._websocket: websockets.WebSocketServerProtocol | None = None
+        self._input: Optional[WebsocketServerInputTransport] = None
+        self._output: Optional[WebsocketServerOutputTransport] = None
+        self._websocket: Optional[websockets.WebSocketServerProtocol] = None
 
         # Register supported handlers. The user will only be able to register
         # these handlers.

src/pipecat/transports/services/daily.py

@@ -62,12 +62,12 @@ VAD_RESET_PERIOD_MS = 2000
 
 @dataclass
 class DailyTransportMessageFrame(TransportMessageFrame):
-    participant_id: str | None = None
+    participant_id: Optional[str] = None
 
 
 @dataclass
 class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame):
-    participant_id: str | None = None
+    participant_id: Optional[str] = None
 
 
 class WebRTCVADAnalyzer(VADAnalyzer):
@@ -175,7 +175,7 @@ class DailyTransportClient(EventHandler):
     def __init__(
         self,
         room_url: str,
-        token: str | None,
+        token: Optional[str],
         bot_name: str,
         params: DailyParams,
         callbacks: DailyCallbacks,
@@ -188,7 +188,7 @@ class DailyTransportClient(EventHandler):
             Daily.init()
 
         self._room_url: str = room_url
-        self._token: str | None = token
+        self._token: Optional[str] = token
         self._bot_name: str = bot_name
         self._params: DailyParams = params
         self._callbacks = callbacks
@@ -226,9 +226,9 @@ class DailyTransportClient(EventHandler):
         self._in_sample_rate = 0
         self._out_sample_rate = 0
 
-        self._camera: VirtualCameraDevice | None = None
-        self._mic: VirtualMicrophoneDevice | None = None
-        self._speaker: VirtualSpeakerDevice | None = None
+        self._camera: Optional[VirtualCameraDevice] = None
+        self._mic: Optional[VirtualMicrophoneDevice] = None
+        self._speaker: Optional[VirtualSpeakerDevice] = None
 
     def _camera_name(self):
         return f"camera-{self}"
@@ -257,7 +257,7 @@ class DailyTransportClient(EventHandler):
         )
         await future
 
-    async def read_next_audio_frame(self) -> InputAudioRawFrame | None:
+    async def read_next_audio_frame(self) -> Optional[InputAudioRawFrame]:
         if not self._speaker:
             return None
 
@@ -542,7 +542,7 @@ class DailyTransportClient(EventHandler):
         self._client.stop_recording(stream_id, completion=completion_callback(future))
         await future
 
-    async def send_prebuilt_chat_message(self, message: str, user_name: str | None = None):
+    async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None):
         if not self._joined:
             return
 
@@ -723,10 +723,10 @@ class DailyInputTransport(BaseInputTransport):
         # internally to be processed.
         self._audio_in_task = None
 
-        self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer
+        self._vad_analyzer: Optional[VADAnalyzer] = params.vad_analyzer
 
     @property
-    def vad_analyzer(self) -> VADAnalyzer | None:
+    def vad_analyzer(self) -> Optional[VADAnalyzer]:
         return self._vad_analyzer
 
     async def start(self, frame: StartFrame):
@@ -891,11 +891,11 @@ class DailyTransport(BaseTransport):
     def __init__(
         self,
         room_url: str,
-        token: str | None,
+        token: Optional[str],
         bot_name: str,
         params: DailyParams = DailyParams(),
-        input_name: str | None = None,
-        output_name: str | None = None,
+        input_name: Optional[str] = None,
+        output_name: Optional[str] = None,
     ):
         super().__init__(input_name=input_name, output_name=output_name)
 
@@ -926,8 +926,8 @@ class DailyTransport(BaseTransport):
         self._params = params
 
         self._client = DailyTransportClient(room_url, token, bot_name, params, callbacks, self.name)
-        self._input: DailyInputTransport | None = None
-        self._output: DailyOutputTransport | None = None
+        self._input: Optional[DailyInputTransport] = None
+        self._output: Optional[DailyOutputTransport] = None
 
         self._other_participant_has_joined = False
 
@@ -1014,7 +1014,7 @@ class DailyTransport(BaseTransport):
     async def stop_recording(self, stream_id=None):
         await self._client.stop_recording(stream_id)
 
-    async def send_prebuilt_chat_message(self, message: str, user_name: str | None = None):
+    async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None):
         """Sends a chat message to Daily's Prebuilt main room.
 
         Args:

src/pipecat/transports/services/livekit.py

@@ -40,12 +40,12 @@ except ModuleNotFoundError as e:
 
 @dataclass
 class LiveKitTransportMessageFrame(TransportMessageFrame):
-    participant_id: str | None = None
+    participant_id: Optional[str] = None
 
 
 @dataclass
 class LiveKitTransportMessageUrgentFrame(TransportMessageUrgentFrame):
-    participant_id: str | None = None
+    participant_id: Optional[str] = None
 
 
 class LiveKitParams(TransportParams):
@@ -79,12 +79,12 @@ class LiveKitTransportClient:
         self._params = params
         self._callbacks = callbacks
         self._transport_name = transport_name
-        self._room: rtc.Room | None = None
+        self._room: Optional[rtc.Room] = None
         self._participant_id: str = ""
         self._connected = False
         self._disconnect_counter = 0
-        self._audio_source: rtc.AudioSource | None = None
-        self._audio_track: rtc.LocalAudioTrack | None = None
+        self._audio_source: Optional[rtc.AudioSource] = None
+        self._audio_track: Optional[rtc.LocalAudioTrack] = None
         self._audio_tracks = {}
         self._audio_queue = asyncio.Queue()
         self._other_participant_has_joined = False
@@ -172,7 +172,7 @@ class LiveKitTransportClient:
         logger.info(f"Disconnected from {self._room_name}")
         await self._callbacks.on_disconnected()
 
-    async def send_data(self, data: bytes, participant_id: str | None = None):
+    async def send_data(self, data: bytes, participant_id: Optional[str] = None):
         if not self._connected:
             return
 
@@ -349,11 +349,11 @@ class LiveKitInputTransport(BaseInputTransport):
         super().__init__(params, **kwargs)
         self._client = client
         self._audio_in_task = None
-        self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer
+        self._vad_analyzer: Optional[VADAnalyzer] = params.vad_analyzer
         self._resampler = create_default_resampler()
 
     @property
-    def vad_analyzer(self) -> VADAnalyzer | None:
+    def vad_analyzer(self) -> Optional[VADAnalyzer]:
         return self._vad_analyzer
 
     async def start(self, frame: StartFrame):
@@ -463,8 +463,8 @@ class LiveKitTransport(BaseTransport):
         token: str,
         room_name: str,
         params: LiveKitParams = LiveKitParams(),
-        input_name: str | None = None,
-        output_name: str | None = None,
+        input_name: Optional[str] = None,
+        output_name: Optional[str] = None,
     ):
         super().__init__(input_name=input_name, output_name=output_name)
 
@@ -483,8 +483,8 @@ class LiveKitTransport(BaseTransport):
         self._client = LiveKitTransportClient(
             url, token, room_name, self._params, callbacks, self.name
         )
-        self._input: LiveKitInputTransport | None = None
-        self._output: LiveKitOutputTransport | None = None
+        self._input: Optional[LiveKitInputTransport] = None
+        self._output: Optional[LiveKitOutputTransport] = None
 
         self._register_event_handler("on_connected")
         self._register_event_handler("on_disconnected")
@@ -562,12 +562,12 @@ class LiveKitTransport(BaseTransport):
             await self._input.push_app_message(data.decode(), participant_id)
         await self._call_event_handler("on_data_received", data, participant_id)
 
-    async def send_message(self, message: str, participant_id: str | None = None):
+    async def send_message(self, message: str, participant_id: Optional[str] = None):
         if self._output:
             frame = LiveKitTransportMessageFrame(message=message, participant_id=participant_id)
             await self._output.send_message(frame)
 
-    async def send_message_urgent(self, message: str, participant_id: str | None = None):
+    async def send_message_urgent(self, message: str, participant_id: Optional[str] = None):
         if self._output:
             frame = LiveKitTransportMessageUrgentFrame(
                 message=message, participant_id=participant_id