# Pipecat

Run a Pipecat bot on Vonyx calls through LiveKit SIP.

A Pipecat bot can make and take calls on your Vonyx number. Vonyx connects to LiveKit SIP over your SIP trunk, and the bot joins the call's LiveKit room with Pipecat's `LiveKitTransport`.

```text
Phone ⇄ Vonyx ⇄ (SIP) ⇄ LiveKit SIP ⇄ LiveKit room ⇄ Pipecat bot (LiveKitTransport)
```

<Callout title="Why LiveKit, not Daily or a WebSocket serializer">
Pipecat's Daily SIP route doesn't fit a username/password trunk: Daily's `sipUri` dial-out has no field for SIP credentials, and Daily dial-in expects the carrier to forward each call to a per-room SIP URI, while Vonyx forwards to one fixed inbound address. Pipecat's Twilio, Telnyx, Plivo and Exotel serializers only speak those providers' WebSocket formats. If you already run [Asterisk on Vonyx](/docs/integrations/asterisk), Pipecat lists a community-maintained Asterisk serializer (`pipecat-asterisk`) that Pipecat doesn't test or support. A native Vonyx WebSocket route for Pipecat is coming.
</Callout>

## Before you start

- A Vonyx [SIP trunk](/docs/console/sip-trunks) with a [number](/docs/console/numbers) attached
- A LiveKit Cloud project with its `LIVEKIT_URL`, `LIVEKIT_API_KEY` and `LIVEKIT_API_SECRET`
- The [LiveKit CLI](https://docs.livekit.io/home/cli/) signed in to that project
- Python with Pipecat and the LiveKit extra, plus keys for your STT, LLM and TTS services

```bash
pip install "pipecat-ai[livekit,deepgram,openai,cartesia,silero]" fastapi uvicorn python-dotenv
```

```bash title=".env"
LIVEKIT_URL=wss://your-project.livekit.cloud
LIVEKIT_API_KEY=your_livekit_api_key
LIVEKIT_API_SECRET=your_livekit_api_secret
LIVEKIT_SIP_TRUNK_ID=ST_xxxxxxxx
DEEPGRAM_API_KEY=...
OPENAI_API_KEY=...
CARTESIA_API_KEY=...
```

## Write the bot

One bot file serves both directions. It joins a LiveKit room as a participant. With a number to dial, it places the call itself and greets the callee once they answer. Without one, it waits for the caller who is already in the room.

```python title="bot.py"
import asyncio
import os

from dotenv import load_dotenv
from livekit import api
from loguru import logger

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.runner.livekit import generate_token_with_agent
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.livekit.transport import LiveKitParams, LiveKitTransport
from pipecat.workers.runner import WorkerRunner

load_dotenv(override=True)

GREETING = "Hi, this is the assistant. How can I help?"


async def run_bot(room_name: str, dial_to: str | None = None):
    token = generate_token_with_agent(
        room_name, "Pipecat Bot", os.environ["LIVEKIT_API_KEY"], os.environ["LIVEKIT_API_SECRET"]
    )
    transport = LiveKitTransport(
        url=os.environ["LIVEKIT_URL"],
        token=token,
        room_name=room_name,
        params=LiveKitParams(audio_in_enabled=True, audio_out_enabled=True),
    )

    stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        settings=OpenAILLMService.Settings(
            system_instruction="You are a helpful phone assistant. Keep replies short and spoken.",
        ),
    )
    tts = CartesiaTTSService(
        api_key=os.environ["CARTESIA_API_KEY"],
        settings=CartesiaTTSService.Settings(voice="YOUR_CARTESIA_VOICE_ID"),
    )

    context = LLMContext()
    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
        context, user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer())
    )
    pipeline = Pipeline([
        transport.input(),
        stt,
        user_aggregator,
        llm,
        tts,
        transport.output(),
        assistant_aggregator,
    ])
    worker = PipelineWorker(pipeline)

    async def dial():
        lkapi = api.LiveKitAPI()
        try:
            await lkapi.sip.create_sip_participant(api.CreateSIPParticipantRequest(
                sip_trunk_id=os.environ["LIVEKIT_SIP_TRUNK_ID"],
                sip_call_to=dial_to,
                room_name=room_name,
                participant_identity="callee",
                wait_until_answered=True,
            ))
        except api.SipCallError as e:
            logger.error(f"Call failed: {e.sip_status_code} {e.sip_status}")
            await worker.cancel()
            return
        finally:
            await lkapi.aclose()
        await worker.queue_frame(TTSSpeakFrame(GREETING))

    @transport.event_handler("on_connected")
    async def on_connected(transport):
        if dial_to:
            asyncio.create_task(dial())

    @transport.event_handler("on_first_participant_joined")
    async def on_first_participant_joined(transport, participant_id):
        if not dial_to:  # inbound: the caller is already on the line
            await worker.queue_frame(TTSSpeakFrame(GREETING))

    @transport.event_handler("on_participant_left")
    async def on_participant_left(transport, participant_id, reason):
        await worker.cancel()  # the phone side hung up

    runner = WorkerRunner()
    await runner.add_workers(worker)
    await runner.run()
```

Dialling waits for the answer (`wait_until_answered=True`) before the greeting, so the callee doesn't miss it while the phone rings.

## Outbound calls

<Steps>
<Step>

### Create the LiveKit outbound trunk

Follow [Outbound calls in the LiveKit guide](/docs/integrations/livekit#outbound-calls) to create an outbound trunk with your Vonyx address, trunk username, password and number. Put its trunk ID (`ST_…`) in `LIVEKIT_SIP_TRUNK_ID`.

</Step>
<Step>

### Start the bot with a number

```python title="call.py"
import asyncio
import uuid

from bot import run_bot

asyncio.run(run_bot(f"out-{uuid.uuid4().hex[:8]}", dial_to="91XXXXXXXXXX"))
```

```bash
python call.py
```

The bot joins a new room, LiveKit dials the number through Vonyx, and the bot speaks once the callee answers. The callee sees the Vonyx number on the outbound trunk as caller ID. To use a different one of your Vonyx numbers per call, set `sip_number` on the `CreateSIPParticipantRequest`.

</Step>
</Steps>

## Inbound calls

LiveKit SIP creates a room for each call. A Pipecat bot isn't a LiveKit Agents worker, so LiveKit can't dispatch it. Instead, a LiveKit webhook tells your server when a call room starts, and the server runs the bot in it.

<Steps>
<Step>

### Create the inbound trunk and point Vonyx at it

Follow [Inbound calls in the LiveKit guide](/docs/integrations/livekit#inbound-calls) to find your LiveKit SIP endpoint, create the inbound trunk and set it as the trunk's inbound address in the Vonyx console. Skip that guide's dispatch rule and use the one below.

</Step>
<Step>

### Add a dispatch rule without an agent

This rule puts each caller in a new room prefixed `call-`. It has no `room_config.agents`, because your webhook starts the bot.

```json title="dispatch-rule.json"
{
  "dispatch_rule": {
    "name": "Vonyx to Pipecat",
    "trunk_ids": ["ST_inbound_xxxxxxxx"],
    "rule": {
      "dispatch_rule_individual": {
        "room_prefix": "call-"
      }
    }
  }
}
```

```bash
lk sip dispatch create dispatch-rule.json
```

</Step>
<Step>

### Run a webhook server

The server checks LiveKit's signature, then starts the bot for every `room_started` event on a `call-` room. Outbound rooms use the `out-` prefix, so they are ignored.

```python title="server.py"
import asyncio

from fastapi import FastAPI, HTTPException, Request
from livekit import api

from bot import run_bot

app = FastAPI()
receiver = api.WebhookReceiver(api.TokenVerifier())  # reads LIVEKIT_API_KEY / LIVEKIT_API_SECRET


@app.post("/livekit/webhook")
async def livekit_webhook(request: Request):
    body = (await request.body()).decode()
    try:
        event = receiver.receive(body, request.headers.get("Authorization", ""))
    except Exception:
        raise HTTPException(status_code=401)

    if event.event == "room_started" and event.room.name.startswith("call-"):
        asyncio.create_task(run_bot(event.room.name))
    return {"ok": True}
```

```bash
uvicorn server:app --port 8000
```

The server must be reachable from the internet (for local testing, use a tunnel).

</Step>
<Step>

### Register the webhook in LiveKit Cloud

In LiveKit Cloud, open **Settings** → **Webhooks** → **Create new webhook**. Enter a **Name**, set the **URL** to `https://your-server.example.com/livekit/webhook`, pick the **Signing API key** that matches `LIVEKIT_API_KEY`, and select **Create**.

</Step>
</Steps>

## Hang up from the bot

Stopping the pipeline only takes the bot out of the room; the caller would hear silence. To end the call, delete the room:

```python
lkapi = api.LiveKitAPI()
await lkapi.room.delete_room(api.DeleteRoomRequest(room=room_name))
await lkapi.aclose()
```

## Test the connection

1. **Trunk first:** check that `lk sip participant create` reaches your phone, as in the [LiveKit guide](/docs/integrations/livekit#test-the-connection). That rules out trunk problems before the bot is involved.
2. **Outbound:** run `python call.py` with your own phone number. Your phone should ring with your Vonyx number, and the bot should greet you when you answer.
3. **Inbound:** with `server.py` running and the webhook registered, call your Vonyx number. A `call-…` room should appear in LiveKit Cloud with a SIP participant and `Pipecat Bot`, and the bot should greet you.

## Limitations

- DTMF and call transfer are not available on Vonyx yet. Don't rely on Pipecat's `on_dtmf_event`, `OutputDTMFFrame` or `DTMFAggregator`, or on LiveKit SIP transfers.
- New accounts make 1 call at a time ([raise limits](/docs/sip/limits)).
- Codecs are PCMU and PCMA only, with no TLS or SRTP. LiveKit SIP handles the conversion to WebRTC audio for the bot.
- Caller ID must be one of your Vonyx numbers.
- The bot code above runs one bot per call inside the webhook server's process. Under real traffic, run bots as separate processes or jobs.

## Troubleshooting

| Problem | Fix |
| --- | --- |
| `Call failed: 401` or `407` | The trunk username or password on the LiveKit outbound trunk is wrong. |
| `Call failed: 486` or `603` | The callee rejected the call. |
| `Call failed: 408` or `480` | No answer, or the callee was unreachable. |
| Outbound call rejected right away | Check the outbound trunk's number is one of your Vonyx numbers and the dialled number is `91XXXXXXXXXX`. |
| The callee hears nothing after answering | Check the bot logs for STT, LLM or TTS key errors. The greeting is only queued after `wait_until_answered` returns. |
| Inbound room appears but the bot never joins | The webhook isn't reaching `server.py`, the signature check fails (wrong signing key), or the room name doesn't start with `call-`. Use **Send a test event** on the webhook in LiveKit Cloud. |
| Inbound call never reaches LiveKit | Check **Inbound calls** is on and the inbound address has no `sip:` prefix. See the [LiveKit guide](/docs/integrations/livekit#troubleshooting). |
| Caller hears silence after the bot stops | Delete the room to hang up (see [Hang up from the bot](#hang-up-from-the-bot)). |
| Second parallel call fails | New accounts have 1 channel. See [limits](/docs/sip/limits). |
| Call connects with no audio | See [SIP troubleshooting](/docs/sip/troubleshooting). Vonyx sends RTP over UDP 10000–20000. |