---
name: Pipecat
description: Use when building voice AI agents and bots, designing conversation flows, configuring real-time speech pipelines, deploying to production, or integrating with LLMs, speech services, and telephony providers. Reach for this skill when working with Pipecat's pipeline architecture, multi-agent systems, or Pipecat Cloud deployments.
metadata:
    mintlify-proj: pipecat
    version: "1.0"
---

# Pipecat Skill Reference

## Product Summary

Pipecat is a Python framework for building real-time voice AI agents and bots. It orchestrates pipelines that connect transports (WebRTC, WebSocket, telephony), speech-to-text (STT), language models (LLM), and text-to-speech (TTS) services into conversational agents. Key files: `bot.py` (agent entry point), `pcc-deploy.toml` (Pipecat Cloud config), `.env` (API keys). CLI: `pipecat init`, `pipecat cloud deploy`, `pipecat eval`. Integrations: 50+ STT/TTS/LLM providers, Daily/Twilio/LiveKit transports, Flows for structured conversations. Primary docs: https://docs.pipecat.ai

## When to Use

- **Building voice agents**: Create conversational bots that listen, understand, and respond in real-time
- **Structuring complex conversations**: Use Flows to break multi-step tasks into focused nodes with specific tools
- **Deploying to production**: Ship agents to Pipecat Cloud or self-host with scaling patterns
- **Integrating services**: Wire together STT (Deepgram, OpenAI), LLMs (OpenAI, Anthropic, Gemini), TTS (Cartesia, ElevenLabs), and transports
- **Handling telephony**: Build phone agents with Twilio, Telnyx, Plivo, or Exotel
- **Multi-agent systems**: Coordinate multiple agents over a shared bus with handoffs and job dispatch
- **Testing and evaluation**: Run scripted or simulated scenarios to validate agent behavior

## Quick Reference

### Core Architecture

| Component | Purpose | Example |
|-----------|---------|---------|
| **Pipeline** | Ordered sequence of frame processors | `Pipeline([transport.input(), stt, llm, tts, transport.output()])` |
| **Transport** | User connection layer (WebRTC, WebSocket, etc.) | `DailyTransport`, `FastAPIWebsocketTransport` |
| **STT** | Speech-to-text service | `DeepgramSTTService`, `OpenAISTTService` |
| **LLM** | Language model | `OpenAILLMService`, `AnthropicLLMService` |
| **TTS** | Text-to-speech service | `CartesiaTTSService`, `ElevenLabsTTSService` |
| **Worker** | Agent that runs a pipeline | `PipelineWorker`, `LLMWorker`, `LLMContextWorker` |
| **WorkerRunner** | Manages workers and the message bus | `WorkerRunner()` |

### Essential Commands

```bash
# Initialize a new project
pipecat init quickstart

# Run locally
uv run bot.py

# Deploy to Pipecat Cloud
pipecat cloud auth login
pipecat cloud secrets set <secret-set> --file .env
pipecat cloud deploy

# Run evaluations
pipecat eval --suite my_suite.yaml

# Manage context hub (for coding agents)
pipecat context-hub install
```

### Configuration Files

| File | Purpose |
|------|---------|
| `.env` | API keys for STT, LLM, TTS services |
| `pcc-deploy.toml` | Pipecat Cloud deployment config (agent name, secrets, scaling) |
| `Dockerfile` | Container image for cloud deployment |
| `flow.yaml` or `flow.json` | Declarative conversation flow (Flows) |

### Pipeline Parameter Essentials

```python
from pipecat.pipeline.worker import PipelineParams

params = PipelineParams(
    audio_in_sample_rate=16000,      # Input audio sample rate
    audio_out_sample_rate=24000,     # Output audio sample rate
    enable_metrics=True,              # Collect performance metrics
    enable_usage_metrics=True,        # Track token/character usage
)
```

### Context Management

```python
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair

context = LLMContext()
user_agg, assistant_agg = LLMContextAggregatorPair(context)

# Add to pipeline
pipeline = Pipeline([
    transport.input(),
    stt,
    user_agg,
    llm,
    tts,
    transport.output(),
    assistant_agg,
])
```

## Decision Guidance

### When to Use Flows vs. Direct Pipeline

| Scenario | Use Flows | Use Direct Pipeline |
|----------|-----------|-------------------|
| Simple single-turn conversation | ❌ | ✅ |
| Multi-step task with different tools per step | ✅ | ❌ |
| Need to change prompts without redeploying | ✅ | ❌ |
| Complex conditional logic based on runtime state | ✅ | ❌ |
| Real-time node creation based on conversation | ✅ | ❌ |
| Straightforward STT → LLM → TTS flow | ❌ | ✅ |

### When to Use Each Transport

| Transport | Best For | Latency | Resilience |
|-----------|----------|---------|-----------|
| **Daily WebRTC** | Browser/mobile clients, video | Low | High (built-in) |
| **SmallWebRTC** | P2P, development, low-latency | Very Low | Medium |
| **FastAPI WebSocket** | Telephony (Twilio, Telnyx), server-to-server | Medium | Medium |
| **LiveKit** | Scalable WebRTC, multi-participant | Low | High |
| **HeyGen/Tavus** | Avatar video generation | Medium | Medium |

### Error Handling Strategy

| Situation | Policy | Rationale |
|-----------|--------|-----------|
| Service can fail over (e.g., STT with backup) | `CONTINUE` | Let ServiceSwitcher handle it |
| Service failure makes bot useless | `END` | Gracefully drain queued frames |
| Immediate shutdown required | `CANCEL` | Abandon pending work (user disconnected) |

