{
  "sip": {
    "admin": true,
    "call": true
  }
}

Authorization: Bearer <your-token>

'sip_status': 'SIP_STATUS_FORBIDDEN'

TransferSIPParticipantRequest

play_dialtone

bool play_dialtone = 4;

PlayDialtone bool

play_dialtone: bool

playDialtone?: boolean

play_dialtone: nil

playDialtone

play_dialtone

participant_identity = await livekit_manager.create_sip_participant(
    phone_number=transfer_to_phone_number,
    room_name=room_name,
    user_id=user_id_for_trunk,
    is_outbound=True,  # Transfer is always outbound
    play_dialtone=True  # Optional: Play dial tone
)

participant_identity = await livekit_manager.create_sip_participant(
    phone_number=transfer_to_phone_number,
    room_name=room_name,
    user_id=user_id_for_trunk,
    is_outbound=True,  # Transfer is always outbound
    play_dialtone=True  # Optional: Play dial tone
)

play_dialtone

TransferSIPParticipant

twilio api trunking v1 trunks update --sid <twilio-trunk-sid> \
--transfer-mode enable-all \
--transfer-caller-id from-transferee

TransferSIPParticipant

tel:+15105550100

sip:+15105550100@host

False

import { SipClient } from 'livekit-server-sdk';

async function transferParticipant(participant) {
  const sipTransferOptions = { playDialtone: false };
  const sipClient = new SipClient(process.env.LIVEKIT_URL, process.env.LIVEKIT_API_KEY, process.env.LIVEKIT_API_SECRET);
  const transferTo = "tel:+15105550100";
  await sipClient.transferSipParticipant('open-room', participant.identity, transferTo, sipTransferOptions);
}

LocalParticipant

RemoteParticipant

sid

identity

metadata

+15105550100

+

1234567890

+11234567890

(123) 456 7890

+1234567890

123

+123

+

user

destination_country

webhook

participant_joined

listRooms

roomList

lk token create \
--api-key $LIVEKIT_API_KEY \
--api-secret $LIVEKIT_SECRET_KEY \
--list \
--valid-for 24h

limit

page

start

YYYY-MM-DD

end

lk agent logs --log-type=build

setLogExtension()

max_call_duration

session_duration_ms

/metrics

identity

autoSubscribe

// Node.js Example
import { defineAgent, voice } from '@livekit/agents';
import * as deepgram from '@livekit/agents-plugin-deepgram';
import * as openai from '@livekit/agents-plugin-openai';

export default defineAgent({
  entry: async (ctx) => {
    await ctx.connect();
    const session = new voice.AgentSession({
      stt: new deepgram.STT(),
      llm: new openai.LLM(),
    });
    await session.start({
      agent: new voice.Agent({ instructions: "You are a helpful assistant." }),
      room: ctx.room,
    });
  },
});

participant identity

room SID

from fastapi import FastAPI
from livekit.api import AccessToken, VideoGrants

app = FastAPI()

@app.post("/start-agent")
async def start_agent():
    token = (
        AccessToken(api_key="YOUR_API_KEY", api_secret="YOUR_API_SECRET")
        .with_grants(VideoGrants(room_join=True, room="my-room"))
        .to_jwt()
    )
    # Call external avatar API or dispatch agent worker
    return {"livekit_token": token, "room": "my-room"}

pip install livekit-plugins-deepgram

DEEPGRAM_API_KEY=your_deepgram_api_key

from livekit.agents import tts
from livekit.plugins import deepgram

# Initialize Deepgram TTS
tts_plugin = deepgram.TTS(
    model="aura-2-andromeda-en",
    encoding="linear16",
    sample_rate=24000
)

# Use in an AgentSession
session = voice.AgentSession(
    stt=deepgram.STT(),  # optional, if you want to use STT as well
    llm=openai.LLM(),    # your LLM
    tts=tts_plugin
)

stream = tts_plugin.stream()
await stream.input_text("Hello, this is a test.")
await stream.flush()  # ends the segment

config.yaml

port

port: 7880  # Change this to your desired port

