> ## Documentation Index
> Fetch the complete documentation index at: https://daily-docs-pr-4892.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
# Pipeline Termination
> Learn how to properly terminate Pipecat pipelines for clean shutdown and resource management
**Pipeline termination** ensures your voice AI applications shut down cleanly without resource leaks or hanging processes. Understanding the different termination methods helps you handle various scenarios from natural conversation endings to unexpected disconnections.
## Pipeline Integration
Pipeline termination works through the same frame-based system as other pipeline operations:
```python theme={null}
pipeline = Pipeline([
transport.input(),
stt,
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
])
# EndFrame or CancelFrame flows through entire pipeline for shutdown
```
**Termination frames:**
* **`EndFrame`**: A queued ControlFrame that triggers graceful shutdown after processing pending frames
* **`CancelFrame`**: A SystemFrame that triggers immediate shutdown, discarding pending frames
Both frames flow downstream through the pipeline, allowing each processor to clean up resources appropriately.
### Termination frames at a glance
| Frame | Job | Push from / direction |
| ------------------- | -------------------------------------------------- | ------------------------------------------------------------------------------------------------------- |
| `EndFrame` | Graceful shutdown (drains pending frames) | From outside the pipeline via `worker.queue_frame(EndFrame())` |
| `CancelFrame` | Immediate shutdown (discards pending frames) | Via `worker.cancel()` |
| `EndWorkerFrame` | Graceful shutdown signal from inside the pipeline | `push_frame(EndWorkerFrame(), FrameDirection.DOWNSTREAM)`; the source converts it to an `EndFrame` |
| `CancelWorkerFrame` | Immediate shutdown signal from inside the pipeline | `push_frame(CancelWorkerFrame(), FrameDirection.DOWNSTREAM)`; the source converts it to a `CancelFrame` |
If you see `EndTaskFrame`, `CancelTaskFrame`, `PipelineTask`, or
`task.cancel()` in older code or examples, those are **deprecated aliases**
(since 1.3.0/1.4.0) of `EndWorkerFrame`, `CancelWorkerFrame`,
`PipelineWorker`, and `worker.cancel()`. They still work but will be removed
in 2.0.0. Use the `Worker` names in new code.
## Termination Methods
Pipecat provides two primary approaches for pipeline termination, each designed for different scenarios:
### 1. Graceful Termination
Graceful termination allows the bot to complete its current processing before shutting down. This is ideal when you want the bot to properly end a conversation. For example, after completing a specific task or reaching a natural conclusion.
**When to use:**
* Natural conversation endings
* Task completion scenarios
* When the bot should say goodbye
**Implementation options:**
Push an `EndFrame` from outside your pipeline:
```python theme={null}
# From outside the pipeline
from pipecat.frames.frames import EndFrame
await worker.queue_frame(EndFrame())
```
Push an `EndWorkerFrame` downstream from inside your pipeline. Here `end_conversation` is a **direct function** — an `async` function whose first parameter is `params: FunctionCallParams`, with a Google-style docstring that becomes the tool description. Register it by passing the function itself in your tools list:
```python theme={null}
from pipecat.frames.frames import EndWorkerFrame, TTSSpeakFrame
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import FunctionCallParams
async def end_conversation(params: FunctionCallParams):
"""End the conversation and shut down the bot.
Call this when the user says goodbye or the task is complete.
"""
await params.llm.push_frame(TTSSpeakFrame("Have a nice day!"))
# Resolve the function call so the LLM call doesn't hang
await params.result_callback({"status": "ended"})
# Signal that the worker should end after processing this frame
await params.llm.push_frame(EndWorkerFrame(), FrameDirection.DOWNSTREAM)
# Pass the function directly in the tools list; it's registered automatically
context = LLMContext(tools=[end_conversation])
```
Always call `params.result_callback(...)` in your handler before pushing the
end frame. Skipping it can leave the LLM function call unresolved. If you
don't want the LLM to respond, you can provide `None` as the result.
**How graceful termination works:**
1. `EndFrame` is queued and processes after any pending frames (like goodbye messages)
2. All processors shutdown when they receive the `EndFrame`
3. Once the `EndFrame` reaches the end of the pipeline, shutdown is complete
4. Resources are cleaned up and the process terminates
Graceful termination allows your bot to say goodbye and complete any final
actions before terminating.
### 2. Immediate Termination
Immediate termination cancels the pipeline without waiting for pending frames to complete. This is appropriate when the user is no longer active in the conversation.
**When to use:**
* User disconnections (browser closed, call ended)
* Error conditions requiring immediate shutdown
* When completing the conversation is no longer necessary
**Implementation:**
Use event handlers to detect disconnections and trigger cancellation:
```python theme={null}
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info("Client disconnected - terminating pipeline")
await worker.cancel()
```
**How immediate termination works:**
1. An event triggers the cancellation (like client disconnection)
2. `worker.cancel()` pushes a `CancelFrame` downstream from the PipelineWorker
3. `CancelFrame`s are `SystemFrame`s, so they use the high-priority input queue and are processed before queued non-system frames
4. Processors handle the `CancelFrame` and shut down without waiting for pending non-system work to drain
5. Any pending frames are discarded during shutdown
Immediate termination will discard any pending frames in the pipeline. Use
this approach when completing the conversation is no longer necessary.
## Automatic Termination
### Pipeline Idle Detection
Pipecat includes automatic idle detection to prevent hanging pipelines. This feature monitors activity and can automatically cancel tasks when no meaningful bot interactions occur for an extended period.
**How it works:**
* Monitors pipeline activity for meaningful bot interactions
* Automatically triggers termination after configured idle timeout
* Serves as a safety net for anomalous behavior or forgotten sessions
**Configuration:**
```python theme={null}
worker = PipelineWorker(
pipeline,
# Configure idle detection timeout
cancel_on_idle_timeout=True, # Default is True
idle_timeout_secs=600, # Default is 300 seconds
idle_timeout_frames=(BotSpeakingFrame,), # Default is (BotSpeakingFrame, UserSpeakingFrame)
)
```
You can further configure the idle detection behavior. To learn more, refer to the Pipeline Idle Detection documentation:
Learn how to configure and customize idle detection for your use case
Pipeline Idle Detection is enabled by default and helps prevent resources from
being wasted on inactive conversations.
### Maximum Call Duration
Idle detection ends a call that goes quiet, but it does not cap the total length of an active call. To enforce a maximum call duration, run an `asyncio` timer that speaks a goodbye and then queues an `EndFrame` so the bot can sign off gracefully before shutdown:
```python theme={null}
import asyncio
from pipecat.frames.frames import EndFrame, TTSSpeakFrame
async def end_after(worker, timeout_secs: float):
await asyncio.sleep(timeout_secs)
await worker.queue_frame(TTSSpeakFrame("We've reached our time limit. Goodbye!"))
await worker.queue_frame(EndFrame()) # Graceful: plays the goodbye, then shuts down
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
asyncio.create_task(end_after(worker, timeout_secs=300)) # 5-minute cap
```
On Pipecat Cloud, there is also a platform-level hard cap via
[`maxSessionDuration`](/pipecat-cloud/fundamentals/active-sessions#session-duration-limits)
(default 7200s). That cap is a forced cut with no goodbye, so use the
bot-level timer above when you want the bot to speak before the call ends.
## Implementation Patterns
### Event-Driven Termination
Connect termination to transport events for automatic cleanup:
```python theme={null}
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info("Client connected - starting conversation")
await worker.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info("Client disconnected - immediate termination")
await worker.cancel()
# Run the pipeline
runner = WorkerRunner(handle_sigint=False)
await runner.run(worker)
```
### Conditional Termination
Use function calling or other logic to determine when conversations should end:
```python theme={null}
async def check_conversation_complete(params: FunctionCallParams):
# Your logic to determine if conversation should end
conversation_complete = await evaluate_completion_criteria()
if conversation_complete:
await params.llm.push_frame(TTSSpeakFrame("Thank you for using our service!"))
await params.llm.push_frame(EndWorkerFrame(), FrameDirection.DOWNSTREAM)
await params.result_callback({"status": "complete" if conversation_complete else "continuing"})
```
### Error Handling
Ensure pipelines can terminate properly even when exceptions occur:
```python theme={null}
try:
runner = WorkerRunner(handle_sigint=False)
await runner.run(worker)
except Exception as e:
logger.error(f"Pipeline error: {e}")
# Ensure cleanup happens even on errors
await worker.cancel()
```
### Running Cleanup Code on Shutdown
To run cleanup or persist data when a call ends, use the `on_pipeline_finished` event handler. It fires after the pipeline reaches any terminal state, so it runs for **both** graceful (`EndFrame`) and cancelled (`CancelFrame`) shutdowns. This makes it the single write point for end-of-call work like saving a transcript or recording:
```python theme={null}
@worker.event_handler("on_pipeline_finished")
async def on_pipeline_finished(worker, frame):
# Runs for both graceful and cancelled shutdown
await save_transcript_to_db()
```
`on_client_disconnected`, by contrast, fires only when the client disconnects. Use it to *tag the reason* for the shutdown (for example, "user hung up"), and do the actual persistence in `on_pipeline_finished` so you write data exactly once regardless of how the call ended. See the [`PipelineWorker` events](/api-reference/server/pipeline/pipeline-worker) reference for the full event signature.
## Troubleshooting
If your pipeline isn't shutting down properly, check these common issues:
### Custom Processors Not Propagating Frames
**Problem:** Custom processors that don't call `push_frame()` can block termination frames from reaching the end of the pipeline.
**Solution:** Ensure your custom processors propagate all frames downstream, including `EndFrame` and `CancelFrame`:
```python theme={null}
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
# Your custom processing logic here
# Always push frames downstream (including termination frames)
await self.push_frame(frame, direction)
```
### Incorrect Termination Frame Direction
**Problem:** Pushing `EndFrame` or `CancelFrame` from the middle of the pipeline may not reach the pipeline source properly.
**Solution:** Use the appropriate frame type and direction:
```python theme={null}
await self.push_frame(EndWorkerFrame(), FrameDirection.DOWNSTREAM)
await self.push_frame(CancelWorkerFrame(), FrameDirection.DOWNSTREAM)
# The pipeline source will then convert these to proper termination frames
# and push them downstream through the entire pipeline
```
The pipeline source automatically converts `EndWorkerFrame` to `EndFrame` and
`CancelWorkerFrame` to `CancelFrame` when pushing downstream, ensuring proper
termination handling throughout the pipeline.
### "dangling tasks detected" Warning
**Problem:** On shutdown you see a log warning like `PipelineWorker#0 dangling tasks detected: [...]`.
**Solution:** This means `asyncio` tasks created during the session were never cancelled or awaited before the pipeline shut down. The usual causes are the two above: a custom processor that doesn't propagate termination frames, or a background task started inside a processor (for example, a timer or long-running coroutine) that isn't cleaned up. Create background tasks through the pipeline task manager so they are tracked and cancelled on shutdown, and make sure your processors push `EndFrame`/`CancelFrame` downstream.
## Key Takeaways
* **Frame-based termination** - shutdown uses the same frame system as processing
* **Choose the right method** - graceful for natural endings, immediate for disconnections
* **Event handlers enable automatic termination** - respond to user disconnections cleanly
* **Idle detection provides safety net** - prevents hanging processes and resource waste
* **SystemFrames have priority** - CancelFrames are processed before queued non-system frames for fast shutdown
* **Resource cleanup is automatic** - proper termination ensures clean resource disposal
## What's Next
You now understand how to build, run, and properly terminate voice AI pipelines! With the single-agent basics covered, let's see how Pipecat coordinates multiple agents, starting with giving an agent its own LLM and tools.
Give an agent its own LLM and register tools with the @tool decorator