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

CHANGELOG.md

@@ -5,6 +5,15 @@ 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
+
+- `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
@@ -1378,9 +1384,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

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."""
@@ -720,7 +728,7 @@ class PipelineTask(BasePipelineTask):
         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}")