rtc:
  port_range_start: 50000
  port_range_end: 60000

rtc:
  tcp_port: 7881  # Change to desired port

rtc:
  udp_port: 7882

values.yaml

livekit:
  port: 7880
  rtc:
    tcp_port: 7881
    port_range_start: 50000
    port_range_end: 60000

port

WorkerOptions

from livekit.agents import WorkerOptions, cli

def entrypoint():
    # Your agent logic here
    pass

if __name__ == "__main__":
    worker_options = WorkerOptions(
        entrypoint_fnc=entrypoint,
        port=9091  # Set custom port here
    )
    cli.run_app(worker_options)

RoomInputOptions.close_on_disconnect=False

CLIENT_INITIATED

Room.disconnect()

RoomInputOptions.close_on_disconnect

True

close_on_disconnect=False

CLIENT_INITIATED

disconnect()

close_on_disconnect

ctx.connect()

ctx.connect()

AgentSession

AgentSession

ctx.connect()

BVCTelephony

LIVEKIT_URL

LIVEKIT_API_KEY

LIVEKIT_API_SECRET

AZURE_OPENAI_ENDPOINT

AZURE_OPENAI_DEPLOYMENT

AZURE_OPENAI_API_KEY

ctx.connect()

silero.VAD.load()

entrypoint

_SegmentSynchronizerImpl.resume called after close

resume()

resume()

close

drain

_SegmentSynchronizerImpl.resume called after close

resume()

BVCTelephony

.env.local

AZURE_OPENAI_ENDPOINT

AZURE_OPENAI_DEPLOYMENT

AZURE_OPENAI_API_KEY

LIVEKIT_URL

LIVEKIT_API_KEY

LIVEKIT_API_SECRET

silero.VAD.load()

ctx.connect()

ctx.wait_for_participant()

_SegmentSynchronizerImpl.resume called after close

resume()

LIVEKIT_URL

LIVEKIT_API_KEY

LIVEKIT_API_SECRET

AZURE_OPENAI_ENDPOINT

.env.local

silero.VAD.load()

ctx.connect()

ctx.wait_for_participant()

_SegmentSynchronizerImpl.resume called after close

from pydantic import BaseModel, Field
from typing import Literal
import time

class CustomMetrics(BaseModel):
    type: Literal["custom_metrics"] = "custom_metrics"
    label: str
    timestamp: float
    custom_value: float

class CustomMetricsCollectedEvent(BaseModel):
    type: Literal["custom_metrics_collected"] = "custom_metrics_collected"
    metrics: CustomMetrics
    created_at: float = Field(default_factory=time.time)

session.emit()

custom_metric = CustomMetrics(
    label="api_response_time",
    timestamp=time.time(),
    custom_value=0.45  # e.g., response time in seconds
)

session.emit("custom_metrics_collected", CustomMetricsCollectedEvent(metrics=custom_metric))

@session.on("custom_metrics_collected")
def _on_custom_metrics_collected(ev: CustomMetricsCollectedEvent):
    print(f"Received custom metric: {ev.metrics.label} = {ev.metrics.custom_value}")

BVCTelephony

from dotenv import load_dotenv
import os
import time
from livekit import agents
from livekit.agents import AgentSession, Agent, RoomInputOptions
from livekit.plugins import deepgram, openai, silero, noise_cancellation
from pydantic import BaseModel, Field
from typing import Literal

# Load environment variables
load_dotenv(".env.local")

azure_endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
azure_deployment = os.getenv("AZURE_OPENAI_DEPLOYMENT")
openai_api_key = os.getenv("AZURE_OPENAI_API_KEY")

# Define custom metrics
class CustomMetrics(BaseModel):
    type: Literal["custom_metrics"] = "custom_metrics"
    label: str
    timestamp: float
    custom_value: float

class CustomMetricsCollectedEvent(BaseModel):
    type: Literal["custom_metrics_collected"] = "custom_metrics_collected"
    metrics: CustomMetrics
    created_at: float = Field(default_factory=time.time)

