Merge remote-tracking branch 'origin/main' into openrouter-integration

src/pipecat/audio/filters/base_audio_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/audio/filters/koala_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/audio/filters/krisp_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/audio/filters/noisereduce_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/audio/mixers/base_audio_mixer.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/audio/mixers/soundfile_mixer.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/audio/utils.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/audio/vad/silero.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/audio/vad/vad_analyzer.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/clocks/base_clock.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/clocks/system_clock.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/frames/frames.proto

@@ -1,5 +1,5 @@
 //
-// Copyright (c) 2025, Daily
+// Copyright (c) 2024–2025, Daily
 //
 // SPDX-License-Identifier: BSD 2-Clause License
 //

src/pipecat/frames/frames.py

@@ -1,11 +1,11 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
 from dataclasses import dataclass, field
-from typing import Any, List, Literal, Mapping, Optional, Tuple
+from typing import Any, Awaitable, Callable, List, Literal, Mapping, Optional, Tuple
 
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.clocks.base_clock import BaseClock
@@ -321,6 +321,14 @@ class LLMEnablePromptCachingFrame(DataFrame):
     enable: bool
 
 
+@dataclass
+class FunctionCallResultProperties:
+    """Properties for a function call result frame."""
+
+    run_llm: Optional[bool] = None
+    on_context_updated: Optional[Callable[[], Awaitable[None]]] = None
+
+
 @dataclass
 class FunctionCallResultFrame(DataFrame):
     """A frame containing the result of an LLM function (tool) call."""
@@ -329,7 +337,7 @@ class FunctionCallResultFrame(DataFrame):
     tool_call_id: str
     arguments: str
     result: Any
-    run_llm: bool = True
+    properties: Optional[FunctionCallResultProperties] = None
 
 
 @dataclass

src/pipecat/pipeline/base_pipeline.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/pipeline/parallel_pipeline.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/pipeline/pipeline.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/pipeline/runner.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/pipeline/sync_parallel_pipeline.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/pipeline/task.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
@@ -180,6 +180,7 @@ class PipelineTask:
         if should_cleanup:
             await self._source.cleanup()
             await self._pipeline.cleanup()
+            await self._sink.cleanup()
         # We just enqueue None to terminate the task gracefully.
         self._process_up_task.cancel()
         await self._process_up_task

src/pipecat/processors/aggregators/gated.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/aggregators/gated_openai_llm_context.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/aggregators/llm_response.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
@@ -9,7 +9,7 @@ import copy
 import io
 import json
 from dataclasses import dataclass
-from typing import Any, Awaitable, Callable, List
+from typing import Any, Awaitable, Callable, List, Optional
 
 from loguru import logger
 from PIL import Image
@@ -218,23 +218,22 @@ class OpenAILLMContext:
         await llm.push_frame(progress_frame_upstream, FrameDirection.UPSTREAM)
 
         # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream.
-        async def function_call_result_callback(result):
+        async def function_call_result_callback(result, *, properties=None):
             result_frame_downstream = FunctionCallResultFrame(
                 function_name=function_name,
                 tool_call_id=tool_call_id,
                 arguments=arguments,
                 result=result,
-                run_llm=run_llm,
+                properties=properties,
             )
             result_frame_upstream = FunctionCallResultFrame(
                 function_name=function_name,
                 tool_call_id=tool_call_id,
                 arguments=arguments,
                 result=result,
-                run_llm=run_llm,
+                properties=properties,
             )
 
-            # Push frame both downstream and upstream
             await llm.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM)
             await llm.push_frame(result_frame_upstream, FrameDirection.UPSTREAM)
 

