adding session timeout example in websocket-server example

examples/websocket-server/bot.py

@@ -12,7 +12,7 @@ from dotenv import load_dotenv
 from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import LLMMessagesFrame
+from pipecat.frames.frames import LLMMessagesFrame, BotInterruptionFrame, EndFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -31,6 +31,57 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
+class SessionTimeoutHandler:
+    """
+    Handles actions to be performed when a session times out.
+    Inputs:
+    - task: Pipeline task (used to queue frames).
+    - tts: TTS service (used to generate speech output).
+    """
+
+    def __init__(self, task, tts):
+        self.task = task
+        self.tts = tts
+        self.background_tasks = set()
+
+    async def handle_timeout(self, client_address):
+        """
+        Handles the timeout event for a session.
+        """
+        try:
+            logger.info(f"Connection timeout for {client_address}")
+
+            # Queue a BotInterruptionFrame to notify the user
+            await self.task.queue_frames([BotInterruptionFrame()])
+
+            # Send the TTS message to inform the user about the timeout
+            await self.tts.say(
+                "I'm sorry, we are ending the call now. Please feel free to reach out again if you need assistance."
+            )
+
+            # Start the process to gracefully end the call in the background
+            end_call_task = asyncio.create_task(self._end_call())
+            self.background_tasks.add(end_call_task)
+            end_call_task.add_done_callback(self.background_tasks.discard)
+        except Exception as e:
+            logger.error(f"Error during session timeout handling: {e}")
+
+    async def _end_call(self):
+        """
+        Completes the session termination process after the TTS message.
+        """
+        try:
+            # Wait for a duration to ensure TTS has completed
+            await asyncio.sleep(15)
+
+            # Queue both BotInterruptionFrame and EndFrame to conclude the session
+            await self.task.queue_frames([BotInterruptionFrame(), EndFrame()])
+
+            logger.info("TTS completed and EndFrame pushed successfully.")
+        except Exception as e:
+            logger.error(f"Error during call termination: {e}")
+
+
 async def main():
     transport = WebsocketServerTransport(
         params=WebsocketServerParams(
@@ -40,6 +91,7 @@ async def main():
             vad_enabled=True,
             vad_analyzer=SileroVADAnalyzer(),
             vad_audio_passthrough=True,
+            session_timeout=60 * 3,  # 3 minutes
         )
     )
 
@@ -83,6 +135,14 @@ async def main():
         messages.append({"role": "system", "content": "Please introduce yourself to the user."})
         await task.queue_frames([LLMMessagesFrame(messages)])
 
+    @transport.event_handler("on_session_timeout")
+    async def on_session_timeout(transport, client):
+        logger.info(f"Entering in timeout for {client.remote_address}")
+
+        timeout_handler = SessionTimeoutHandler(task, tts)
+
+        await timeout_handler.handle_timeout(client)
+
     runner = PipelineRunner()
 
     await runner.run(task)

src/pipecat/transports/network/fastapi_websocket.py

@@ -67,7 +67,10 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
-        self._monitor_websocket_task = self.get_event_loop().create_task(self._monitor_websocket())
+        if self._params.session_timeout:
+            self._monitor_websocket_task = self.get_event_loop().create_task(
+                self._monitor_websocket()
+            )
         await self._callbacks.on_client_connected(self._websocket)
         self._receive_task = self.get_event_loop().create_task(self._receive_messages())
 
@@ -92,9 +95,7 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
         await self._callbacks.on_client_disconnected(self._websocket)
 
     async def _monitor_websocket(self):
-        """
+        """Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event."""
-        Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event.
-        """
         try:
             await asyncio.sleep(self._params.session_timeout)
             await self._callbacks.on_session_timeout(self._websocket)

src/pipecat/transports/network/websocket_server.py

@@ -124,9 +124,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
         logger.info(f"Client {websocket.remote_address} disconnected")
 
     async def _monitor_websocket(self, websocket: websockets.WebSocketServerProtocol):
-        """
+        """Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event."""
-        Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event.
-        """
         try:
             await asyncio.sleep(self._params.session_timeout)
             if not websocket.closed:
@@ -228,7 +226,7 @@ class WebsocketServerTransport(BaseTransport):
         self._callbacks = WebsocketServerCallbacks(
             on_client_connected=self._on_client_connected,
             on_client_disconnected=self._on_client_disconnected,
-            on_session_timeout=self._on_session_timeout
+            on_session_timeout=self._on_session_timeout,
         )
         self._input: WebsocketServerInputTransport | None = None
         self._output: WebsocketServerOutputTransport | None = None
@@ -268,4 +266,3 @@ class WebsocketServerTransport(BaseTransport):
 
     async def _on_session_timeout(self, websocket):
         await self._call_event_handler("on_session_timeout", websocket)
-