# Define the Assistant class
class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a helpful voice AI assistant.
            You eagerly assist users with their questions by providing information from your extensive knowledge.
            Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.
            You are curious, friendly, and have a sense of humor.""",
        )

# Define the entrypoint
async def entrypoint(ctx: agents.JobContext):
    # Connect to the LiveKit room before interacting with participants
    await ctx.connect()

    # Wait for at least one participant to join
    participant = await ctx.wait_for_participant()
    try:
        import json
        meta = json.loads(participant.metadata) if participant.metadata else {}
    except Exception:
        meta = participant.metadata
    print("[agent] participant", participant.identity, "metadata:", meta)

    # Initialize the session
    session = AgentSession(
        stt=deepgram.STT(model="nova-2"),
        llm=openai.LLM(
            api_key=openai_api_key,
        ).with_azure(
            model="gpt-4o-mini",
            azure_endpoint=azure_endpoint,
            azure_deployment=azure_deployment,
            api_version="2024-08-01-preview"
        ),
        tts=deepgram.TTS(model="aura-asteria-en"),
        vad=silero.VAD.load(),
    )

    # Emit custom metrics
    custom_metric = CustomMetrics(
        label="session_start",
        timestamp=time.time(),
        custom_value=1.0  # Example metric value
    )
    session.emit("custom_metrics_collected", CustomMetricsCollectedEvent(metrics=custom_metric))

    # Register a listener for custom metrics
    @session.on("custom_metrics_collected")
    def _on_custom_metrics_collected(ev: CustomMetricsCollectedEvent):
        print(f"Received custom metric: {ev.metrics.label} = {ev.metrics.custom_value}")

    # Start the session
    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_input_options=RoomInputOptions(
            noise_cancellation=noise_cancellation.BVC(),
        ),
    )

    # Generate a reply
    await session.generate_reply(
        instructions="Greet the user and offer your assistance."
    )

if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint, port=9091))

session.emit()

livekit.agents.llm

chat()

ChatContext

LLM.chat()

# Basic Agent Setup
agent = Agent(
    instructions="You are a friendly voice assistant",
    llm="openai/gpt-4o-mini",
    tools=[lookup_weather],
)

max_endpointing_delay

min_remaining_audio

max_text_length

tts=cartesia.TTS(
    text_pacing=tts.SentenceStreamPacer(
        min_remaining_audio=3.0,  # Reduce from default 5.0
        max_text_length=200,      # Reduce from default 300
    )
)

await agent.say("Let me look that up for you...")

use_tts_aligned_transcript=True

AgentSession

async def tts_node(self, text: AsyncIterable[str], model_settings: ModelSettings):
    MAX_CHUNKS = 20
    chunk_count = 0
    async def process_text():
        nonlocal chunk_count
        async for chunk in text:
            chunk_count += 1
            if chunk_count > MAX_CHUNKS:
                self.session.interrupt()
                self.session.say("I’m sorry, that will take too long to say.")
                break
            yield chunk
    return Agent.default.tts_node(self, process_text(), model_settings)

session.say()

add_to_chat_ctx=False

True

# Greet the user
await session.say("Hello, how can I help you today?")

BVCTelephony

from dotenv import load_dotenv
import os
import logging
from livekit import agents
from livekit.agents import AgentSession, Agent, RoomInputOptions, metrics, MetricsCollectedEvent
from livekit.plugins import deepgram, openai, silero, noise_cancellation

# Load environment variables
load_dotenv(".env.local")

azure_endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
azure_deployment = os.getenv("AZURE_OPENAI_DEPLOYMENT")
openai_api_key = os.getenv("AZURE_OPENAI_API_KEY")

# Preload models for faster response times
stt = deepgram.STT(model="nova-2")
llm = openai.LLM(
    api_key=openai_api_key,
).with_azure(
    model="gpt-4o-mini",
    azure_endpoint=azure_endpoint,
    azure_deployment=azure_deployment,
    api_version="2024-08-01-preview"
)
tts = deepgram.TTS(model="aura-asteria-en")
vad = silero.VAD.load()
noise_cancellation_filter = noise_cancellation.BVC()

# Set up logging and a global metrics collector
logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
logger = logging.getLogger(__name__)
usage_collector = metrics.UsageCollector()

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a helpful voice AI assistant.
            You eagerly assist users with their questions by providing information from your extensive knowledge.
            Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.
            You are curious, friendly, and have a sense of humor.""",
        )

