Build a phone agent with Pipecat, Deepgram, and ElevenLabs

Building a phone agent that feels natural requires stitching together speech-to-text, LLM reasoning, and text-to-speech with sub-second latency. Pipecat handles the orchestration, Deepgram provides streaming transcription, and ElevenLabs delivers expressive synthesis. This tutorial walks through a complete implementation that answers real calls on a Twilio number.

Step 1: Provision the infrastructure

You need three API keys and a Twilio account. Create a .env file to keep them out of source control:

# .env
DEEPGRAM_API_KEY=your_deepgram_key
ELEVENLABS_API_KEY=your_elevenlabs_key
OPENAI_API_KEY=your_openai_key
TWILIO_ACCOUNT_SID=your_twilio_sid
TWILIO_AUTH_TOKEN=your_twilio_token
TWILIO_PHONE_NUMBER=+15551234567

Install the dependencies. Pipecat’s pipecat-ai package includes the pipeline primitives; pipecat-services-deepgram and pipecat-services-elevenlabs provide the service adapters.

pip install pipecat-ai pipecat-services-deepgram pipecat-services-elevenlabs twilio python-dotenv

Verify the install:

python -c "import pipecat; print(pipecat.__version__)"

You should see a version string like 0.0.52 or newer.

Step 2: Define the pipeline topology

Pipecat pipelines are directed graphs of frames. For a phone agent, the flow is:

Twilio audio → Deepgram STT → LLM → ElevenLabs TTS → Twilio audio

Create pipeline.py with the core assembly:

# pipeline.py
import os
from dotenv import load_dotenv
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.twilio import TwilioTransport
from pipecat.frames.frames import Frame, AudioRawFrame, TextFrame
from pipecat.processors.frame_processor import FrameProcessor

load_dotenv()

class TurnDetector(FrameProcessor):
    """Simple VAD-based turn detection for demo purposes."""
    def __init__(self):
        super().__init__()
        self.buffer = bytearray()
        self.silence_threshold = 500  # ms
        self.last_audio_ms = 0

    async def process_frame(self, frame: Frame, direction):
        if isinstance(frame, AudioRawFrame):
            self.buffer.extend(frame.audio)
            self.last_audio_ms = frame.pts
        await self.push_frame(frame, direction)

The TurnDetector is a placeholder. Production systems use Deepgram’s built-in endpointing or a dedicated VAD service. For this tutorial, we rely on Deepgram’s interim_results and endpointing parameters.

Step 3: Wire the services

Configure each service with the parameters that matter for latency and quality.

# services.py
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService

def build_stt():
    return DeepgramSTTService(
        api_key=os.getenv("DEEPGRAM_API_KEY"),
        model="nova-2",
        language="en-US",
        interim_results=True,
        endpointing=300,  # ms of silence before finalizing
        punctuate=True,
        smart_format=True,
    )

def build_tts():
    return ElevenLabsTTSService(
        api_key=os.getenv("ELEVENLABS_API_KEY"),
        voice_id="21m00Tcm4TlvDq8ikWAM",  # Rachel
        model_id="eleven_turbo_v2_5",      # lowest latency
        output_format="ulaw_8000",         # Twilio expects μ-law 8kHz
        optimize_streaming_latency=4,      # max optimization
    )

def build_llm():
    return OpenAILLMService(
        api_key=os.getenv("OPENAI_API_KEY"),
        model="gpt-4o-mini",
        temperature=0.7,
        system_prompt=(
            "You are a helpful phone assistant. Keep responses under two sentences. "
            "Speak naturally, like you're talking to a friend. Never use markdown."
        ),
    )

Key choices:

• Deepgram Nova-2 balances accuracy and speed for telephony audio.
• ElevenLabs Turbo v2.5 with optimize_streaming_latency=4 streams audio before the full response generates.
• μ-law 8kHz matches Twilio’s native codec, avoiding transcoding delay.
• GPT-4o-mini provides strong reasoning at lower cost and latency than GPT-4o.

Step 4: Connect Twilio transport

Pipecat’s TwilioTransport handles the WebSocket connection to Twilio’s media streams. It expects a publicly reachable HTTPS endpoint.