src/pipecat/processors/aggregators/sentence.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/aggregators/user_response.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/aggregators/vision_image_frame.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/async_generator.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/audio/audio_buffer_processor.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/audio/vad/silero.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/filters/frame_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/filters/function_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/filters/identity_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/filters/null_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/filters/stt_mute_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/filters/wake_check_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/filters/wake_notifier_filter.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/frame_processor.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/frameworks/langchain.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/frameworks/rtvi.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/gstreamer/pipeline_source.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/idle_frame_processor.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/logger.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/metrics/frame_processor_metrics.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/metrics/sentry.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/text_transformer.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/transcript_processor.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/processors/user_idle_processor.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/serializers/base_serializer.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/serializers/livekit.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/serializers/protobuf.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/serializers/twilio.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/ai_services.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
@@ -25,6 +25,7 @@ from pipecat.frames.frames import (
     STTMuteFrame,
     STTUpdateSettingsFrame,
     TextFrame,
+    TranscriptionFrame,
     TTSAudioRawFrame,
     TTSSpeakFrame,
     TTSStartedFrame,
@@ -290,7 +291,7 @@ class TTSService(AIService):
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
-        if isinstance(frame, TextFrame):
+        if isinstance(frame, TextFrame) and not isinstance(frame, TranscriptionFrame):
             await self._process_text_frame(frame)
         elif isinstance(frame, StartInterruptionFrame):
             await self._handle_interruption(frame, direction)

src/pipecat/services/anthropic.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
@@ -21,6 +21,7 @@ from pipecat.frames.frames import (
     Frame,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
+    FunctionCallResultProperties,
     LLMEnablePromptCachingFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
@@ -742,6 +743,7 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
             return
 
         run_llm = False
+        properties: Optional[FunctionCallResultProperties] = None
 
         aggregation = self._aggregation
         self._reset()
@@ -749,6 +751,7 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
         try:
             if self._function_call_result:
                 frame = self._function_call_result
+                properties = frame.properties
                 self._function_call_result = None
                 if frame.result:
                     assistant_message = {"role": "assistant", "content": []}
@@ -775,7 +778,12 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
                             ],
                         }
                     )
-                    run_llm = True
+                    if properties and properties.run_llm is not None:
+                        # If the tool call result has a run_llm property, use it
+                        run_llm = properties.run_llm
+                    else:
+                        # Default behavior
+                        run_llm = True
             elif aggregation:
                 self._context.add_message({"role": "assistant", "content": aggregation})
 
@@ -793,6 +801,10 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Emit the on_context_updated callback once the function call result is added to the context
+            if properties and properties.on_context_updated is not None:
+                await properties.on_context_updated()
+
             # Push context frame
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)

src/pipecat/services/assemblyai.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/aws.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
@@ -121,6 +121,7 @@ class PollyTTSService(TTSService):
         *,
         api_key: Optional[str] = None,
         aws_access_key_id: Optional[str] = None,
+        aws_session_token: Optional[str] = None,
         region: Optional[str] = None,
         voice_id: str = "Joanna",
         sample_rate: int = 24000,
@@ -133,6 +134,7 @@ class PollyTTSService(TTSService):
             "polly",
             aws_access_key_id=aws_access_key_id,
             aws_secret_access_key=api_key,
