From 3da711ba8b4aaa1d1ef4f4f6aa45406b18719b5c Mon Sep 17 00:00:00 2001 From: ezun-kim <48287335+ezun-kim@users.noreply.github.com> Date: Sat, 24 May 2025 22:35:57 +0900 Subject: [PATCH 1/8] Fix SSE server connection handling for MCP client ### Summary This PR improves the MCP (Model Context Protocol) client's SSE (Server-Sent Events) server connection handling by replacing the generic string parameter with a proper `SseServerParameters` class. ### Changes - **Breaking Change**: Changed `server_params` type from `Union[StdioServerParameters, str]` to `Union[StdioServerParameters, SseServerParameters]` - Added import for `SseServerParameters` from `mcp.client.session_group` - Updated SSE client connection to use structured parameters instead of a simple URL string - Fixed error message to correctly reflect the expected parameter types - Improved logging by changing info-level log to debug-level for consistency ### Details #### Before The SSE client connection only accepted a URL string: ```python async with self._client(self._server_params) as (read, write): ``` #### After Now properly unpacks SSE server parameters: ```python 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): ``` ### Benefits - **Type Safety**: Stronger type checking with dedicated `SseServerParameters` class - **Extended Configuration**: Support for custom headers (authentication), timeouts, and SSE-specific settings - **Better Error Messages**: Clear type error messages when incorrect parameters are provided - **Improved Debugging**: Debug logging of SSE server parameters for troubleshooting ### Migration Guide Users need to update their SSE server initialization: ```python # Before client = MCPClient("https://example.com/sse") # After from mcp.client.session_group import SseServerParameters client = MCPClient(SseServerParameters( url="https://example.com/sse", headers={"Authorization": "Bearer token"}, timeout=30, sse_read_timeout=60 )) ``` ### Testing - [ ] Tested with StdioServerParameters (unchanged behavior) - [ ] Tested with SseServerParameters with various configurations - [ ] Verified error handling for invalid parameter types --- This is a necessary change to support production-ready SSE connections with proper authentication and timeout handling. --- src/pipecat/services/mcp_service.py | 33 ++++++++++++++++++----------- 1 file changed, 21 insertions(+), 12 deletions(-) diff --git a/src/pipecat/services/mcp_service.py b/src/pipecat/services/mcp_service.py index ae54b84ff..2e519dd24 100644 --- a/src/pipecat/services/mcp_service.py +++ b/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.client.session import ClientSession + from mcp import ClientSession, StdioServerParameters + 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) @@ -98,12 +103,16 @@ class MCPClient(BaseObject): error_msg = f"Error calling mcp tool {function_name}: {str(e)}" logger.error(error_msg) logger.exception("Full exception details:") - await result_callback(error_msg) - - logger.debug("Starting registration of mcp.run tools") - tool_schemas: List[FunctionSchema] = [] - - async with self._client(self._server_params) as (read, write): + await result_callback(error_msg)\ + + logger.debug(f"SSE server parameters: {self._server_params}") + + 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) From 5cc9b7e0d164776ebbb0c73e9af75b68f627fef1 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 20 Jun 2025 15:46:18 -0400 Subject: [PATCH 2/8] Fix: Correctly close the context for ElevenLabsTTSService --- CHANGELOG.md | 3 ++ src/pipecat/services/elevenlabs/tts.py | 39 ++++++++++++-------------- 2 files changed, 21 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 466b3f28b..aecab04f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,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` diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index e0301360e..f48ba5552 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/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,13 +423,6 @@ 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: @@ -441,14 +449,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 +473,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: From 6b24f89fa7e57afaa3c5b3fd2fe25723b1f16328 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Thu, 19 Jun 2025 17:05:52 -0700 Subject: [PATCH 3/8] small fix for processor pause/resume frames --- src/pipecat/frames/frames.py | 34 +++++++++++++---------- src/pipecat/processors/frame_processor.py | 4 +-- 2 files changed, 22 insertions(+), 16 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index ec368789d..c303d99ab 100644 --- a/src/pipecat/frames/frames.py +++ b/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.""" @@ -529,25 +533,25 @@ class StopTaskFrame(SystemFrame): @dataclass class FrameProcessorPauseUrgentFrame(SystemFrame): - """This processor is used to pause frame processing for the given processor - as fast as possible. Pausing frame processing will keep frames in the - internal queue which will then be processed when frame processing is resumed - with `FrameProcessorResumeFrame`. + """This frame is used to pause frame processing for the given processor as + fast as possible. Pausing frame processing will keep frames in the internal + queue which will then be processed when frame processing is resumed with + `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 @@ -879,23 +883,25 @@ class StopFrame(ControlFrame): @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 - if it was previously paused. After resuming frame processing all queued - frames will be processed in the order received. + """This frame is used to resume frame processing for the given processor if + it was previously paused. After resuming frame processing all queued frames + will be processed in the order received. """ - processor: str + processor: "FrameProcessor" @dataclass diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 1d2f066ed..680465e2d 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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() # From 2eb244c80a28470200f3f004b6c8ec0a93905649 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 24 Jun 2025 10:50:44 -0400 Subject: [PATCH 4/8] Send context_id when available in ElevenLabsTTSService keepalive message --- src/pipecat/services/elevenlabs/tts.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index f48ba5552..ccd9b5b3f 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -428,9 +428,21 @@ class ElevenLabsTTSService(AudioContextWordTTSService): 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 From 365260ec44eb0f8ff7fc470556fe748a29f109a7 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 24 Jun 2025 11:57:14 -0300 Subject: [PATCH 5/8] Creating an environment variable for sentry dsn. --- dot-env.template | 5 ++++- examples/sentry-metrics/bot.py | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/dot-env.template b/dot-env.template index 20d73b3ad..4cf716137 100644 --- a/dot-env.template +++ b/dot-env.template @@ -107,4 +107,7 @@ MINIMAX_API_KEY=... MINIMAX_GROUP_ID=... # Sarvam AI -SARVAM_API_KEY=... \ No newline at end of file +SARVAM_API_KEY=... + +# Sentry +SENTRY_DSN=... \ No newline at end of file diff --git a/examples/sentry-metrics/bot.py b/examples/sentry-metrics/bot.py index 8ff412bb7..44f9a0daa 100644 --- a/examples/sentry-metrics/bot.py +++ b/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, ) From dd1ff237a83580ef39e077abcb50252b5ebdd766 Mon Sep 17 00:00:00 2001 From: vipyne Date: Tue, 24 Jun 2025 12:35:55 -0500 Subject: [PATCH 6/8] lint mcp_service --- src/pipecat/services/mcp_service.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/pipecat/services/mcp_service.py b/src/pipecat/services/mcp_service.py index 2e519dd24..a644d8f1b 100644 --- a/src/pipecat/services/mcp_service.py +++ b/src/pipecat/services/mcp_service.py @@ -94,7 +94,7 @@ class MCPClient(BaseObject): url=self._server_params.url, headers=self._server_params.headers, timeout=self._server_params.timeout, - sse_read_timeout=self._server_params.sse_read_timeout + sse_read_timeout=self._server_params.sse_read_timeout, ) as (read, write): async with self._session(read, write) as session: await session.initialize() @@ -103,15 +103,15 @@ class MCPClient(BaseObject): error_msg = f"Error calling mcp tool {function_name}: {str(e)}" logger.error(error_msg) logger.exception("Full exception details:") - await result_callback(error_msg)\ - + await result_callback(error_msg) + logger.debug(f"SSE server parameters: {self._server_params}") - + 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 + sse_read_timeout=self._server_params.sse_read_timeout, ) as (read, write): async with self._session(read, write) as session: await session.initialize() From 20047c369e2f07453f2b3d8e0c785dcc3f66e583 Mon Sep 17 00:00:00 2001 From: vipyne Date: Tue, 24 Jun 2025 12:37:18 -0500 Subject: [PATCH 7/8] mcp: update examples to use SseServerParameter --- examples/foundational/39a-mcp-run-sse.py | 3 ++- examples/foundational/39b-multiple-mcp.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/examples/foundational/39a-mcp-run-sse.py b/examples/foundational/39a-mcp-run-sse.py index 8b43b3022..fbbcbcc6a 100644 --- a/examples/foundational/39a-mcp-run-sse.py +++ b/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:") diff --git a/examples/foundational/39b-multiple-mcp.py b/examples/foundational/39b-multiple-mcp.py index 3ba90d580..7d1396834 100644 --- a/examples/foundational/39b-multiple-mcp.py +++ b/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:") From a4e6ea5a3fea03a3d4aa3313d745ab34756a8585 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 24 Jun 2025 11:27:39 -0700 Subject: [PATCH 8/8] HeartbeatFrames are now control frames --- CHANGELOG.md | 5 +++++ src/pipecat/frames/frames.py | 20 ++++++++++---------- src/pipecat/pipeline/task.py | 2 +- 3 files changed, 16 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e4ef1e019..62c8f4d84 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index ec368789d..82626654a 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -485,16 +485,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 @@ -877,6 +867,16 @@ 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 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 3f13e7eb2..7a2c4fd36 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -41,7 +41,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):