# transport.py
from pipecat.transports.twilio import TwilioTransport
from pipecat.pipeline.task import PipelineTask

def build_transport(task: PipelineTask):
    return TwilioTransport(
        account_sid=os.getenv("TWILIO_ACCOUNT_SID"),
        auth_token=os.getenv("TWILIO_AUTH_TOKEN"),
        phone_number=os.getenv("TWILIO_PHONE_NUMBER"),
        task=task,
        sample_rate=8000,
        channels=1,
    )

The transport registers a /webhook route on your server. Twilio will POST call events there. You need a tunnel for local development:

ngrok http 8000

Copy the HTTPS URL (e.g., https://abc123.ngrok.io) and configure it in the Twilio console under Phone Numbers → Manage → Active Numbers → Voice Configuration → A Call Comes In as https://abc123.ngrok.io/webhook.

Step 5: Assemble and run the pipeline

# main.py
import asyncio
import os
from dotenv import load_dotenv
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext

from services import build_stt, build_tts, build_llm
from transport import build_transport

load_dotenv()

async def main():
    stt = build_stt()
    llm = build_llm()
    tts = build_tts()

    context = OpenAILLMContext()
    context.set_messages([
        {"role": "system", "content": llm.system_prompt},
    ])

    pipeline = Pipeline([
        stt,
        llm,
        tts,
    ])

    task = PipelineTask(pipeline)
    transport = build_transport(task)

    runner = PipelineRunner()
    await runner.run(task)

if __name__ == "__main__":
    asyncio.run(main())

Run it:

python main.py

You should see:

[INFO] Pipecat pipeline started
[INFO] Twilio transport listening on port 8000

Step 6: Handle the first interaction

The pipeline as written streams audio but doesn’t initiate conversation. Add a greeting when the call connects. Modify main.py to inject a TextFrame into the LLM context on on_call_started:

# main.py (add to PipelineTask initialization)
task = PipelineTask(
    pipeline,
    params={
        "on_call_started": lambda transport, call_sid: asyncio.create_task(
            greet_caller(transport, call_sid)
        ),
    },
)

async def greet_caller(transport, call_sid):
    await asyncio.sleep(0.5)  # let media stream establish
    greeting = "Hi, this is your AI assistant. How can I help you?"
    await transport.send_text(call_sid, greeting)

send_text injects a TextFrame directly into the TTS service, bypassing STT and LLM for the initial prompt.

Step 7: Add interruption handling

Real conversations need barge-in. Pipecat supports this via the interruptible flag on TTS output. Update services.py:

def build_tts():
    return ElevenLabsTTSService(
        api_key=os.getenv("ELEVENLABS_API_KEY"),
        voice_id="21m00Tcm4TlvDq8ikWAM",
        model_id="eleven_turbo_v2_5",
        output_format="ulaw_8000",
        optimize_streaming_latency=4,
        interruptible=True,  # enable barge-in
    )

When the user speaks during TTS playback, Deepgram emits an interim transcript. The pipeline cancels the current TTS stream and feeds the new transcript to the LLM. This works out of the box with interruptible=True because Pipecat’s PipelineTask monitors upstream audio frames.

Step 8: Deploy to a real server

Local tunnels are fine for testing. For production, deploy to a container platform with a static IP and TLS. A minimal Dockerfile:

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["python", "main.py"]

requirements.txt:

pipecat-ai==0.0.52
pipecat-services-deepgram==0.0.12
pipecat-services-elevenlabs==0.0.8
twilio==9.0.0
python-dotenv==1.0.1

Build and push:

docker build -t gcr.io/my-project/phone-agent .
docker push gcr.io/my-project/phone-agent

Deploy to Cloud Run, Fly.io, or any service that provides HTTPS. Update the Twilio webhook URL to the production endpoint.

Step 9: Verify end-to-end

Call your Twilio number. You should hear the greeting within 2-3 seconds of answer. Speak a query — “What’s the weather in Seattle?” — and expect a response under 1.5 seconds end-to-end.

Check logs for these markers:

[INFO] Call started: CAxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
[INFO] STT connected: nova-2
[INFO] LLM request: 127 tokens
[INFO] TTS streaming: 1.2s first byte
[INFO] Call ended: CAxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

If latency exceeds 2 seconds, inspect:

• STT latency: Deepgram dashboard shows processing_time_ms. Target < 300ms.
• LLM latency: Log time_to_first_token. GPT-4o-mini should be < 500ms.
• TTS latency: ElevenLabs optimize_streaming_latency=4 yields ~200ms first byte.
• Network: Run the server in the same region as your Twilio edge (us1, ie1, de1, sg1, jp1, au1, br1).

Step 10: Add observability

Production agents need metrics. Pipecat emits structured logs; pipe them to your observability stack. Add a simple middleware:

# observability.py
import time
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.frames.frames import Frame

class LatencyTracker(FrameProcessor):
    def __init__(self, name: str):
        super().__init__()
        self.name = name
        self.start_times = {}

    async def process_frame(self, frame: Frame, direction):
        if hasattr(frame, "pts") and frame.pts not in self.start_times:
            self.start_times[frame.pts] = time.perf_counter()
        elif hasattr(frame, "pts") and frame.pts in self.start_times:
            latency_ms = (time.perf_counter() - self.start_times.pop(frame.pts)) * 1000
            print(f"[{self.name}] Frame latency: {latency_ms:.1f}ms")
        await self.push_frame(frame, direction)

Insert trackers between each stage:

pipeline = Pipeline([
    stt,
    LatencyTracker("stt->llm"),
    llm,
    LatencyTracker("llm->tts"),
    tts,
])

You now have per-frame latency visibility without external dependencies.

Step 11: Handle failures gracefully

Network blips happen. Wrap the runner in a restart loop and add Twilio call status callbacks:

# main.py (replace asyncio.run(main()))
async def run_with_restart():
    while True:
        try:
            await main()
        except Exception as e:
            print(f"Pipeline crashed: {e}. Restarting in 5s...")
            await asyncio.sleep(5)

asyncio.run(run_with_restart())

In the Twilio console, set Status Callback URL to https://your-domain.com/status and handle completed, failed, busy, no-answer to clean up resources.

Step 12: Extend with function calling

The LLM can invoke tools. Define a function schema and handler:

# tools.py
from pipecat.services.openai import OpenAILLMService

WEATHER_TOOL = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City, state or ZIP"},
            },
            "required": ["location"],
        },
    },
}