+            aws_session_token=aws_session_token,
             region_name=region,
         )
         self._settings = {

src/pipecat/services/azure.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/canonical.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/cartesia.py

@@ -1,10 +1,9 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import asyncio
 import base64
 import json
 import uuid
@@ -12,7 +11,6 @@ from typing import AsyncGenerator, List, Optional, Union
 
 from loguru import logger
 from pydantic import BaseModel
-from tenacity import AsyncRetrying, RetryCallState, stop_after_attempt, wait_exponential
 
 from pipecat.frames.frames import (
     BotStoppedSpeakingFrame,
@@ -30,6 +28,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import TTSService, WordTTSService
+from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
 # See .env.example for Cartesia configuration needed
@@ -76,7 +75,7 @@ def language_to_cartesia_language(language: Language) -> str | None:
     return result
 
 
-class CartesiaTTSService(WordTTSService):
+class CartesiaTTSService(WordTTSService, WebsocketService):
     class InputParams(BaseModel):
         language: Optional[Language] = Language.EN
         speed: Optional[Union[str, float]] = ""
@@ -106,12 +105,14 @@ class CartesiaTTSService(WordTTSService):
         # if we're interrupted. Cartesia gives us word-by-word timestamps. We
         # can use those to generate text frames ourselves aligned with the
         # playout timing of the audio!
-        super().__init__(
+        WordTTSService.__init__(
+            self,
             aggregate_sentences=True,
             push_text_frames=False,
             sample_rate=sample_rate,
             **kwargs,
         )
+        WebsocketService.__init__(self)
 
         self._api_key = api_key
         self._cartesia_version = cartesia_version
@@ -131,7 +132,6 @@ class CartesiaTTSService(WordTTSService):
         self.set_model_name(model)
         self.set_voice(voice_id)
 
-        self._websocket = None
         self._context_id = None
         self._receive_task = None
 
@@ -187,7 +187,9 @@ class CartesiaTTSService(WordTTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
+        self._receive_task = self.get_event_loop().create_task(
+            self._receive_task_handler(self.push_error)
+        )
 
     async def _disconnect(self):
         await self._disconnect_websocket()
@@ -275,30 +277,6 @@ class CartesiaTTSService(WordTTSService):
             else:
                 logger.error(f"{self} error, unknown message type: {msg}")
 
-    async def _reconnect_websocket(self, retry_state: RetryCallState):
-        logger.warning(f"{self} reconnecting (attempt: {retry_state.attempt_number})")
-        await self._disconnect_websocket()
-        await self._connect_websocket()
-
-    async def _receive_task_handler(self):
-        while True:
-            try:
-                async for attempt in AsyncRetrying(
-                    stop=stop_after_attempt(3),
-                    wait=wait_exponential(multiplier=1, min=4, max=10),
-                    before_sleep=self._reconnect_websocket,
-                    reraise=True,
-                ):
-                    with attempt:
-                        await self._receive_messages()
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                message = f"{self} error receiving messages: {e}"
-                logger.error(message)
-                await self.push_error(ErrorFrame(message, fatal=True))
-                break
-
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 

src/pipecat/services/cerebras.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/deepgram.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/deepseek.py

@@ -0,0 +1,88 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+from typing import List
+
+from loguru import logger
+
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.openai import OpenAILLMService
+
+try:
+    from openai import (
+        AsyncStream,
+    )
+    from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error(
+        "In order to use Fireworks, you need to `pip install pipecat-ai[deepseek]`. Also, set `DEEPSEEK_API_KEY` environment variable."
+    )
+    raise Exception(f"Missing module: {e}")
+
+
+class DeepSeekLLMService(OpenAILLMService):
+    """A service for interacting with DeepSeek's API using the OpenAI-compatible interface.
+
+    This service extends OpenAILLMService to connect to DeepSeek's API endpoint while
+    maintaining full compatibility with OpenAI's interface and functionality.
+
+    Args:
+        api_key (str): The API key for accessing DeepSeek's API
+        base_url (str, optional): The base URL for DeepSeek API. Defaults to "https://api.deepseek.com/v1"
+        model (str, optional): The model identifier to use. Defaults to "deepseek-chat"
+        **kwargs: Additional keyword arguments passed to OpenAILLMService
+    """
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        base_url: str = "https://api.deepseek.com/v1",
+        model: str = "deepseek-chat",
+        **kwargs,
+    ):
+        super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs)
+
+    def create_client(self, api_key=None, base_url=None, **kwargs):
+        """Create OpenAI-compatible client for DeepSeek API endpoint."""
+        logger.debug(f"Creating DeepSeek client with api {base_url}")
+        return super().create_client(api_key, base_url, **kwargs)
+
+    async def get_chat_completions(
+        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
+    ) -> AsyncStream[ChatCompletionChunk]:
+        """Create a streaming chat completion using Cerebras's API.
+
+        Args:
+        context (OpenAILLMContext): The context object containing tools configuration
+            and other settings for the chat completion.
+        messages (List[ChatCompletionMessageParam]): The list of messages comprising
+            the conversation history and current request.
+
+        Returns:
+        AsyncStream[ChatCompletionChunk]: A streaming response of chat completion
+            chunks that can be processed asynchronously.
+        """
+        params = {
+            "model": self.model_name,
+            "stream": True,
+            "messages": messages,
+            "tools": context.tools,
+            "tool_choice": context.tool_choice,
+            "stream_options": {"include_usage": True},
+            "frequency_penalty": self._settings["frequency_penalty"],
+            "presence_penalty": self._settings["presence_penalty"],
+            "temperature": self._settings["temperature"],
+            "top_p": self._settings["top_p"],
+            "max_tokens": self._settings["max_tokens"],
+        }
+
+        params.update(self._settings["extra"])
+
+        chunks = await self._client.chat.completions.create(**params)
+        return chunks

src/pipecat/services/elevenlabs.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
@@ -11,13 +11,11 @@ from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional,
 
 from loguru import logger
 from pydantic import BaseModel, model_validator
-from tenacity import AsyncRetrying, RetryCallState, stop_after_attempt, wait_exponential
 
 from pipecat.frames.frames import (
     BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
-    ErrorFrame,
     Frame,
     LLMFullResponseEndFrame,
     StartFrame,
@@ -29,6 +27,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import WordTTSService
+from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
 # See .env.example for ElevenLabs configuration needed
@@ -133,7 +132,7 @@ def calculate_word_times(
     return word_times
 
 
-class ElevenLabsTTSService(WordTTSService):
+class ElevenLabsTTSService(WordTTSService, WebsocketService):
     class InputParams(BaseModel):
         language: Optional[Language] = Language.EN
         optimize_streaming_latency: Optional[str] = None
@@ -178,7 +177,8 @@ class ElevenLabsTTSService(WordTTSService):
         # Finally, ElevenLabs doesn't provide information on when the bot stops
         # speaking for a while, so we want the parent class to send TTSStopFrame
         # after a short period not receiving any audio.
-        super().__init__(
+        WordTTSService.__init__(
+            self,
             aggregate_sentences=True,
             push_text_frames=False,
             push_stop_frames=True,
@@ -186,6 +186,7 @@ class ElevenLabsTTSService(WordTTSService):
             sample_rate=sample_rate_from_output_format(output_format),
             **kwargs,
         )
+        WebsocketService.__init__(self)
 
         self._api_key = api_key
         self._url = url
@@ -206,8 +207,6 @@ class ElevenLabsTTSService(WordTTSService):
         self.set_voice(voice_id)
         self._voice_settings = self._set_voice_settings()
 
-        # Websocket connection to ElevenLabs.
-        self._websocket = None
         # Indicates if we have sent TTSStartedFrame. It will reset to False when
         # there's an interruption or TTSStoppedFrame.
         self._started = False
@@ -297,7 +296,9 @@ class ElevenLabsTTSService(WordTTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
+        self._receive_task = self.get_event_loop().create_task(
+            self._receive_task_handler(self.push_error)
+        )
         self._keepalive_task = self.get_event_loop().create_task(self._keepalive_task_handler())
 
     async def _disconnect(self):
@@ -377,30 +378,6 @@ class ElevenLabsTTSService(WordTTSService):
                 await self.add_word_timestamps(word_times)
                 self._cumulative_time = word_times[-1][1]
 
-    async def _reconnect_websocket(self, retry_state: RetryCallState):
-        logger.warning(f"{self} reconnecting (attempt: {retry_state.attempt_number})")
-        await self._disconnect_websocket()
-        await self._connect_websocket()
-
-    async def _receive_task_handler(self):
-        while True:
-            try:
-                async for attempt in AsyncRetrying(
-                    stop=stop_after_attempt(3),
-                    wait=wait_exponential(multiplier=1, min=4, max=10),
-                    before_sleep=self._reconnect_websocket,
-                    reraise=True,
-                ):
-                    with attempt:
-                        await self._receive_messages()
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                message = f"{self} error receiving messages: {e}"
-                logger.error(message)
-                await self.push_error(ErrorFrame(message, fatal=True))
-                break
-
     async def _keepalive_task_handler(self):
         while True:
             try:

src/pipecat/services/fal.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/fireworks.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/fish.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/gemini_multimodal_live/audio_transcriber.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/gemini_multimodal_live/events.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/gladia.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/google.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
@@ -19,6 +19,7 @@ from pipecat.frames.frames import (
     AudioRawFrame,
     ErrorFrame,
     Frame,
+    FunctionCallResultProperties,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
@@ -245,6 +246,7 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
             return
 
         run_llm = False
+        properties: Optional[FunctionCallResultProperties] = None
 
         aggregation = self._aggregation
         self._reset()
@@ -252,6 +254,7 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
         try:
             if self._function_call_result:
                 frame = self._function_call_result
+                properties = frame.properties
                 self._function_call_result = None
                 if frame.result:
                     logger.debug(f"FunctionCallResultFrame result: {frame.arguments}")
@@ -282,7 +285,12 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
                             ],
                         )
                     )
-                    run_llm = not bool(self._function_calls_in_progress)
+                    if properties and properties.run_llm is not None:
+                        # If the tool call result has a run_llm property, use it
+                        run_llm = properties.run_llm
+                    else:
+                        # Default behavior is to run the LLM if there are no function calls in progress
+                        run_llm = not bool(self._function_calls_in_progress)
             else:
                 if aggregation.strip():
                     self._context.add_message(
@@ -303,6 +311,10 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Emit the on_context_updated callback once the function call result is added to the context
+            if properties and properties.on_context_updated is not None:
+                await properties.on_context_updated()
+
             # Push context frame
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)

src/pipecat/services/grok.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
@@ -7,9 +7,11 @@
 
 import json
 from dataclasses import dataclass
+from typing import Optional
 
 from loguru import logger
 
+from pipecat.frames.frames import FunctionCallResultProperties
 from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
@@ -32,6 +34,7 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
             return
 
         run_llm = False
+        properties: Optional[FunctionCallResultProperties] = None
 
         aggregation = self._aggregation
         self._reset()
@@ -39,6 +42,7 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
         try:
             if self._function_call_result:
                 frame = self._function_call_result
+                properties = frame.properties
                 self._function_call_result = None
                 if frame.result:
                     # Grok requires an empty content field for function calls
@@ -65,8 +69,13 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
                             "tool_call_id": frame.tool_call_id,
                         }
                     )
-                    # Only run the LLM if there are no more function calls in progress.
-                    run_llm = not bool(self._function_calls_in_progress)
+                    if properties and properties.run_llm is not None:
+                        # If the tool call result has a run_llm property, use it
+                        run_llm = properties.run_llm
+                    else:
+                        # Default behavior is to run the LLM if there are no function calls in progress
+                        run_llm = not bool(self._function_calls_in_progress)
+
             else:
                 self._context.add_message({"role": "assistant", "content": aggregation})
 
@@ -84,6 +93,10 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Emit the on_context_updated callback once the function call result is added to the context
+            if properties and properties.on_context_updated is not None:
+                await properties.on_context_updated()
+
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)
 

src/pipecat/services/groq.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/lmnt.py

@@ -1,14 +1,13 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import asyncio
+import json
 from typing import AsyncGenerator
 
 from loguru import logger
-from tenacity import AsyncRetrying, RetryCallState, stop_after_attempt, wait_exponential
 
 from pipecat.frames.frames import (
     CancelFrame,
@@ -23,11 +22,12 @@ from pipecat.frames.frames import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import TTSService
+from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
 # See .env.example for LMNT configuration needed
 try:
-    from lmnt.api import Speech
+    import websockets
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error(
@@ -60,7 +60,7 @@ def language_to_lmnt_language(language: Language) -> str | None:
     return result
 
 
-class LmntTTSService(TTSService):
+class LmntTTSService(TTSService, WebsocketService):
     def __init__(
         self,
         *,
@@ -70,27 +70,21 @@ class LmntTTSService(TTSService):
         language: Language = Language.EN,
         **kwargs,
     ):
-        # Let TTSService produce TTSStoppedFrames after a short delay of
-        # no activity.
-        super().__init__(push_stop_frames=True, sample_rate=sample_rate, **kwargs)
+        TTSService.__init__(
+            self,
+            push_stop_frames=True,
+            sample_rate=sample_rate,
+            **kwargs,
+        )
+        WebsocketService.__init__(self)
 
         self._api_key = api_key
+        self._voice_id = voice_id
         self._settings = {
-            "output_format": {
-                "container": "raw",
-                "encoding": "pcm_s16le",
-                "sample_rate": sample_rate,
-            },
+            "sample_rate": sample_rate,
             "language": self.language_to_service_language(language),
+            "format": "raw",  # Use raw format for direct PCM data
         }
-
-        self.set_voice(voice_id)
-
-        self._speech = None
-        self._connection = None
-        self._receive_task = None
-        # Indicates if we have sent TTSStartedFrame. It will reset to False when
-        # there's an interruption or TTSStoppedFrame.
         self._started = False
 
     def can_generate_metrics(self) -> bool:
@@ -117,106 +111,107 @@ class LmntTTSService(TTSService):
             self._started = False
 
     async def _connect(self):
-        await self._connect_lmnt()
+        await self._connect_websocket()
 
-        self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
+        self._receive_task = self.get_event_loop().create_task(
+            self._receive_task_handler(self.push_error)
+        )
 
     async def _disconnect(self):
-        await self._disconnect_lmnt()
+        await self._disconnect_websocket()
 
         if self._receive_task:
             self._receive_task.cancel()
             await self._receive_task
             self._receive_task = None
 
-    async def _connect_lmnt(self):
+    async def _connect_websocket(self):
+        """Connect to LMNT websocket."""
         try:
             logger.debug("Connecting to LMNT")
 
-            self._speech = Speech()
-            self._connection = await self._speech.synthesize_streaming(
-                self._voice_id,
-                format="raw",
-                sample_rate=self._settings["output_format"]["sample_rate"],
-                language=self._settings["language"],
-            )
+            # Build initial connection message
+            init_msg = {
+                "X-API-Key": self._api_key,
+                "voice": self._voice_id,
+                "format": self._settings["format"],
+                "sample_rate": self._settings["sample_rate"],
+                "language": self._settings["language"],
+            }
+
+            # Connect to LMNT's websocket directly
+            self._websocket = await websockets.connect("wss://api.lmnt.com/v1/ai/speech/stream")
+
+            # Send initialization message
+            await self._websocket.send(json.dumps(init_msg))
+
         except Exception as e:
             logger.error(f"{self} initialization error: {e}")
-            self._connection = None
+            self._websocket = None
 
-    async def _disconnect_lmnt(self):
+    async def _disconnect_websocket(self):
+        """Disconnect from LMNT websocket."""
         try:
             await self.stop_all_metrics()
 
-            if self._connection:
+            if self._websocket:
                 logger.debug("Disconnecting from LMNT")
-                await self._connection.socket.close()
-                self._connection = None
-            if self._speech:
-                await self._speech.close()
-                self._speech = None
+                # Send EOF message before closing
+                await self._websocket.send(json.dumps({"eof": True}))
+                await self._websocket.close()
+                self._websocket = None
 
             self._started = False
         except Exception as e:
-            logger.error(f"{self} error closing connection: {e}")
+            logger.error(f"{self} error closing websocket: {e}")
+
+    def _get_websocket(self):
+        if self._websocket:
+            return self._websocket
+        raise Exception("Websocket not connected")
 
     async def _receive_messages(self):
-        async for msg in self._connection:
-            if "error" in msg:
-                logger.error(f'{self} error: {msg["error"]}')
-                await self.push_frame(TTSStoppedFrame())
-                await self.stop_all_metrics()
-                await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}'))
-            elif "audio" in msg:
+        """Receive messages from LMNT websocket."""
+        async for message in self._get_websocket():
+            if isinstance(message, bytes):
+                # Raw audio data
                 await self.stop_ttfb_metrics()
                 frame = TTSAudioRawFrame(
-                    audio=msg["audio"],
-                    sample_rate=self._settings["output_format"]["sample_rate"],
+                    audio=message,
+                    sample_rate=self._settings["sample_rate"],
                     num_channels=1,
                 )
                 await self.push_frame(frame)
             else:
-                logger.error(f"{self}: LMNT error, unknown message type: {msg}")
-
-    async def _reconnect_websocket(self, retry_state: RetryCallState):
-        logger.warning(f"{self} reconnecting (attempt: {retry_state.attempt_number})")
-        await self._disconnect_lmnt()
-        await self._connect_lmnt()
-
-    async def _receive_task_handler(self):
-        while True:
-            try:
-                async for attempt in AsyncRetrying(
-                    stop=stop_after_attempt(3),
-                    wait=wait_exponential(multiplier=1, min=4, max=10),
-                    before_sleep=self._reconnect_websocket,
-                    reraise=True,
-                ):
-                    with attempt:
-                        await self._receive_messages()
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                message = f"{self} error receiving messages: {e}"
-                logger.error(message)
-                await self.push_error(ErrorFrame(message, fatal=True))
-                break
+                try:
+                    msg = json.loads(message)
+                    if "error" in msg:
+                        logger.error(f'{self} error: {msg["error"]}')
+                        await self.push_frame(TTSStoppedFrame())
+                        await self.stop_all_metrics()
+                        await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}'))
+                        return
+                except json.JSONDecodeError:
+                    logger.error(f"Invalid JSON message: {message}")
 
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        """Generate TTS audio from text."""
         logger.debug(f"Generating TTS: [{text}]")
 
         try:
-            if not self._connection:
+            if not self._websocket:
                 await self._connect()
 
-            if not self._started:
-                await self.start_ttfb_metrics()
-                yield TTSStartedFrame()
-                self._started = True
-
             try:
-                await self._connection.append_text(text)
-                await self._connection.flush()
+                if not self._started:
+                    await self.start_ttfb_metrics()
+                    yield TTSStartedFrame()
+                    self._started = True
+
+                # Send text to LMNT
+                await self._get_websocket().send(json.dumps({"text": text}))
+                # Force synthesis
+                await self._get_websocket().send(json.dumps({"flush": True}))
                 await self.start_tts_usage_metrics(text)
             except Exception as e:
                 logger.error(f"{self} error sending message: {e}")

src/pipecat/services/moondream.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/nim.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/ollama.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/openai.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
@@ -21,6 +21,7 @@ from pipecat.frames.frames import (
     Frame,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
+    FunctionCallResultProperties,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
@@ -549,6 +550,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
             return
 
         run_llm = False
+        properties: Optional[FunctionCallResultProperties] = None
 
         aggregation = self._aggregation
         self._reset()
@@ -556,6 +558,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
         try:
             if self._function_call_result:
                 frame = self._function_call_result
+                properties = frame.properties
                 self._function_call_result = None
                 if frame.result:
                     self._context.add_message(
@@ -580,8 +583,13 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
                             "tool_call_id": frame.tool_call_id,
                         }
                     )
-                    # Only run the LLM if there are no more function calls in progress.
-                    run_llm = not bool(self._function_calls_in_progress)
+                    if properties and properties.run_llm is not None:
+                        # If the tool call result has a run_llm property, use it
+                        run_llm = properties.run_llm
+                    else:
+                        # Default behavior is to run the LLM if there are no function calls in progress
+                        run_llm = not bool(self._function_calls_in_progress)
+
             else:
                 self._context.add_message({"role": "assistant", "content": aggregation})
 
@@ -599,6 +607,10 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Emit the on_context_updated callback once the function call result is added to the context
+            if properties and properties.on_context_updated is not None:
+                await properties.on_context_updated()
+
             # Push context frame
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)