## Workflow

### 1. Build a Local Agent

1. **Scaffold the project**: `pipecat init quickstart`
2. **Configure API keys**: Copy `.env.example` to `.env`, add keys for STT, LLM, TTS
3. **Review the pipeline**: Understand the order: `transport.input() → stt → llm → tts → transport.output()`
4. **Run locally**: `uv run bot.py`, open the browser URL, test the agent
5. **Iterate**: Modify prompts, swap services, adjust parameters

### 2. Deploy to Pipecat Cloud

1. **Authenticate**: `pipecat cloud auth login`
2. **Upload secrets**: `pipecat cloud secrets set <secret-set> --file .env`
3. **Configure deployment**: Edit `pcc-deploy.toml` (agent name, secret set, scaling)
4. **Deploy**: `pipecat cloud deploy` (builds and deploys automatically)
5. **Test**: Open Pipecat Cloud dashboard, select agent, click Sandbox, connect

### 3. Build a Structured Conversation with Flows

1. **Write a flow config** (YAML/JSON) or **build nodes in Python**
2. **Define tools** as direct functions or `FunctionSchema` objects
3. **Create FlowManager** and pass the flow
4. **Integrate with pipeline**: Place FlowManager in the pipeline
5. **Test transitions**: Verify tool calls move between nodes correctly

### 4. Handle Multi-Agent Coordination

1. **Create multiple workers**: `PipelineWorker` or `LLMWorker` for each agent
2. **Register with runner**: `await runner.add_workers(agent1, agent2, ...)`
3. **Implement handoff**: Agent calls `activate_worker()` to transfer control
4. **Share context**: Use the bus to pass messages between agents

### 5. Add Observability

1. **Enable metrics**: Set `enable_metrics=True` in `PipelineParams`
2. **Add observers**: Attach `MetricsLogObserver`, `LLMLogObserver`, etc. to the worker
3. **Log errors**: Implement `on_error` handlers on services
4. **Monitor in production**: Use Pipecat Cloud logs or external tools (Datadog, Sentry)

## Common Gotchas

- **Frame order matters**: Audio must be transcribed before LLM processes it. Arrange processors in the correct sequence.
- **Sample rate mismatches**: Set audio sample rates in `PipelineParams`, not on individual services. Mismatches cause silent failures.
- **Forgetting context aggregators**: Without `LLMContextAggregatorPair`, the LLM has no conversation history. Always add both user and assistant aggregators.
- **Muting user input**: By default, users can interrupt the bot. If you need to suppress input during bot speech, configure `user_mute_strategies` on the user aggregator.
- **Flows require text LLMs**: Speech-to-speech models (Gemini Live, OpenAI Realtime) don't support Flows because they don't expose context/tools mid-session.
- **Permanent vs. recoverable errors**: An unusable processor (bad API key) won't retry. Check `is_usable` and use `ServiceSwitcher` for failover.
- **Transport placement**: `transport.output()` doesn't have to be last. Post-output processors (recording, context capture) can follow it.
- **Missing event handlers**: Forgetting `@transport.event_handler("on_client_disconnected")` leaves the runner hanging when clients disconnect.
- **Secrets in code**: Never hardcode API keys. Use `.env` files and `pipecat cloud secrets` for production.
- **Interruptions disabled**: If interruptions are off, users can't stop the bot mid-response. Enable them for natural conversations.

## Verification Checklist

Before submitting a Pipecat agent or deployment:

- [ ] All API keys are in `.env` (not hardcoded) and uploaded to Pipecat Cloud secrets
- [ ] Pipeline processors are in the correct order (input → STT → LLM → TTS → output)
- [ ] Audio sample rates are set in `PipelineParams`, not on individual services
- [ ] Context aggregators (user and assistant) are included in the pipeline
- [ ] Transport `on_client_disconnected` handler calls `runner.cancel()` to clean up
- [ ] Error handling is configured: `ProcessorUnusablePolicy` is set appropriately
- [ ] Metrics are enabled: `enable_metrics=True` in `PipelineParams`
- [ ] For Flows: all tool functions are registered and transitions are tested
- [ ] For multi-agent: all workers are registered with `runner.add_workers()`
- [ ] For production: `pcc-deploy.toml` has correct agent name, secret set, and scaling config
- [ ] Tested locally with `uv run bot.py` before deploying
- [ ] Pipecat Cloud deployment succeeds: `pipecat cloud deploy` completes without errors

## Resources

**Comprehensive navigation**: https://docs.pipecat.ai/llms.txt — page-by-page reference for all Pipecat documentation

**Critical pages**:
- [Quickstart](https://docs.pipecat.ai/pipecat/get-started/quickstart) — Build and deploy your first agent in 10 minutes
- [Your First Agent](https://docs.pipecat.ai/pipecat/learn/your-first-agent) — Understand the agent architecture and worker lifecycle
- [Pipeline & Frame Processing](https://docs.pipecat.ai/pipecat/learn/pipeline) — Learn how data flows through processors and frames
- [Pipecat Flows](https://docs.pipecat.ai/pipecat/flows/introduction) — Structure conversations as a graph of nodes
- [Deployment Overview](https://docs.pipecat.ai/pipecat/deployment/overview) — Choose a hosting pattern and deployment strategy
- [Error Handling](https://docs.pipecat.ai/pipecat/fundamentals/error-handling) — Handle service failures and recovery

---

> For additional documentation and navigation, see: https://docs.pipecat.ai/llms.txt