async def handle_get_weather(args: dict) -> str:
    # Replace with real API call
    return f"It's 72°F and sunny in {args['location']}."

llm = OpenAILLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    model="gpt-4o-mini",
    temperature=0.7,
    system_prompt=(
        "You are a helpful phone assistant. Keep responses under two sentences. "
        "Speak naturally. Never use markdown. Use get_weather for weather questions."
    ),
    tools=[WEATHER_TOOL],
    tool_handler=handle_get_weather,
)

Pipecat executes the handler and feeds the result back to the LLM automatically. The caller hears a seamless response.

Verification checklist

• Call connects and greeting plays within 3 seconds
• STT transcribes speech accurately (check Deepgram dashboard)
• LLM responds relevantly (check logs for prompt/completion)
• TTS audio is clear, no artifacts (μ-law 8kHz matches Twilio)
• Interruption works: speak during agent response, agent stops and listens
• Function calling executes and result is spoken
• Call cleanup on hangup (no orphaned processes)
• Latency p50 < 1.5s end-to-end (track with LatencyTracker)

Next steps

• Replace the placeholder VAD with Silero VAD or Deepgram’s native endpointing for tighter turn detection.
• Add a persistent conversation store (Redis + vector index) for multi-call context.
• Implement per-user voice cloning via ElevenLabs voice_id lookup.
• Load-test with locust or k6 simulating 100 concurrent calls; monitor CPU/memory on your container.

The architecture scales horizontally: each call gets its own pipeline instance. Deploy behind a load balancer with sticky sessions on call_sid if you add in-memory state. For stateless pipelines, any instance can handle any call.