async def entrypoint(ctx: agents.JobContext):
    # Connect to the LiveKit room before interacting with participants
    await ctx.connect()

    # Wait for at least one participant to join
    participant = await ctx.wait_for_participant()
    try:
        import json
        meta = json.loads(participant.metadata) if participant.metadata else {}
    except Exception:
        meta = participant.metadata
    <http://logger.info|logger.info>(f"[agent] participant {participant.identity} metadata: {meta}")

    # Initialize the session
    session = AgentSession(
        stt=stt,
        llm=llm,
        tts=tts,
        vad=vad,
    )

    # Add logging for STT, TTS, and LLM calls
    <http://logger.info|logger.info>("STT initialized with model nova-2.")
    <http://logger.info|logger.info>("LLM initialized with model gpt-4o-mini.")
    <http://logger.info|logger.info>("TTS initialized with model aura-asteria-en.")

    # Start the session
    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_input_options=RoomInputOptions(
            noise_cancellation=noise_cancellation_filter,
        ),
    )

    # Collect metrics
    @session.on("metrics_collected")
    def on_metrics_collected(event: MetricsCollectedEvent):
        usage_collector.collect(event.metrics)

    await session.say("Hello, how can I help you today?")

    # Log aggregated usage metrics when the session ends
    summary = usage_collector.get_summary()
    <http://logger.info|logger.info>(f"Usage: {summary}")

    # Define a shutdown callback to log usage when the job is finalized
    async def log_usage():
        summary = usage_collector.get_summary()
        <http://logger.info|logger.info>(f"Usage summary on shutdown: {summary}")

    ctx.add_shutdown_callback(log_usage)

if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint, port=9091))

max_tool_steps

max_tool_steps

AgentSession

session = AgentSession(
    max_tool_steps=5,  # Increase from default of 3
    # ... other parameters
)

speech_id

VoicePipelineAgent

from livekit.agents import llm
from livekit.agents.pipeline import VoicePipelineAgent

# Define function context with tools
function_context = llm.FunctionContext()

@function_context.ai_callable()
def get_weather(location: str) -> str:
    """Get the weather for a specific location."""
    return f"The weather in {location} is sunny."

# Create the VoicePipelineAgent
agent = VoicePipelineAgent(
    chat_ctx=llm.ChatContext().append(
        role="system",
        text="You are a helpful voice assistant."
    ),
    vad=silero.VAD.load(),
    stt=deepgram.STT(),
    llm=openai.LLM(),
    tts=openai.TTS(),
    fnc_ctx=function_context  # Pass function context to enable tools
)

FunctionContext

get_weather

fnc_ctx

VoicePipelineAgent

max_tool_steps

VoicePipelineAgent

agent = VoicePipelineAgent(
    chat_ctx=llm.ChatContext().append(
        role="system",
        text="You are a helpful voice assistant."
    ),
    vad=silero.VAD.load(),
    stt=deepgram.STT(),
    llm=openai.LLM(),
    tts=openai.TTS(),
    fnc_ctx=function_context,  # Pass function context to enable tools
    max_tool_steps=5  # Increase from the default of 3
)

import logging
logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
logger = logging.getLogger(__name__)

@function_context.ai_callable()
def get_weather(location: str) -> str:
    <http://logger.info|logger.info>(f"get_weather called with location: {location}")
    return f"The weather in {location} is sunny."

max_tool_steps

session = AgentSession(
    max_tool_steps=5,  # Allow up to 5 consecutive tool calls
    # other parameters...
)

max_nested_calls

chat_context

chat_context

chat_context

chat_context

ChatItem

role

content

callId

name

arguments

output

isError

chat_context

addMessage

update_chat_ctx

chat_context.addMessage({
  role: 'user',
  content: 'Hello, how can you help me today?'
})

BVCTelephony