Integrate Telnyx with AssemblyAI

Telnyx is a global connectivity platform that provides programmable voice, messaging, and wireless services. By combining Telnyx with AssemblyAI, you can build real-time voice agents with industry-leading speech recognition accuracy and advanced turn detection.

This guide shows you how to integrate Telnyx with AssemblyAI using two popular voice agent orchestrators: LiveKit and Pipecat.

LiveKit Telephony Integration

LiveKit is a real-time communication platform for building voice, video, and data applications. You can integrate Telnyx SIP trunking with LiveKit to enable phone calls to your voice agents that use AssemblyAI for speech recognition.

How it works

Telnyx SIP trunks bridge phone calls into LiveKit rooms as special SIP participants. Your LiveKit agent (configured with AssemblyAI STT) connects to the room and interacts with the caller. The flow is:

1. Phone call → Telnyx SIP trunk
2. Telnyx → LiveKit room (creates SIP participant)
3. LiveKit agent (with AssemblyAI STT) → joins room and handles conversation

Before you begin

• Telnyx account: Create an account and purchase a phone number at telnyx.com
• LiveKit project: Get your SIP URI from your LiveKit project settings
• LiveKit CLI: Install the LiveKit CLI on your computer
• Environment variables: Configure LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET, and ASSEMBLYAI_API_KEY

Step 1: Configure Telnyx SIP connection

Configure your Telnyx SIP connection to route calls to LiveKit. Follow the detailed setup guide:

Telnyx LiveKit SIP Configuration Guide

Key steps:

1. Create a SIP connection in Telnyx Mission Control Portal
2. Set connection type to FQDN and provide your LiveKit SIP URI
3. Configure outbound call authentication (username/password)
4. Assign your phone number(s) to the SIP connection

Step 2: Configure LiveKit SIP trunks

Create inbound and outbound SIP trunks in LiveKit using the LiveKit CLI.

Inbound trunk

Create inboundTrunk.json:

1
{
2
  "trunk": {
3
    "name": "Telnyx Inbound Trunk",
4
    "numbers": ["YOUR_TELNYX_NUMBER"]
5
  }
6
}

Create the trunk:

$
lk sip inbound create inboundTrunk.json

Save the returned SIPTrunkID for the next step.

Dispatch rule

Create dispatchRule.json to route incoming calls to your agent:

1
{
2
  "name": "Agent Dispatch Rule",
3
  "trunk_ids": ["YOUR_TRUNK_ID"],
4
  "rule": {
5
    "dispatchRuleIndividual": {
6
      "roomPrefix": "call-"
7
    }
8
  },
9
  "roomConfig": {
10
    "agents": [{
11
      "agentName": "my-telephony-agent"
12
    }]
13
  }
14
}

Create the dispatch rule:

$
lk sip dispatch create dispatchRule.json

This automatically dispatches your agent to each incoming call in a new room.

Outbound trunk (optional)

For outbound calling, create outboundTrunk.json:

1
{
2
  "trunk": {
3
    "name": "Telnyx Outbound Trunk",
4
    "address": "sip.telnyx.com",
5
    "numbers": ["YOUR_TELNYX_NUMBER"],
6
    "auth_username": "YOUR_OUTBOUND_USER",
7
    "auth_password": "YOUR_OUTBOUND_PASS"
8
  }
9
}

Create the trunk:

$
lk sip outbound create outboundTrunk.json

Step 3: Build your LiveKit agent with AssemblyAI

Create a LiveKit agent that uses AssemblyAI for speech recognition. Once your SIP trunks and dispatch rules are configured, no special telephony code is required—the agent simply joins the room when a call comes in.

Prerequisites

Setup and activate a virtual environment:

$
python -m venv venv
>
source venv/bin/activate

Install the required dependencies:

$
pip install livekit-agents livekit-plugins-assemblyai livekit-plugins-openai livekit-plugins-silero livekit-plugins-rime

Download model files:

$
python agent.py download-files

Agent code

Here’s a minimal agent using AssemblyAI STT:

