Merge pull request #2959 from pipecat-ai/aleix/cancel-frame-reason

CancelFrame: add reason field to indicate why pipeline is being cancelled

CHANGELOG.md

@@ -5,6 +5,18 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- `EndFrame` and `EndTaskFrame` have an optional `reason` field to indicate why
+  the pipeline is being ended.
+
+- `CancelFrame` and `CancelTaskFrame` have an optional `reason` field to
+  indicate why the pipeline is being canceled. This can be also specified when
+  you cancel a task with `PipelineTask.cancel(reason="cancellation your
+  reason")`.
+
 ## [0.0.92] - 2025-10-31 🎃 "The Haunted Edition" 👻
 
 ### Added

src/pipecat/frames/frames.py

@@ -773,9 +773,15 @@ class CancelFrame(SystemFrame):
 
     Indicates that a pipeline needs to stop right away without
     processing remaining queued frames.
+
+    Parameters:
+        reason: Optional reason for pushing a cancel frame.
     """
 
-    pass
+    reason: Optional[str] = None
+
+    def __str__(self):
+        return f"{self.name}(reason: {self.reason})"
 
 
 @dataclass
@@ -1366,9 +1372,15 @@ class EndTaskFrame(TaskFrame):
     This is used to notify the pipeline task that the pipeline should be
     closed nicely (flushing all the queued frames) by pushing an EndFrame
     downstream. This frame should be pushed upstream.
+
+    Parameters:
+        reason: Optional reason for pushing an end frame.
     """
 
-    pass
+    reason: Optional[str] = None
+
+    def __str__(self):
+        return f"{self.name}(reason: {self.reason})"
 
 
 @dataclass
@@ -1378,9 +1390,15 @@ class CancelTaskFrame(TaskFrame):
     This is used to notify the pipeline task that the pipeline should be
     stopped immediately by pushing a CancelFrame downstream. This frame
     should be pushed upstream.
+
+    Parameters:
+        reason: Optional reason for pushing a cancel frame.
     """
 
-    pass
+    reason: Optional[str] = None
+
+    def __str__(self):
+        return f"{self.name}(reason: {self.reason})"
 
 
 @dataclass
@@ -1451,9 +1469,15 @@ class EndFrame(ControlFrame):
     sending frames to its output channel(s) and close all its threads. Note,
     that this is a control frame, which means it will be received in the order it
     was sent.
+
+    Parameters:
+        reason: Optional reason for pushing an end frame.
     """
 
-    pass
+    reason: Optional[str] = None
+
+    def __str__(self):
+        return f"{self.name}(reason: {self.reason})"
 
 
 @dataclass

src/pipecat/pipeline/task.py

@@ -446,10 +446,14 @@ class PipelineTask(BasePipelineTask):
         logger.debug(f"Task {self} scheduled to stop when done")
         await self.queue_frame(EndFrame())
 
-    async def cancel(self):
-        """Request the running pipeline to cancel."""
+    async def cancel(self, *, reason: Optional[str] = None):
+        """Request the running pipeline to cancel.
+
+        Args:
+            reason: Optional reason to indicate why the pipeline is being cancelled.
+        """
         if not self._finished:
-            await self._cancel()
+            await self._cancel(reason=reason)
 
     async def run(self, params: PipelineTaskParams):
         """Start and manage the pipeline execution until completion or cancellation.
@@ -513,12 +517,16 @@ class PipelineTask(BasePipelineTask):
             for frame in frames:
                 await self.queue_frame(frame)
 
-    async def _cancel(self):
-        """Internal cancellation logic for the pipeline task."""
+    async def _cancel(self, *, reason: Optional[str] = None):
+        """Internal cancellation logic for the pipeline task.
+
+        Args:
+            reason: Optional reason to indicate why the pipeline is being cancelled.
+        """
         if not self._cancelled:
             logger.debug(f"Cancelling pipeline task {self}")
             self._cancelled = True
-            await self.queue_frame(CancelFrame())
+            await self.queue_frame(CancelFrame(reason=reason))
 
     async def _create_tasks(self):
         """Create and start all pipeline processing tasks."""
@@ -716,11 +724,11 @@ class PipelineTask(BasePipelineTask):
         if isinstance(frame, EndTaskFrame):
             # Tell the task we should end nicely.
             logger.debug(f"{self}: received end task frame {frame}")
-            await self.queue_frame(EndFrame())
+            await self.queue_frame(EndFrame(reason=frame.reason))
         elif isinstance(frame, CancelTaskFrame):
             # Tell the task we should end right away.
             logger.debug(f"{self}: received cancel task frame {frame}")
-            await self.queue_frame(CancelFrame())
+            await self.queue_frame(CancelFrame(reason=frame.reason))
         elif isinstance(frame, StopTaskFrame):
             # Tell the task we should stop nicely.
             logger.debug(f"{self}: received stop task frame {frame}")