# Pipecat

> Run an unmodified Pipecat pipeline on real phone calls. Install pipecat-clutchcall, serve one pipeline for each engine-routed call, and route a number to your worker.

You have a Pipecat bot: a pipeline of frame processors connected to STT, an
LLM, and TTS. You want it to answer real phone calls, and you do not want to
operate a media server. The `pipecat-clutchcall` package runs your bot as an
**external agent**. The external agent is a worker process. The worker dials
**one outbound QUIC/WebTransport session** to FirstByte. It authenticates
with a media app key/secret. It registers under an agent handle. The engine
routes each phone call to the worker through a `VENDOR_BRIDGE` agent config.
For each call, the engine hands your bot a live, drop-in Pipecat transport.
Your pipeline, services, aggregators, and event handlers run **verbatim**.
Only the transport changes.

```
 PSTN / SIP trunk ─▶ ┌──────────────────┐                    ┌───────────────────────────┐
                     │  voice gateway   │  ONE QUIC/WT       │ your Pipecat worker       │
                     │  (owns the       │◀── session ───────▶│  + pipecat-clutchcall      │
                     │  caller leg)     │  (worker dials out)│  (your STT/LLM/TTS keys)  │
                     └──────────────────┘                    └───────────────────────────┘
```

The worker is always the QUIC client. It dials out to `:443` and opens no
inbound ports. It runs behind NAT with no public URL and no tunnel.

## What maps to what

| In a stock Pipecat deployment | On FirstByte |
| --- | --- |
| Daily / WebRTC / WebSocket transport | `ClutchCallTransport(call)`, a drop-in Pipecat `BaseTransport` |
| `transport.input()` / `transport.output()` | Same calls, backed by the call's 20 ms `pcm16` frames |
| `on_client_connected` / `on_client_disconnected` | Fire when the engine routes the call to you / when the caller hangs up |
| One room or socket per user | One `Call` for each phone call. `run_pipecat_worker` runs one pipeline for each call |
| Wideband mic audio | 8 kHz telephony leg. The transport declares 8 kHz, thus Pipecat resamples your TTS output automatically |
| Caller identity from your own signaling | `call.caller_number` (ANI) / `call.called_number` (DNIS) |

## Run the quickstart bot on a phone number

The runnable starter is a port of `pipecat-quickstart`. It is the official
quickstart `bot.py` with one change: `ClutchCallTransport` replaces the
Daily/WebRTC transport. It ships in the provider starters repo next to the
`clutchcall-agents` SDK.

  1. **Install the packages**
```bash
pip install pipecat-clutchcall "clutchcall-agents[quic]" \
  --extra-index-url https://artifacts.clutchcall.dev/pip/simple/
```

    The `quic` extra installs the WebTransport client. Your Pipecat services
    (`pipecat-ai[...]` extras, provider keys) do not change.

  2. **Swap the transport in your bot**
Your bot becomes a `bot(transport, call)` coroutine. The content inside it
    is the upstream quickstart, verbatim:

```python bot.py
import asyncio
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.frames.frames import LLMRunFrame

from pipecat_clutchcall import ClutchCallTransport, run_pipecat_worker
from clutchcall_agents import AgentConfig, Call

async def run_bot(transport: ClutchCallTransport, call: Call):
    stt, llm, tts = ...  # your services, unchanged (Deepgram/OpenAI/Cartesia…)

    pipeline = Pipeline([
        transport.input(),    # caller audio in
        stt, user_aggregator, llm, tts,
        transport.output(),   # agent audio back to the caller
        assistant_aggregator,
    ])
    task = PipelineTask(pipeline)

    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport, client):
        # Phone callers expect the agent to speak first.
        await task.queue_frames([LLMRunFrame()])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        await task.cancel()

    await PipelineRunner(handle_sigint=False).run(task)

if __name__ == "__main__":
    asyncio.run(run_pipecat_worker(run_bot, AgentConfig.from_env()))
```

    `run_pipecat_worker` holds the session open. It invokes `run_bot` one time
    for each engine-assigned call, with a fresh transport bound to that call's
    media.

    > **WARNING:**
> Do not share stateful `TransportParams` (for example, VAD analyzers)
>       across calls. Build them inside `run_bot`, or pass `params_factory=` to
>       `run_pipecat_worker`. Then each call gets a fresh instance.

  3. **Create the external agent in the console**
The console's **External agents** create flow (`agent.firstbyte.fast`)
    does three tasks at the same time:

    - It reserves the **agent handle** that your worker registers under.
    - It mints the **media app key/secret** that the worker authenticates with.
    - It writes the `VENDOR_BRIDGE` agent config that routes calls to the
      worker:

    ```json agent-config.json
    {
      "entry_node": "bridge",
      "pipeline": {
        "bridge": {
          "node_type": "VENDOR_BRIDGE",
          "vendor_provider": "clutchcall",
          "vendor_room": "quickstart-bot"
        }
      }
    }
    ```

    Export the credentials. `AgentConfig.from_env()` reads them:

```bash
export CLUTCHCALL_HOST=engine.firstbyte.fast
export CLUTCHCALL_MEDIA_KEY=...      # minted in the External-agent create flow
export CLUTCHCALL_MEDIA_SECRET=...
export CLUTCHCALL_AGENT=quickstart-bot
python bot.py
```

  4. **Route a number and dial it**
Attach the agent to an inbound DID or trunk in the same way as a native
    agent. See [SIP Trunking](/modalities/voice/transport-telephony/sip-trunking).
    Dial the number. The engine dispatches the call to your worker. CDR /
    MOS / recording land in
    [Observability](/modalities/voice/observability/overview) in the same way
    as any call.

## Troubleshooting

  **The worker registers, but calls find no agent**

`vendor_room` in the agent config must exactly match the handle in
    `CLUTCHCALL_AGENT`. Presence is tied to the live session. Check that the
    worker logged a ready line after its most recent reconnect.

  **The agent's voice sounds fast, slow, or robotic**

A resampling stage is skipped. The transport declares the call's sample
    rate (8 kHz on PSTN) on `transport.output()`. If you construct your own
    `TransportParams`, do not set `audio_out_sample_rate`. It then inherits
    the call's rate.

  **The worker connects but no audio flows**

Media rides QUIC datagrams. Check that outbound UDP/443 to
    `engine.firstbyte.fast` is open. A network that permits TCP/443 but drops UDP
    completes presence and stalls media.

## Related

  - **[LiveKit integration](/modalities/voice/integrations/livekit)** — The same external-agent seam for LiveKit Agents workers.
  - **[Run a LiveKit Agent on the Transport](/modalities/voice/recipes/livekit-transport)** — The end-to-end external-agent recipe on the same seam.
  - **[SIP Trunking](/modalities/voice/transport-telephony/sip-trunking)** — Bring a carrier trunk and route DIDs to your agents.
  - **[Custom Agent Runtime](/modalities/voice/integrations/custom-runtime)** — The same worker pattern without Pipecat: raw frames in, raw frames out.