Merge pull request #742 from Vaibhav159/vl_feature_websocket_fastapi_timeout

adding session_timeout param

CHANGELOG.md

@@ -15,10 +15,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `LiveKitTransportLayer`.
 - Added `enable_prejoin_ui`, `max_participants` and `start_video_off` params
   to `DailyRoomProperties`.
+- Added `session_timeout` to `FastAPIWebsocketTransport` and `WebsocketServerTransport`
+  for configuring session timeouts (in seconds). Triggers `on_session_timeout` for custom timeout handling. 
+  See [examples/websocket-server/bot.py](https://github.com/pipecat-ai/pipecat/blob/main/examples/websocket-server/bot.py).
 
 ### Changed
 
 - api_key, aws_access_key_id and region are no longer required parameters for the PollyTTSService (AWSTTSService)
+- Added `session_timeout` example in `examples/websocket-server/bot.py` to handle session timeout event.
 
 ### Fixed
 

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 BotInterruptionFrame, EndFrame, LLMMessagesFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -31,6 +31,56 @@ 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 +90,7 @@ async def main():
             vad_enabled=True,
             vad_analyzer=SileroVADAnalyzer(),
             vad_audio_passthrough=True,
+            session_timeout=60 * 3,  # 3 minutes
         )
     )
 
@@ -83,6 +134,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

@@ -42,11 +42,13 @@ except ModuleNotFoundError as e:
 class FastAPIWebsocketParams(TransportParams):
     add_wav_header: bool = False
     serializer: FrameSerializer
+    session_timeout: int | None = None
 
 
 class FastAPIWebsocketCallbacks(BaseModel):
     on_client_connected: Callable[[WebSocket], Awaitable[None]]
     on_client_disconnected: Callable[[WebSocket], Awaitable[None]]
+    on_session_timeout: Callable[[WebSocket], Awaitable[None]]
 
 
 class FastAPIWebsocketInputTransport(BaseInputTransport):
@@ -65,6 +67,10 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+        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())
 
@@ -88,6 +94,14 @@ 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."""
+        try:
+            await asyncio.sleep(self._params.session_timeout)
+            await self._callbacks.on_session_timeout(self._websocket)
+        except asyncio.CancelledError:
+            logger.info(f"Monitoring task cancelled for: {self._websocket}")
+
 
 class FastAPIWebsocketOutputTransport(BaseOutputTransport):
     def __init__(self, websocket: WebSocket, params: FastAPIWebsocketParams, **kwargs):
@@ -176,6 +190,7 @@ class FastAPIWebsocketTransport(BaseTransport):
         self._callbacks = FastAPIWebsocketCallbacks(
             on_client_connected=self._on_client_connected,
             on_client_disconnected=self._on_client_disconnected,
+            on_session_timeout=self._on_session_timeout,
         )
 
         self._input = FastAPIWebsocketInputTransport(
@@ -189,6 +204,7 @@ class FastAPIWebsocketTransport(BaseTransport):
         # these handlers.
         self._register_event_handler("on_client_connected")
         self._register_event_handler("on_client_disconnected")
+        self._register_event_handler("on_session_timeout")
 
     def input(self) -> FastAPIWebsocketInputTransport:
         return self._input
@@ -201,3 +217,6 @@ class FastAPIWebsocketTransport(BaseTransport):
 
     async def _on_client_disconnected(self, websocket):
         await self._call_event_handler("on_client_disconnected", websocket)
+
+    async def _on_session_timeout(self, websocket):
+        await self._call_event_handler("on_session_timeout", websocket)

src/pipecat/transports/network/websocket_server.py

@@ -40,11 +40,13 @@ except ModuleNotFoundError as e:
 class WebsocketServerParams(TransportParams):
     add_wav_header: bool = False
     serializer: FrameSerializer = ProtobufFrameSerializer()
+    session_timeout: int | None = None
 
 
 class WebsocketServerCallbacks(BaseModel):
     on_client_connected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]]
     on_client_disconnected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]]
+    on_session_timeout: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]]
 
 
 class WebsocketServerInputTransport(BaseInputTransport):
@@ -97,6 +99,10 @@ class WebsocketServerInputTransport(BaseInputTransport):
         # Notify
         await self._callbacks.on_client_connected(websocket)
 
+        # Create a task to monitor the websocket connection
+        if self._params.session_timeout:
+            self.get_event_loop().create_task(self._monitor_websocket(websocket))
+
         # Handle incoming messages
         async for message in websocket:
             frame = self._params.serializer.deserialize(message)
@@ -117,6 +123,15 @@ 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."""
+        try:
+            await asyncio.sleep(self._params.session_timeout)
+            if not websocket.closed:
+                await self._callbacks.on_session_timeout(websocket)
+        except asyncio.CancelledError:
+            logger.info(f"Monitoring task cancelled for: {websocket.remote_address}")
+
 
 class WebsocketServerOutputTransport(BaseOutputTransport):
     def __init__(self, params: WebsocketServerParams, **kwargs):
@@ -211,6 +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,
         )
         self._input: WebsocketServerInputTransport | None = None
         self._output: WebsocketServerOutputTransport | None = None
@@ -220,6 +236,7 @@ class WebsocketServerTransport(BaseTransport):
         # these handlers.
         self._register_event_handler("on_client_connected")
         self._register_event_handler("on_client_disconnected")
+        self._register_event_handler("on_session_timeout")
 
     def input(self) -> WebsocketServerInputTransport:
         if not self._input:
@@ -246,3 +263,6 @@ class WebsocketServerTransport(BaseTransport):
             await self._call_event_handler("on_client_disconnected", websocket)
         else:
             logger.error("A WebsocketServerTransport output is missing in the pipeline")
+
+    async def _on_session_timeout(self, websocket):
+        await self._call_event_handler("on_session_timeout", websocket)