1
import asyncio
2
from dotenv import load_dotenv
3
from livekit import rtc
4
from livekit.agents import AutoSubscribe, JobContext, WorkerOptions, cli, llm
5
from livekit.agents.voice_assistant import VoiceAssistant
6
from livekit.plugins import assemblyai, openai, silero, rime
7
8
load_dotenv()
9
10
async def entrypoint(ctx: JobContext):
11
    initial_ctx = llm.ChatContext().append(
12
        role="system",
13
        text=(
14
            "You are a helpful voice assistant. Your interface with users will be voice. "
15
            "Use short and concise responses, avoiding unpronounceable punctuation."
16
        ),
17
    )
18
19
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
20
21
    assistant = VoiceAssistant(
22
        vad=silero.VAD.load(),
23
        stt=assemblyai.STT(),
24
        llm=openai.LLM(),
25
        tts=rime.TTS(),
26
        chat_ctx=initial_ctx,
27
    )
28
29
    assistant.start(ctx.room)
30
31
    await asyncio.sleep(1)
32
    await assistant.say("Hello! How can I help you today?", allow_interruptions=True)
33
34
35
if __name__ == "__main__":
36
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

Environment variables

Set the following environment variables:

$
export LIVEKIT_URL=<your LiveKit server URL>
>
export LIVEKIT_API_KEY=<your API Key>
>
export LIVEKIT_API_SECRET=<your API Secret>
>
export ASSEMBLYAI_API_KEY=<your AssemblyAI API key>
>
export OPENAI_API_KEY=<your OpenAI API key>
>
export RIME_API_KEY=<your Rime API key>

Run the agent

Start your agent:

$
python agent.py dev

Now dial your Telnyx phone number to test the integration. The call will be routed through Telnyx → LiveKit → your agent with AssemblyAI speech recognition.

Testing outbound calls

To test an outbound call, create sipParticipant.json:

1
{
2
  "sip_trunk_id": "YOUR_OUTBOUND_TRUNK_ID",
3
  "sip_call_to": "+15105551234",
4
  "room_name": "test-outbound-call",
5
  "participant_identity": "outbound-test",
6
  "participant_name": "Test Call"
7
}

Place the call:

$
lk sip participant create sipParticipant.json

Troubleshooting

Call connects but agent never speaks: Verify your dispatch rule includes roomConfig.agents with the correct agentName. Without this, the agent won’t be automatically dispatched to the room.

Audio quality issues: Check your Telnyx SIP connection settings and ensure proper codec configuration. See the Telnyx troubleshooting guide.

Pipecat Integration

Pipecat is an open-source framework for building voice and multimodal conversational AI agents. You can integrate Telnyx Media Streaming with Pipecat to enable phone calls to your voice agents that use AssemblyAI for speech recognition.

This guide covers both dial-in (users call your number) and dial-out (your bot calls users) functionality.

Prerequisites

• Telnyx account: Create an account and purchase phone numbers at telnyx.com
• Public server or tunnel: For dial-out, you’ll need a public-facing server or tunneling service like ngrok
• API keys: Get API keys for AssemblyAI, OpenAI (or other LLM), and your preferred TTS service

Install the required dependencies:

$
pip install pipecat-ai python-dotenv loguru

Set up your environment variables:

$
TELNYX_API_KEY=your_telnyx_api_key
>
ASSEMBLYAI_API_KEY=your_assemblyai_api_key
>
OPENAI_API_KEY=your_openai_api_key
>
DEEPGRAM_API_KEY=your_deepgram_api_key  # or other TTS provider

How it works

Dial-in flow:

1. User calls your Telnyx number
2. Telnyx executes your TeXML application which establishes a WebSocket connection
3. Telnyx opens a WebSocket to your server with real-time audio and call metadata
4. Your bot processes the audio using the Pipecat pipeline with AssemblyAI STT
5. The bot responds with audio sent back to Telnyx over WebSocket
6. Telnyx plays the audio to the caller in real-time

Dial-out flow:

1. Your application triggers a dial-out via API
2. Server initiates a Telnyx call using the Call Control API
3. Telnyx establishes the call and opens a WebSocket connection
4. Your bot joins the WebSocket and sets up the pipeline
5. The recipient answers and is connected to your bot
6. The bot handles the conversation with real-time audio streaming

Dial-in Setup

Step 1: Create TeXML application

Telnyx uses TeXML (Telnyx Extensible Markup Language) to control call flow. Create a TeXML application that establishes a WebSocket connection to your bot:

1
<?xml version="1.0" encoding="UTF-8"?>
2
<Response>
3
  <Connect>
4
    <Stream url="wss://your-server.com/ws" bidirectionalMode="rtp"></Stream>
5
  </Connect>
6
  <Pause length="40"/>
7
</Response>

The bidirectionalMode="rtp" parameter enables real-time audio streaming in both directions.

Custom data with query parameters: You can pass custom data to your bot by adding query parameters to the WebSocket URL:

1
<?xml version="1.0" encoding="UTF-8"?>
2
<Response>
3
  <Connect>
4
    <Stream url="wss://your-server.com/ws?user_id=12345&session_type=support" bidirectionalMode="rtp"></Stream>
5
  </Connect>
6
  <Pause length="40"/>
7
</Response>

Step 2: Configure Telnyx phone number

1. Go to the Telnyx Portal
2. Navigate to Voice → Programmable Voice → TeXML Applications
3. Create a new TeXML Application with your WebSocket URL
4. Assign the TeXML Application to your phone number

Step 3: Build your bot

Here’s a complete dial-in bot using AssemblyAI for speech recognition:

1
import os
2
from dotenv import load_dotenv
3
from loguru import logger
4
5
from pipecat.audio.vad.silero import SileroVADAnalyzer
6
from pipecat.frames.frames import EndFrame
7
from pipecat.pipeline.pipeline import Pipeline
8
from pipecat.pipeline.runner import PipelineRunner
9
from pipecat.pipeline.task import PipelineParams, PipelineTask
10
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
11
from pipecat.runner.utils import parse_telephony_websocket
12
from pipecat.serializers.telnyx import TelnyxFrameSerializer
13
from pipecat.services.assemblyai.stt import AssemblyAISTTService
14
from pipecat.services.openai.llm import OpenAILLMService
15
from pipecat.services.deepgram.tts import DeepgramTTSService
16
from pipecat.transports.websocket.fastapi import (
17
    FastAPIWebsocketTransport,
18
    FastAPIWebsocketParams,
19
)
20
21
load_dotenv()
22
23
async def run_bot(websocket):
24
    """Run the voice agent bot with AssemblyAI STT."""
25
    
26
    # Parse Telnyx WebSocket data - automatically extracts call information
27
    transport_type, call_data = await parse_telephony_websocket(websocket)
28
    
29
    # Extract call information (automatically provided by Telnyx)
30
    stream_id = call_data["stream_id"]
31
    call_control_id = call_data["call_control_id"]
32
    outbound_encoding = call_data["outbound_encoding"]
33
    from_number = call_data["from"]  # Caller's number
34
    to_number = call_data["to"]      # Your Telnyx number
35
    
36
    logger.info(f"Incoming call from {from_number} to {to_number}")
37
    
38
    # Create Telnyx serializer with call details
39
    serializer = TelnyxFrameSerializer(
40
        stream_id=stream_id,
41
        call_control_id=call_control_id,
42
        api_key=os.getenv("TELNYX_API_KEY"),
43
    )
44
    
45
    # Configure WebSocket transport
46
    transport = FastAPIWebsocketTransport(
47
        websocket=websocket,
48
        params=FastAPIWebsocketParams(
49
            audio_in_enabled=True,
50
            audio_out_enabled=True,
51
            add_wav_header=False,
52
            vad_analyzer=SileroVADAnalyzer(),
53
            serializer=serializer,
54
        ),
55
    )
56
    
57
    # Configure AI services
58
    stt = AssemblyAISTTService(api_key=os.getenv("ASSEMBLYAI_API_KEY"))
59
    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini")
60
    tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"))
61
    
62
    # Customize bot behavior based on call information
63
    messages = [
64
        {
65
            "role": "system",
66
            "content": f"You are a helpful voice assistant. The caller is calling from {from_number}. Keep responses concise and conversational."
67
        }
68
    ]
69
    
70
    context = OpenAILLMContext(messages)
71
    context_aggregator = llm.create_context_aggregator(context)
72
    
73
    # Build pipeline
74
    pipeline = Pipeline([
75
        transport.input(),
76
        stt,
77
        context_aggregator.user(),
78
        llm,
79
        tts,
80
        transport.output(),
81
        context_aggregator.assistant(),
82
    ])
83
    
84
    task = PipelineTask(
85
        pipeline,
86
        params=PipelineParams(
87
            allow_interruptions=True,
88
            audio_in_sample_rate=8000,
89
            audio_out_sample_rate=8000,
90
        ),
91
    )
92
    
93
    @transport.event_handler("on_client_disconnected")
94
    async def on_client_disconnected(transport, client):
95
        logger.info("Call ended")
96
        await task.queue_frame(EndFrame())
97
    
98
    # Run the pipeline
99
    runner = PipelineRunner()
100
    await runner.run(task)

See the complete dial-in example for full implementation details including server setup.

Dial-out Setup

Dial-out allows your bot to initiate calls to phone numbers using Telnyx’s outbound calling capabilities.

How dial-out works

1. Your application triggers a dial-out (via API call or user action)
2. Server initiates a Telnyx call using the Call Control API
3. Telnyx establishes the call and opens a WebSocket connection
4. Your bot joins the WebSocket and sets up the pipeline
5. The recipient answers and is connected to your bot
6. The bot handles the conversation with real-time audio streaming

Bot configuration for dial-out

The dial-out bot configuration is similar to dial-in. Telnyx automatically provides call information in the WebSocket messages:

1
# Parse WebSocket data (same as dial-in)
2
transport_type, call_data = await parse_telephony_websocket(websocket)
3
4
# Extract call information
5
stream_id = call_data["stream_id"]
6
call_control_id = call_data["call_control_id"]
7
from_number = call_data["from"]  # Your Telnyx number
8
to_number = call_data["to"]      # Target number you're calling
9
10
# Customize bot behavior for outbound calls
11
greeting = f"Hi! This is an automated call from {from_number}. How are you today?"

The transport and pipeline configuration are identical to dial-in. See the complete dial-out example for full server implementation with outbound call creation.

Key Features

Audio format: Telnyx Media Streaming uses 8kHz mono audio with 16-bit PCM encoding. Configure your pipeline accordingly:

1
task = PipelineTask(
2
    pipeline,
3
    params=PipelineParams(
4
        audio_in_sample_rate=8000,
5
        audio_out_sample_rate=8000,
6
        allow_interruptions=True,
7
    ),
8
)

Automatic call termination: When you provide Telnyx API credentials to the TelnyxFrameSerializer, it automatically ends calls when your pipeline ends:

1
serializer = TelnyxFrameSerializer(
2
    stream_id=stream_id,
3
    call_control_id=call_control_id,
4
    api_key=os.getenv("TELNYX_API_KEY"),  # Enables auto-termination
5
)

Built-in call information: Unlike other providers, Telnyx automatically includes caller information (to/from numbers) in the WebSocket messages, eliminating the need for custom webhook servers in basic dial-in scenarios.

Advanced Configuration

For production use cases, you can customize AssemblyAI’s turn detection and add keyterms for improved accuracy:

1
from pipecat.services.assemblyai.stt import AssemblyAISTTService, AssemblyAIConnectionParams
2
3
# Configure AssemblyAI with custom parameters
4
stt_params = AssemblyAIConnectionParams(
5
    sample_rate=8000,
6
    end_of_turn_confidence_threshold=0.4,
7
    min_end_of_turn_silence_when_confident=400,
8
    max_turn_silence=1280,
9
    keyterms_prompt=["NPI", "TIN", "CMS", "PTAN", "CPT", "CDT", "DOB", "SSN"]
10
)
11
12
stt = AssemblyAISTTService(
13
    api_key=os.getenv("ASSEMBLYAI_API_KEY"),
14
    connection_params=stt_params,
15
)

Turn detection options: AssemblyAI has built-in VAD and turn detection. You can either:

• Use AssemblyAI’s turn detection (recommended for best accuracy)
• Use Silero VAD by including vad_analyzer=SileroVADAnalyzer() in the transport params

Keyterms: Add domain-specific terms to improve recognition accuracy for specialized vocabulary.

Resources

• LiveKit Documentation
• LiveKit Telephony Quickstart
• Telnyx SIP Trunk Configuration Guide
• Telnyx LiveKit SIP Setup
• Pipecat Documentation
• Pipecat Telnyx WebSocket Guide
• Pipecat Telnyx Examples