src/pipecat/services/openai_realtime_beta/context.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/openai_realtime_beta/events.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/openai_realtime_beta/frames.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/openai_realtime_beta/openai.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/openpipe.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/playht.py

@@ -1,10 +1,9 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import asyncio
 import io
 import json
 import struct
@@ -15,7 +14,6 @@ import aiohttp
 import websockets
 from loguru import logger
 from pydantic import BaseModel
-from tenacity import AsyncRetrying, RetryCallState, stop_after_attempt, wait_exponential
 
 from pipecat.frames.frames import (
     BotStoppedSpeakingFrame,
@@ -33,6 +31,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import TTSService
+from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
 try:
@@ -101,7 +100,7 @@ def language_to_playht_language(language: Language) -> str | None:
     return result
 
 
-class PlayHTTTSService(TTSService):
+class PlayHTTTSService(TTSService, WebsocketService):
     class InputParams(BaseModel):
         language: Optional[Language] = Language.EN
         speed: Optional[float] = 1.0
@@ -119,15 +118,16 @@ class PlayHTTTSService(TTSService):
         params: InputParams = InputParams(),
         **kwargs,
     ):
-        super().__init__(
+        TTSService.__init__(
+            self,
             sample_rate=sample_rate,
             **kwargs,
         )
+        WebsocketService.__init__(self)
 
         self._api_key = api_key
         self._user_id = user_id
         self._websocket_url = None
-        self._websocket = None
         self._receive_task = None
         self._request_id = None
 
@@ -165,7 +165,9 @@ class PlayHTTTSService(TTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
+        self._receive_task = self.get_event_loop().create_task(
+            self._receive_task_handler(self.push_error)
+        )
 
     async def _disconnect(self):
         await self._disconnect_websocket()
@@ -271,30 +273,6 @@ class PlayHTTTSService(TTSService):
                 except json.JSONDecodeError:
                     logger.error(f"Invalid JSON message: {message}")
 
-    async def _reconnect_websocket(self, retry_state: RetryCallState):
-        logger.warning(f"{self} reconnecting (attempt: {retry_state.attempt_number})")
-        await self._disconnect_websocket()
-        await self._connect_websocket()
-
-    async def _receive_task_handler(self):
-        while True:
-            try:
-                async for attempt in AsyncRetrying(
-                    stop=stop_after_attempt(3),
-                    wait=wait_exponential(multiplier=1, min=4, max=10),
-                    before_sleep=self._reconnect_websocket,
-                    reraise=True,
-                ):
-                    with attempt:
-                        await self._receive_messages()
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                message = f"{self} error receiving messages: {e}"
-                logger.error(message)
-                await self.push_error(ErrorFrame(message, fatal=True))
-                break
-
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 

src/pipecat/services/rime.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/riva.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/simli.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/tavus.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/together.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/websocket_service.py

@@ -0,0 +1,124 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+from abc import ABC, abstractmethod
+from typing import Awaitable, Callable, Optional
+
+import websockets
+from loguru import logger
+
+from pipecat.frames.frames import ErrorFrame
+
+
+class WebsocketService(ABC):
+    """Base class for websocket-based services with reconnection logic."""
+
+    def __init__(self):
+        """Initialize websocket attributes."""
+        self._websocket: Optional[websockets.WebSocketClientProtocol] = None
+
+    async def _verify_connection(self) -> bool:
+        """Verify websocket connection is working.
+
+        Returns:
+            bool: True if connection is verified working, False otherwise
+        """
+        try:
+            if not self._websocket:
+                return False
+            await self._websocket.ping()
+            return True
+        except Exception as e:
+            logger.error(f"{self} connection verification failed: {e}")
+            return False
+
+    async def _reconnect_websocket(self, attempt_number: int) -> bool:
+        """Reconnect the websocket.
+
+        Args:
+            attempt_number: Current retry attempt number
+
+        Returns:
+            bool: True if reconnection and verification successful, False otherwise
+        """
+        logger.warning(f"{self} reconnecting (attempt: {attempt_number})")
+        await self._disconnect_websocket()
+        await self._connect_websocket()
+        return await self._verify_connection()
+
+    def _calculate_wait_time(
+        self, attempt: int, min_wait: float = 4, max_wait: float = 10, multiplier: float = 1
+    ) -> float:
+        """Calculate exponential backoff wait time.
+
+        Args:
+            attempt: Current attempt number (1-based)
+            min_wait: Minimum wait time in seconds
+            max_wait: Maximum wait time in seconds
+            multiplier: Base multiplier for exponential calculation
+
+        Returns:
+            Wait time in seconds
+        """
+        try:
+            exp = 2 ** (attempt - 1) * multiplier
+            result = max(0, min(exp, max_wait))
+            return max(min_wait, result)
+        except (ValueError, ArithmeticError):
+            return max_wait
+
+    async def _receive_task_handler(self, report_error: Callable[[ErrorFrame], Awaitable[None]]):
+        """Handles WebSocket message receiving with automatic retry logic.
+
+        Args:
+            report_error: Callback to report errors
+        """
+        retry_count = 0
+        MAX_RETRIES = 3
+
+        while True:
+            try:
+                await self._receive_messages()
+                logger.debug(f"{self} connection established successfully")
+                retry_count = 0  # Reset counter on successful message receive
+
+            except asyncio.CancelledError:
+                break
+
+            except Exception as e:
+                retry_count += 1
+                if retry_count >= MAX_RETRIES:
+                    message = f"{self} error receiving messages: {e}"
+                    logger.error(message)
+                    await report_error(ErrorFrame(message, fatal=True))
+                    break
+
+                logger.warning(f"{self} connection error, will retry: {e}")
+
+                try:
+                    if await self._reconnect_websocket(retry_count):
+                        retry_count = 0  # Reset counter on successful reconnection
+                    wait_time = self._calculate_wait_time(retry_count)
+                    await asyncio.sleep(wait_time)
+                except Exception as reconnect_error:
+                    logger.error(f"{self} reconnection failed: {reconnect_error}")
+                    continue
+
+    @abstractmethod
+    async def _connect_websocket(self):
+        """Implement service-specific websocket connection logic."""
+        pass
+
+    @abstractmethod
+    async def _disconnect_websocket(self):
+        """Implement service-specific websocket disconnection logic."""
+        pass
+
+    @abstractmethod
+    async def _receive_messages(self):
+        """Implement service-specific message receiving logic."""
+        pass

src/pipecat/services/whisper.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/services/xtts.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/sync/base_notifier.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/sync/event_notifier.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/transcriptions/language.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/transports/base_input.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/transports/base_output.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/transports/base_transport.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/transports/local/audio.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/transports/local/tk.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #

src/pipecat/transports/network/fastapi_websocket.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2025, Daily
+# Copyright (c) 2024–2025, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #