Merge branch 'main' into filipi/pipeline_freeze

CHANGELOG.md

@@ -34,6 +34,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- `HeartbeatFrame`s are now control frames. This will make it easier to detect
+  pipeline freezes. Previously, heartbeat frames were system frames which meant
+  they were not get queued with other frames, making it difficult to detect
+  pipeline stalls.
+
 - Updated `OpenAIRealtimeBetaLLMService` to accept `language` in the
   `InputAudioTranscription` class for all models.
 
@@ -50,6 +55,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue with `ElevenLabsTTSService` where the context was not being
+  closed.
+
 - Fixed function calling in `AWSNovaSonicLLMService`.
 
 - Fixed an issue that would cause multiple `PipelineTask.on_idle_timeout`

dot-env.template

@@ -108,3 +108,6 @@ MINIMAX_GROUP_ID=...
 
 # Sarvam AI
 SARVAM_API_KEY=...
+
+# Sentry
+SENTRY_DSN=...

examples/foundational/39a-mcp-run-sse.py

@@ -9,6 +9,7 @@ import os
 
 from dotenv import load_dotenv
 from loguru import logger
+from mcp.client.session_group import SseServerParameters
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.pipeline.pipeline import Pipeline
@@ -63,7 +64,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si
 
     try:
         # https://docs.mcp.run/integrating/tutorials/mcp-run-sse-openai-agents/
-        mcp = MCPClient(server_params=os.getenv("MCP_RUN_SSE_URL"))
+        mcp = MCPClient(server_params=SseServerParameters(url=os.getenv("MCP_RUN_SSE_URL")))
     except Exception as e:
         logger.error(f"error setting up mcp")
         logger.exception("error trace:")

examples/foundational/39b-multiple-mcp.py

@@ -15,6 +15,7 @@ import aiohttp
 from dotenv import load_dotenv
 from loguru import logger
 from mcp import StdioServerParameters
+from mcp.client.session_group import SseServerParameters
 from PIL import Image
 
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
@@ -149,7 +150,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si
             # https://docs.mcp.run/integrating/tutorials/mcp-run-sse-openai-agents/
             # ie. "https://www.mcp.run/api/mcp/sse?..."
             # ensure the profile has a tool or few installed
-            mcp_run = MCPClient(server_params=os.getenv("MCP_RUN_SSE_URL"))
+            mcp_run = MCPClient(server_params=SseServerParameters(url=os.getenv("MCP_RUN_SSE_URL")))
         except Exception as e:
             logger.error(f"error setting up mcp.run")
             logger.exception("error trace:")

examples/sentry-metrics/bot.py

@@ -49,7 +49,7 @@ async def main():
 
         # Initialize Sentry
         sentry_sdk.init(
-            dsn="your-project-dsn",
+            dsn=os.getenv("SENTRY_DSN"),
             traces_sample_rate=1.0,
         )
 

src/pipecat/frames/frames.py

@@ -7,6 +7,7 @@
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import (
+    TYPE_CHECKING,
     Any,
     Awaitable,
     Callable,
@@ -26,6 +27,9 @@ from pipecat.transcriptions.language import Language
 from pipecat.utils.time import nanoseconds_to_str
 from pipecat.utils.utils import obj_count, obj_id
 
+if TYPE_CHECKING:
+    from pipecat.processors.frame_processor import FrameProcessor
+
 
 class KeypadEntry(str, Enum):
     """DTMF entries."""
@@ -485,16 +489,6 @@ class FatalErrorFrame(ErrorFrame):
     fatal: bool = field(default=True, init=False)
 
 
-@dataclass
-class HeartbeatFrame(SystemFrame):
-    """This frame is used by the pipeline task as a mechanism to know if the
-    pipeline is running properly.
-
-    """
-
-    timestamp: int
-
-
 @dataclass
 class EndTaskFrame(SystemFrame):
     """This is used to notify the pipeline task that the pipeline should be
@@ -529,25 +523,25 @@ class StopTaskFrame(SystemFrame):
 
 @dataclass
 class FrameProcessorPauseUrgentFrame(SystemFrame):
-    """This processor is used to pause frame processing for the given processor
+    """This frame is used to pause frame processing for the given processor as
-    as fast as possible. Pausing frame processing will keep frames in the
+    fast as possible. Pausing frame processing will keep frames in the internal
-    internal queue which will then be processed when frame processing is resumed
+    queue which will then be processed when frame processing is resumed with
-    with `FrameProcessorResumeFrame`.
+    `FrameProcessorResumeFrame`.
 
     """
 
-    processor: str
+    processor: "FrameProcessor"
 
 
 @dataclass
 class FrameProcessorResumeUrgentFrame(SystemFrame):
-    """This processor is used to resume frame processing for the given processor
+    """This frame is used to resume frame processing for the given processor
     if it was previously paused as fast as possible. After resuming frame
     processing all queued frames will be processed in the order received.
 
     """
 
-    processor: str
+    processor: "FrameProcessor"
 
 
 @dataclass
@@ -877,25 +871,37 @@ class StopFrame(ControlFrame):
     pass
 
 
+@dataclass
+class HeartbeatFrame(ControlFrame):
+    """This frame is used by the pipeline task as a mechanism to know if the
+    pipeline is running properly.
+
+    """
+
+    timestamp: int
+
+
 @dataclass
 class FrameProcessorPauseFrame(ControlFrame):
-    """This processor is used to pause frame processing for the given
+    """This frame is used to pause frame processing for the given
     processor. Pausing frame processing will keep frames in the internal queue
     which will then be processed when frame processing is resumed with
-    `FrameProcessorResumeFrame`."""
+    `FrameProcessorResumeFrame`.
 
-    processor: str
+    """
+
+    processor: "FrameProcessor"
 
 
 @dataclass
 class FrameProcessorResumeFrame(ControlFrame):
-    """This processor is used to resume frame processing for the given processor
+    """This frame is used to resume frame processing for the given processor if
-    if it was previously paused. After resuming frame processing all queued
+    it was previously paused. After resuming frame processing all queued frames
-    frames will be processed in the order received.
+    will be processed in the order received.
 
     """
 
-    processor: str
+    processor: "FrameProcessor"
 
 
 @dataclass

src/pipecat/pipeline/task.py

@@ -43,7 +43,7 @@ from pipecat.utils.tracing.setup import is_tracing_available
 from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver
 
 HEARTBEAT_SECONDS = 1.0
-HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5
+HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 10
 
 
 class PipelineParams(BaseModel):

src/pipecat/processors/frame_processor.py

@@ -296,11 +296,11 @@ class FrameProcessor(BaseObject):
         await self.__cancel_push_task()
 
     async def __pause(self, frame: FrameProcessorPauseFrame | FrameProcessorPauseUrgentFrame):
-        if frame.name == self.name:
+        if frame.processor.name == self.name:
             await self.pause_processing_frames()
 
     async def __resume(self, frame: FrameProcessorResumeFrame | FrameProcessorResumeUrgentFrame):
-        if frame.name == self.name:
+        if frame.processor.name == self.name:
             await self.resume_processing_frames()
 
     #

src/pipecat/services/elevenlabs/tts.py

@@ -284,7 +284,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
         logger.trace(f"{self}: flushing audio")
         msg = {"context_id": self._context_id, "flush": True}
         await self._websocket.send(json.dumps(msg))
-        self._context_id = None
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
         await super().push_frame(frame, direction)
@@ -380,6 +379,12 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
         if self._context_id and self._websocket:
             logger.trace(f"Closing context {self._context_id} due to interruption")
             try:
+                # ElevenLabs requires that Pipecat manages the contexts and closes them
+                # when they're not longer in use. Since a StartInterruptionFrame is pushed
+                # every time the user speaks, we'll use this as a trigger to close the context
+                # and reset the state.
+                # Note: We do not need to call remove_audio_context here, as the context is
+                # automatically reset when super ()._handle_interruption is called.
                 await self._websocket.send(
                     json.dumps({"context_id": self._context_id, "close_context": True})
                 )
@@ -391,10 +396,20 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
     async def _receive_messages(self):
         async for message in self._get_websocket():
             msg = json.loads(message)
-            # Check if this message belongs to the current context
+
             received_ctx_id = msg.get("contextId")
+
+            # Handle final messages first, regardless of context availability
+            # At the moment, this message is received AFTER the close_context message is
+            # sent, so it doesn't serve any functional purpose. For now, we'll just log it.
+            if msg.get("isFinal") is True:
+                logger.trace(f"Received final message for context {received_ctx_id}")
+                continue
+
+            # Check if this message belongs to the current context.
+            # This should never happen, so warn about it.
             if not self.audio_context_available(received_ctx_id):
-                logger.trace(f"Ignoring message from unavailable context: {received_ctx_id}")
+                logger.warning(f"Ignoring message from unavailable context: {received_ctx_id}")
                 continue
 
             if msg.get("audio"):
@@ -408,21 +423,26 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 word_times = calculate_word_times(msg["alignment"], self._cumulative_time)
                 await self.add_word_timestamps(word_times)
                 self._cumulative_time = word_times[-1][1]
-            if msg.get("isFinal"):
-                logger.trace(f"Received final message for context {received_ctx_id}")
-                await self.remove_audio_context(received_ctx_id)
-                # Reset context tracking if this was our active context
-                if self._context_id == received_ctx_id:
-                    self._context_id = None
-                    self._started = False
 
     async def _keepalive_task_handler(self):
         while True:
             await asyncio.sleep(10)
             try:
-                # Send an empty message to keep the connection alive
                 if self._websocket and self._websocket.open:
-                    await self._websocket.send(json.dumps({}))
+                    if self._context_id:
+                        # Send keepalive with context ID to keep the connection alive
+                        keepalive_message = {
+                            "text": "",
+                            "context_id": self._context_id,
+                        }
+                        logger.trace(f"Sending keepalive for context {self._context_id}")
+                    else:
+                        # It's possible to have a user interruption which clears the context
+                        # without generating a new TTS response. In this case, we'll just send
+                        # an empty message to keep the connection alive.
+                        keepalive_message = {"text": ""}
+                        logger.trace("Sending keepalive without context")
+                    await self._websocket.send(json.dumps(keepalive_message))
             except websockets.ConnectionClosed as e:
                 logger.warning(f"{self} keepalive error: {e}")
                 break
@@ -441,14 +461,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 await self._connect()
 
             try:
-                # Close previous context if there was one
-                if self._context_id and not self._started:
-                    await self._websocket.send(
-                        json.dumps({"context_id": self._context_id, "close_context": True})
-                    )
-                    await self.remove_audio_context(self._context_id)
-                    self._context_id = None
-
                 if not self._started:
                     await self.start_ttfb_metrics()
                     yield TTSStartedFrame()
@@ -473,9 +485,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 logger.error(f"{self} error sending message: {e}")
                 yield TTSStoppedFrame()
                 self._started = False
-                if self._context_id:
-                    await self.remove_audio_context(self._context_id)
-                    self._context_id = None
                 return
             yield None
         except Exception as e:

src/pipecat/services/mcp_service.py

@@ -8,8 +8,8 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.utils.base_object import BaseObject
 
 try:
-    from mcp import ClientSession, StdioServerParameters, types
+    from mcp import ClientSession, StdioServerParameters
-    from mcp.client.session import ClientSession
+    from mcp.client.session_group import SseServerParameters
     from mcp.client.sse import sse_client
     from mcp.client.stdio import stdio_client
 except ModuleNotFoundError as e:
@@ -21,7 +21,7 @@ except ModuleNotFoundError as e:
 class MCPClient(BaseObject):
     def __init__(
         self,
-        server_params: Union[StdioServerParameters, str],
+        server_params: Union[StdioServerParameters, SseServerParameters],
         **kwargs,
     ):
         super().__init__(**kwargs)
@@ -30,12 +30,12 @@ class MCPClient(BaseObject):
         if isinstance(server_params, StdioServerParameters):
             self._client = stdio_client
             self._register_tools = self._stdio_register_tools
-        elif isinstance(server_params, str):
+        elif isinstance(server_params, SseServerParameters):
             self._client = sse_client
             self._register_tools = self._sse_register_tools
         else:
             raise TypeError(
-                f"{self} invalid argument type: `server_params` must be either StdioServerParameters or an SSE server url string."
+                f"{self} invalid argument type: `server_params` must be either StdioServerParameters or SseServerParameters."
             )
 
     async def register_tools(self, llm) -> ToolsSchema:
@@ -90,7 +90,12 @@ class MCPClient(BaseObject):
             logger.debug(f"Executing tool '{function_name}' with call ID: {tool_call_id}")
             logger.trace(f"Tool arguments: {json.dumps(arguments, indent=2)}")
             try:
-                async with self._client(self._server_params) as (read, write):
+                async with self._client(
+                    url=self._server_params.url,
+                    headers=self._server_params.headers,
+                    timeout=self._server_params.timeout,
+                    sse_read_timeout=self._server_params.sse_read_timeout,
+                ) as (read, write):
                     async with self._session(read, write) as session:
                         await session.initialize()
                         await self._call_tool(session, function_name, arguments, result_callback)
@@ -100,10 +105,14 @@ class MCPClient(BaseObject):
                 logger.exception("Full exception details:")
                 await result_callback(error_msg)
 
-        logger.debug("Starting registration of mcp.run tools")
+        logger.debug(f"SSE server parameters: {self._server_params}")
-        tool_schemas: List[FunctionSchema] = []
 
-        async with self._client(self._server_params) as (read, write):
+        async with self._client(
+            url=self._server_params.url,
+            headers=self._server_params.headers,
+            timeout=self._server_params.timeout,
+            sse_read_timeout=self._server_params.sse_read_timeout,
+        ) as (read, write):
             async with self._session(read, write) as session:
                 await session.initialize()
                 tools_schema = await self._list_tools(session, mcp_tool_wrapper, llm)