From 41f817bf04d0c96217f2fce6adb2f0962f9cc4e5 Mon Sep 17 00:00:00 2001 From: vipyne Date: Wed, 15 Oct 2025 13:02:44 -0500 Subject: [PATCH 1/5] only import whatsapp deps if using whatsapp runner --- src/pipecat/runner/run.py | 21 ++++++++------------- 1 file changed, 8 insertions(+), 13 deletions(-) diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index a7af52e18..c5c9dc9db 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -289,19 +289,6 @@ def _add_lifespan_to_app(app: FastAPI, new_lifespan): def _setup_whatsapp_routes(app: FastAPI): """Set up WebRTC-specific routes.""" - try: - from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI - - from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection - from pipecat.transports.smallwebrtc.request_handler import ( - SmallWebRTCRequest, - SmallWebRTCRequestHandler, - ) - from pipecat.transports.whatsapp.api import WhatsAppWebhookRequest - from pipecat.transports.whatsapp.client import WhatsAppClient - except ImportError as e: - logger.error(f"WebRTC transport dependencies not installed: {e}") - return WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID") @@ -320,6 +307,14 @@ def _setup_whatsapp_routes(app: FastAPI): ) return + # only import WhatsApp dependencies if appropriate env vars are present + try: + from pipecat.transports.whatsapp.api import WhatsAppWebhookRequest + from pipecat.transports.whatsapp.client import WhatsAppClient + except ImportError as e: + logger.error(f"WhatsApp transport dependencies not installed: {e}") + return + # Global WhatsApp client instance whatsapp_client: Optional[WhatsAppClient] = None From 0fd5d2610484909b6162bf4ee528ca221bf20ce6 Mon Sep 17 00:00:00 2001 From: vipyne Date: Wed, 15 Oct 2025 13:16:16 -0500 Subject: [PATCH 2/5] add WHATSAPP_APP_SECRET to required whatsapp env vars --- src/pipecat/runner/run.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index c5c9dc9db..4663b115d 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -289,7 +289,6 @@ def _add_lifespan_to_app(app: FastAPI, new_lifespan): def _setup_whatsapp_routes(app: FastAPI): """Set up WebRTC-specific routes.""" - WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID") WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFICATION_TOKEN") @@ -300,10 +299,17 @@ def _setup_whatsapp_routes(app: FastAPI): WHATSAPP_TOKEN, WHATSAPP_PHONE_NUMBER_ID, WHATSAPP_WEBHOOK_VERIFICATION_TOKEN, + WHATSAPP_APP_SECRET, ] ): logger.trace( - "Missing required environment variables for WhatsApp transport. Keeping it disabled." + """Missing required environment variables for WhatsApp transport: + WHATSAPP_TOKEN + WHATSAPP_PHONE_NUMBER_ID + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN + WHATSAPP_APP_SECRET +Keeping it disabled. + """ ) return From 6381335346ba9622f2eced03aca07891af0c4223 Mon Sep 17 00:00:00 2001 From: vipyne Date: Thu, 16 Oct 2025 10:37:46 -0500 Subject: [PATCH 3/5] Add --whatsapp flag to runner --- CHANGELOG.md | 2 ++ src/pipecat/runner/run.py | 45 ++++++++++++++++++++++++++------------- 2 files changed, 32 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3cd715776..2b078fe80 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `--whatsapp` flag to runner to better surface WhatsApp transport logs. + - Added `on_connected` and `on_disconnected` events to TTS and STT websocket-based services. diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 4663b115d..a3e2984e8 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -166,6 +166,7 @@ def _create_server_app( host: str = "localhost", proxy: str, esp32_mode: bool = False, + whatsapp_enabled: bool = False, folder: Optional[str] = None, ): """Create FastAPI app with transport-specific routes.""" @@ -182,7 +183,8 @@ def _create_server_app( # Set up transport-specific routes if transport_type == "webrtc": _setup_webrtc_routes(app, esp32_mode=esp32_mode, host=host, folder=folder) - _setup_whatsapp_routes(app) + if whatsapp_enabled: + _setup_whatsapp_routes(app) elif transport_type == "daily": _setup_daily_routes(app) elif transport_type in TELEPHONY_TRANSPORTS: @@ -289,32 +291,37 @@ def _add_lifespan_to_app(app: FastAPI, new_lifespan): def _setup_whatsapp_routes(app: FastAPI): """Set up WebRTC-specific routes.""" - WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") - WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID") - WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFICATION_TOKEN") WHATSAPP_APP_SECRET = os.getenv("WHATSAPP_APP_SECRET") + WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID") + WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFICATION_TOKEN") if not all( [ - WHATSAPP_TOKEN, - WHATSAPP_PHONE_NUMBER_ID, - WHATSAPP_WEBHOOK_VERIFICATION_TOKEN, WHATSAPP_APP_SECRET, + WHATSAPP_PHONE_NUMBER_ID, + WHATSAPP_TOKEN, + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN, ] ): - logger.trace( + logger.error( """Missing required environment variables for WhatsApp transport: - WHATSAPP_TOKEN - WHATSAPP_PHONE_NUMBER_ID - WHATSAPP_WEBHOOK_VERIFICATION_TOKEN WHATSAPP_APP_SECRET -Keeping it disabled. + WHATSAPP_PHONE_NUMBER_ID + WHATSAPP_TOKEN + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN """ ) return - # only import WhatsApp dependencies if appropriate env vars are present try: + from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI + + from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection + from pipecat.transports.smallwebrtc.request_handler import ( + SmallWebRTCRequest, + SmallWebRTCRequestHandler, + ) from pipecat.transports.whatsapp.api import WhatsAppWebhookRequest from pipecat.transports.whatsapp.client import WhatsAppClient except ImportError as e: @@ -690,6 +697,12 @@ def main(): parser.add_argument( "--verbose", "-v", action="count", default=0, help="Increase logging verbosity" ) + parser.add_argument( + "--whatsapp", + action="store_true", + default=False, + help="Ensure requried WhatsApp environment variables are present", + ) args = parser.parse_args() @@ -732,10 +745,11 @@ def main(): print() if args.esp32: print(f"🚀 Bot ready! (ESP32 mode)") - print(f" → Open http://{args.host}:{args.port}/client in your browser") + elif args.whatsapp: + print(f"🚀 Bot ready! (WhatsApp)") else: print(f"🚀 Bot ready!") - print(f" → Open http://{args.host}:{args.port}/client in your browser") + print(f" → Open http://{args.host}:{args.port}/client in your browser") print() elif args.transport == "daily": print() @@ -753,6 +767,7 @@ def main(): host=args.host, proxy=args.proxy, esp32_mode=args.esp32, + whatsapp_enabled=args.whatsapp, folder=args.folder, ) From 1b9e96c0168facfef2f704af0d329553135eda49 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 Oct 2025 22:37:16 -0700 Subject: [PATCH 4/5] PipelineTask: fix task cancellation issues --- CHANGELOG.md | 8 ++- src/pipecat/pipeline/runner.py | 8 ++- src/pipecat/pipeline/task.py | 99 +++++++++++++++------------------- tests/test_pipeline.py | 31 +++++------ 4 files changed, 67 insertions(+), 79 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3cd715776..c9f645589 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,12 +25,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `CartesiaSTTService` now inherits from `WebsocketSTTService`. -- Package upgrades: +# Package upgrades: - `openai` upgraded to support up to 2.x.x. - `openpipe` upgraded to support up to 5.x.x. ### Fixed +- Fixed multiple pipeline task cancellation issues. `asyncio.CancelledError` is + now handled properly in `PipelineTask` making it possible to cancel an asyncio + task that it's executing a `PipelineRunner` cleanly. Also, + `PipelineTask.cancel()` does not block anymore waiting for the `CancelFrame` + to reach the end of the pipeline (going back to the behavior in < 0.0.83). + - Fixed an issue in `ElevenLabsTTSService` and `ElevenLabsHttpTTSService` where the Flash models would split words, resulting in a space being inserted between words. diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index 9d82fcd88..c7e246681 100644 --- a/src/pipecat/pipeline/runner.py +++ b/src/pipecat/pipeline/runner.py @@ -70,11 +70,15 @@ class PipelineRunner(BaseObject): """ logger.debug(f"Runner {self} started running {task}") self._tasks[task.name] = task - params = PipelineTaskParams(loop=self._loop) + + # PipelineTask handles asyncio.CancelledError to shutdown the pipeline + # properly and re-raises it in case there's more cleanup to do. try: + params = PipelineTaskParams(loop=self._loop) await task.run(params) except asyncio.CancelledError: - await self._cancel() + pass + del self._tasks[task.name] # Cleanup base object. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index f6d17262e..a511db12a 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -269,6 +269,9 @@ class PipelineTask(BasePipelineTask): # StopFrame) has been received at the end of the pipeline. self._pipeline_end_event = asyncio.Event() + # This event is set when the pipeline truly finishes. + self._pipeline_finished_event = asyncio.Event() + # This is the final pipeline. It is composed of a source processor, # followed by the user pipeline, and ending with a sink processor. The # source allows us to receive and react to upstream frames, and the sink @@ -401,11 +404,7 @@ class PipelineTask(BasePipelineTask): await self.queue_frame(EndFrame()) async def cancel(self): - """Immediately stop the running pipeline. - - Cancels all running tasks and stops frame processing without - waiting for completion. - """ + """Request the running pipeline to cancel.""" if not self._finished: await self._cancel() @@ -417,51 +416,38 @@ class PipelineTask(BasePipelineTask): """ if self.has_finished(): return - cleanup_pipeline = True + + # Setup processors. + await self._setup(params) + + # Create all main tasks and wait for the main push task. This is the + # task that pushes frames to the very beginning of our pipeline (i.e. to + # our controlled source processor). + await self._create_tasks() + try: - # Setup processors. - await self._setup(params) - - # Create all main tasks and wait of the main push task. This is the - # task that pushes frames to the very beginning of our pipeline (our - # controlled source processor). - push_task = await self._create_tasks() - await push_task - - # We have already cleaned up the pipeline inside the task. - cleanup_pipeline = False - - # Pipeline has finished nicely. - self._finished = True + # Wait for pipeline to finish. + await self._wait_for_pipeline_finished() except asyncio.CancelledError: - # Raise exception back to the pipeline runner so it can cancel this - # task properly. + logger.debug(f"Pipeline task {self} got cancelled from outside...") + # We have been cancelled from outside, let's just cancel everything. + await self._cancel() + # Wait again for pipeline to finish. This time we have really + # cancelled, so it should really finish. + await self._wait_for_pipeline_finished() + # Re-raise in case there's more cleanup to do. raise finally: # We can reach this point for different reasons: # - # 1. The task has finished properly (e.g. `EndFrame`). - # 2. By calling `PipelineTask.cancel()`. - # 3. By asyncio task cancellation. - # - # Case (1) will execute the code below without issues because - # `self._finished` is true. - # - # Case (2) will execute the code below without issues because - # `self._cancelled` is true. - # - # Case (3) will raise the exception above (because we are cancelling - # the asyncio task). This will be then captured by the - # `PipelineRunner` which will call `PipelineTask.cancel()` and - # therefore becoming case (2). - if self._finished or self._cancelled: - logger.debug(f"Pipeline task {self} is finishing cleanup...") - await self._cancel_tasks() - await self._cleanup(cleanup_pipeline) - if self._check_dangling_tasks: - self._print_dangling_tasks() - self._finished = True - logger.debug(f"Pipeline task {self} has finished") + # 1. The pipeline task has finished (try case). + # 2. By an asyncio task cancellation (except case). + logger.debug(f"Pipeline task {self} is finishing...") + await self._cancel_tasks() + if self._check_dangling_tasks: + self._print_dangling_tasks() + self._finished = True + logger.debug(f"Pipeline task {self} has finished") async def queue_frame(self, frame: Frame): """Queue a single frame to be pushed down the pipeline. @@ -489,19 +475,7 @@ class PipelineTask(BasePipelineTask): if not self._cancelled: logger.debug(f"Cancelling pipeline task {self}") self._cancelled = True - cancel_frame = CancelFrame() - # Make sure everything is cleaned up downstream. This is sent - # out-of-band from the main streaming task which is what we want since - # we want to cancel right away. - await self._pipeline.queue_frame(cancel_frame) - # Wait for CancelFrame to make it through the pipeline. - await self._wait_for_pipeline_end(cancel_frame) - # Only cancel the push task, we don't want to be able to process any - # other frame after cancel. Everything else will be cancelled in - # run(). - if self._process_push_task: - await self._task_manager.cancel_task(self._process_push_task) - self._process_push_task = None + await self.queue_frame(CancelFrame()) async def _create_tasks(self): """Create and start all pipeline processing tasks.""" @@ -603,6 +577,17 @@ class PipelineTask(BasePipelineTask): self._pipeline_end_event.clear() + # We are really done. + self._pipeline_finished_event.set() + + async def _wait_for_pipeline_finished(self): + await self._pipeline_finished_event.wait() + self._pipeline_finished_event.clear() + # Make sure we wait for the main task to complete. + if self._process_push_task: + await self._process_push_task + self._process_push_task = None + async def _setup(self, params: PipelineTaskParams): """Set up the pipeline task and all processors.""" mgr_params = TaskManagerParams(loop=params.loop) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index df4af49b9..7d9ebec6f 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -254,7 +254,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): try: await asyncio.wait_for( - asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), + task.run(PipelineTaskParams(loop=asyncio.get_event_loop())), timeout=1.0, ) except asyncio.TimeoutError: @@ -290,7 +290,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): await task.queue_frame(TextFrame(text="Hello!")) try: await asyncio.wait_for( - asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), + task.run(PipelineTaskParams(loop=asyncio.get_event_loop())), timeout=1.0, ) except asyncio.TimeoutError: @@ -301,11 +301,8 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): identity = IdentityFilter() pipeline = Pipeline([identity]) task = PipelineTask(pipeline, idle_timeout_secs=0.2) - try: - await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) - assert False - except asyncio.CancelledError: - assert True + # This shouldn't freeze, so nothing to check really. + await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) async def test_no_idle_task(self): identity = IdentityFilter() @@ -313,7 +310,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): task = PipelineTask(pipeline, idle_timeout_secs=0.2, cancel_on_idle_timeout=False) try: await asyncio.wait_for( - asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), + task.run(PipelineTaskParams(loop=asyncio.get_event_loop())), timeout=0.3, ) except asyncio.TimeoutError: @@ -332,11 +329,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): ), idle_timeout_secs=0.3, ) - try: - await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) - assert False - except asyncio.CancelledError: - assert True + await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) async def test_idle_task_event_handler_no_frames(self): identity = IdentityFilter() @@ -351,11 +344,8 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): idle_timeout = True await task.cancel() - try: - await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) - assert False - except asyncio.CancelledError: - assert idle_timeout + await task.run(PipelineTaskParams(loop=asyncio.get_event_loop())) + assert idle_timeout async def test_idle_task_event_handler_quiet_user(self): identity = IdentityFilter() @@ -416,12 +406,15 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): asyncio.create_task(delayed_frames()), ] - await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + _, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) diff_time = time.time() - start_time self.assertGreater(diff_time, sleep_time_secs * 3) + # Wait for the pending tasks to complete. + await asyncio.gather(*pending) + async def test_task_cancel_timeout(self): class CancelFilter(FrameProcessor): def __init__(self, **kwargs): From 8d8503bca7ac3a21ca9ad3eeb516cacdea648346 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 Oct 2025 16:20:12 -0700 Subject: [PATCH 5/5] OpenAITTSService: allow updating instructions and speed --- CHANGELOG.md | 5 +++- src/pipecat/services/openai/tts.py | 43 +++++++++++++++++++++++++----- 2 files changed, 41 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a5f114054..938b175c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added support for updating `OpenAITTSService` settings (`instructions` and + `speed`) at runtime via `TTSUpdateSettingsFrame`. + - Added `--whatsapp` flag to runner to better surface WhatsApp transport logs. - Added `on_connected` and `on_disconnected` events to TTS and STT @@ -27,7 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `CartesiaSTTService` now inherits from `WebsocketSTTService`. -# Package upgrades: +- Package upgrades: - `openai` upgraded to support up to 2.x.x. - `openpipe` upgraded to support up to 5.x.x. diff --git a/src/pipecat/services/openai/tts.py b/src/pipecat/services/openai/tts.py index a67392caf..cdf0d11ac 100644 --- a/src/pipecat/services/openai/tts.py +++ b/src/pipecat/services/openai/tts.py @@ -14,6 +14,7 @@ from typing import AsyncGenerator, Dict, Literal, Optional from loguru import logger from openai import AsyncOpenAI, BadRequestError +from pydantic import BaseModel from pipecat.frames.frames import ( ErrorFrame, @@ -55,6 +56,17 @@ class OpenAITTSService(TTSService): OPENAI_SAMPLE_RATE = 24000 # OpenAI TTS always outputs at 24kHz + class InputParams(BaseModel): + """Input parameters for OpenAI TTS configuration. + + Parameters: + instructions: Instructions to guide voice synthesis behavior. + speed: Voice speed control (0.25 to 4.0, default 1.0). + """ + + instructions: Optional[str] = None + speed: Optional[float] = None + def __init__( self, *, @@ -65,6 +77,7 @@ class OpenAITTSService(TTSService): sample_rate: Optional[int] = None, instructions: Optional[str] = None, speed: Optional[float] = None, + params: Optional[InputParams] = None, **kwargs, ): """Initialize OpenAI TTS service. @@ -77,7 +90,11 @@ class OpenAITTSService(TTSService): sample_rate: Output audio sample rate in Hz. If None, uses OpenAI's default 24kHz. instructions: Optional instructions to guide voice synthesis behavior. speed: Voice speed control (0.25 to 4.0, default 1.0). + params: Optional synthesis controls (acting instructions, speed, ...). **kwargs: Additional keyword arguments passed to TTSService. + + .. deprecated:: 0.0.91 + The `instructions` and `speed` parameters are deprecated, use `InputParams` instead. """ if sample_rate and sample_rate != self.OPENAI_SAMPLE_RATE: logger.warning( @@ -86,12 +103,26 @@ class OpenAITTSService(TTSService): ) super().__init__(sample_rate=sample_rate, **kwargs) - self._speed = speed self.set_model_name(model) self.set_voice(voice) - self._instructions = instructions self._client = AsyncOpenAI(api_key=api_key, base_url=base_url) + if instructions or speed: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "The `instructions` and `speed` parameters are deprecated, use `InputParams` instead.", + DeprecationWarning, + stacklevel=2, + ) + + self._settings = { + "instructions": params.instructions if params else instructions, + "speed": params.speed if params else speed, + } + def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -144,11 +175,11 @@ class OpenAITTSService(TTSService): "response_format": "pcm", } - if self._instructions: - create_params["instructions"] = self._instructions + if self._settings["instructions"]: + create_params["instructions"] = self._settings["instructions"] - if self._speed: - create_params["speed"] = self._speed + if self._settings["speed"]: + create_params["speed"] = self._settings["speed"] async with self._client.audio.speech.with_streaming_response.create( **create_params