if len(room.remote_participants) == 0:
    print("No participants left in the room")

participant_disconnected

@room.on("participant_disconnected")
def on_participant_disconnected(participant):
    if len(room.remote_participants) == 0:
        print("No participants left in the room")

import asyncio
from livekit import rtc

# Create a future that will be set when the room is empty
room_empty_future = asyncio.get_running_loop().create_future()

# Define a callback for when a participant disconnects
def on_participant_disconnected(_: rtc.Participant, fut=room_empty_future):
    if len(ctx.room.remote_participants) == 0 and not fut.done():
        fut.set_result(None)

# Attach the callback to the room's participant_disconnected event
ctx.room.on("participant_disconnected", on_participant_disconnected)

try:
    # Wait for the future to be resolved
    await asyncio.wait([room_empty_future], return_when=asyncio.FIRST_COMPLETED)
finally:
    # Cleanup the event listener
    ctx.room.off("participant_disconnected", on_participant_disconnected)
    # Ensure the session is closed properly
    await session.aclose()

participant_disconnected

stt_node

from livekit import rtc
from livekit.agents import ModelSettings, stt, Agent
from typing import AsyncIterable, Optional

class MaskingAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="Your agent instructions here",
            stt=your_stt_provider,
            llm=your_llm_provider,
            tts=your_tts_provider,
            vad=your_vad_provider
        )

    async def stt_node(
        self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
    ) -> Optional[AsyncIterable[stt.SpeechEvent]]:
        # First, get the default STT processing
        async for event in Agent.default.stt_node(self, audio, model_settings):
            # Here’s where you can post-process the transcribed text
            if hasattr(event, 'alternatives') and event.alternatives:
                # Let’s mask any sensitive info in the transcript
                original_text = event.alternatives[0].text
                masked_text = self.mask_sensitive_content(original_text)
                event.alternatives[0].text = masked_text
            yield event

    def mask_sensitive_content(self, text: str) -> str:
        # This is where your masking magic happens
        import re

        # Mask credit card numbers
        text = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CREDIT_CARD]', text)

        # Mask phone numbers
        text = re.sub(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '[PHONE_NUMBER]', text)

        # Feel free to add more patterns to mask as needed
        return text

stt_node

mask_sensitive_content

def setup_langfuse(
    metadata: dict[str, AttributeValue] | None = None,
    *,
    host: str | None = None,
    public_key: str | None = None,
    secret_key: str | None = None,
) -> TracerProvider:

LANGFUSE_PUBLIC_KEY

LANGFUSE_SECRET_KEY

LANGFUSE_HOST

before_tts_cb

async def _before_tts_cb(agent: VoicePipelineAgent, text: str | AsyncIterable[str]):
    if isinstance(text, str):
        # For non-streaming text
        return text.replace("[thinking]", "")
    else:
        # For streaming text
        async def process_stream():
            async for chunk in text:
                yield chunk.replace("[thinking]", "")
        return process_stream()

agent = VoicePipelineAgent(
    vad=ctx.proc.userdata["vad"],
    stt=openai.STT(),
    llm=openai.LLM.with_groq(model="deepseek-r1-distill-llama-70b"),
    tts=openai.TTS(),
    before_tts_cb=_before_tts_cb
)

llm_node

llm_node

class CustomAgent(Agent):
    async def llm_node(self, chat_ctx, tools, model_settings=None):
        async def process_stream():
            async with self.llm.chat(chat_ctx=chat_ctx, tools=tools) as stream:
                async for chunk in stream:
                    content = getattr(chunk.delta, 'content', None)
                    if content:
                        chunk.delta.content = content.replace("[thinking]", "")
                    yield chunk
        return process_stream()

strip_thinking_tokens

def clean_text(text: str) -> str:
    text = text.replace("[thinking]", "")
    tool_phrases = ["I'll use the", "Let me call the", "Using the tool"]
    for phrase in tool_phrases:
        text = text.replace(phrase, "")
    return text.strip()

tts_node

AsyncIterable[str]

filter_markdown

async def tts_node(
    self, text: AsyncIterable[str], model_settings: ModelSettings
) -> AsyncIterable[rtc.AudioFrame]:
    # Process the text to remove markdown
    filtered_text = filter_markdown(text)
    # Pass the processed text to the default TTS implementation
    return super().tts_node(filtered_text, model_settings)

filter_markdown

stt_node()

transcription_node()

stt_node()

async def stt_node(self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings) -> Optional[AsyncIterable[stt.SpeechEvent]]:
    # Add your custom processing before STT here
    events = Agent.default.stt_node(self, audio, model_settings)
    # Add your custom processing after STT here
    return events

transcription_node()

async def transcription_node(self, text: AsyncIterable[str], model_settings: ModelSettings):
    replacements = {
        "hello": "👋 HELLO",
        "goodbye": "GOODBYE 👋",
    }

    async def process_text():
        async for chunk in text:
            modified_chunk = chunk
            for word, replacement in replacements.items():
                modified_chunk = modified_chunk.replace(word, replacement)
            yield modified_chunk

    return process_text()

stt_node()

transcription_node()

stt_node

audio

text

mask_sensitive_content

async def stt_node(
    self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
) -> AsyncIterable[rtc.AudioFrame]:
    """Content Filtering"""
    async for event in super().stt_node(audio, model_settings):
        if hasattr(event, 'alternatives') and event.alternatives:
            # Apply masking to the transcribed text
            original_text = event.alternatives[0].text
            filtered_text = self.mask_sensitive_content(original_text)
            event.alternatives[0].text = filtered_text
        yield event

stt_node

audio

mask_sensitive_content

stt_node

stt_node

stt_node

rtc.AudioFrame

stt.SpeechEvent

async def stt_node(
    self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
) -> AsyncIterable[rtc.AudioFrame]:
    """Custom STT Node with Content Filtering"""
    # Start by calling the default STT processing to get speech events
    async for event in super().stt_node(audio, model_settings):
        if hasattr(event, 'alternatives') and event.alternatives:
            # Grab the transcribed text from the first alternative
            original_text = event.alternatives[0].text

            # Apply your own filtering or masking here
            filtered_text = self.mask_sensitive_content(original_text)

            # Swap out the original text for your filtered version
            event.alternatives[0].text = filtered_text

        # Send the updated event back into the pipeline
        yield event

audio

AsyncIterable[rtc.AudioFrame]

stt_node

stt.SpeechEvent

alternatives

mask_sensitive_content

def mask_sensitive_content(self, text: str) -> str:
    """Mask sensitive information in the transcribed text."""
    import re

    # Hide credit card numbers
    text = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CREDIT_CARD]', text)

    # Hide phone numbers
    text = re.sub(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '[PHONE_NUMBER]', text)

    # You can add more patterns here as needed
    return text

stt_node

mask_sensitive_content

stt_node

ContentOptimizer

stt_node

from typing import AsyncIterable
from livekit import rtc
from livekit.agents import ModelSettings

class CustomAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="Your agent instructions here",
            stt=your_stt_provider,
            llm=your_llm_provider,
            tts=your_tts_provider,
            vad=your_vad_provider
        )
        self.content_optimizer = ContentOptimizer()

    async def stt_node(
        self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
    ) -> AsyncIterable[rtc.AudioFrame]:
        async for event in super().stt_node(audio, model_settings):
            if hasattr(event, 'alternatives') and event.alternatives:
                # Grab the transcribed text
                original_text = event.alternatives[0].text

                # Run it through ContentOptimizer
                filtered_text = self.content_optimizer.process(original_text)

                # Swap out the original text with the cleaned-up version
                event.alternatives[0].text = filtered_text

            yield event

stt_node

ContentOptimizer.process()

process()

class STTProcessor:
    def __init__(self, stt_provider, content_optimizer):
        self.stt_provider = stt_provider
        self.content_optimizer = content_optimizer

    async def process_audio(self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings) -> str:
        async for event in self.stt_provider.stt_node(audio, model_settings):
            if hasattr(event, 'alternatives') and event.alternatives:
                text = event.alternatives[0].text
                # Apply masking or preprocessing
                return self.content_optimizer.process(text)
        return ""

class LanguageProcessor:
    def __init__(self, llm_provider):
        self.llm_provider = llm_provider

    async def process_text(self, text: str, language: str) -> str:
        # Use the selected language model to process the text
        response = await self.llm_provider.chat(text, language=language)
        return response

class ModularAgent:
    def __init__(self, stt_processor, language_processors):
        self.stt_processor = stt_processor
        self.language_processors = language_processors

    async def handle_audio(self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings, language: str):
        # Process audio to get text
        text = await self.stt_processor.process_audio(audio, model_settings)

        # Pass text to the selected language processor
        if language in self.language_processors:
            response = await self.language_processors[language].process_text(text, language)
            return response
        else:
            raise ValueError(f"Unsupported language: {language}")

class LanguageAgent:
    def __init__(self, stt_processor, llm_processor, tools):
        self.stt_processor = stt_processor
        self.llm_processor = llm_processor
        self.tools = tools

    async def handle_user_input(self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings):
        # Turn the audio into text
        text = await self.stt_processor.process_audio(audio, model_settings)

        # Send the text to the LLM for processing
        response = await self.llm_processor.process_text(text)

        # Check the response and decide what to do
        if "tool" in response:
            tool_name = self.extract_tool_name(response)
            result = await self.tools[tool_name].execute(response)
            return result
        else:
            return response

    def extract_tool_name(self, response: str) -> str:
        # Grab the tool name from the LLM response
        return response.split()[1]  # Just an example

{
  "action": "call_tool",
  "tool_name": "weather_api",
  "parameters": {
    "location": "New York"
  }
}

tools = {
    "weather_api": WeatherAPI(),
    "calendar": CalendarTool(),
    "email": EmailTool()
}

class LanguageAgent:
    def __init__(self, stt_processor, llm_processor, tools):
        self.stt_processor = stt_processor
        self.llm_processor = llm_processor
        self.tools = tools

    async def handle_user_input(self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings):
        # Turn audio into text
        text = await self.stt_processor.process_audio(audio, model_settings)

        # Send the text to the LLM and get a response
        response = await self.llm_processor.process_text(text)

        # Check if the response says to call a tool
        if "action" in response and response["action"] == "call_tool":
            tool_name = response["tool_name"]
            parameters = response.get("parameters", {})

            # Call the tool if it exists
            if tool_name in self.tools:
                result = await self.tools[tool_name].execute(**parameters)
                return result
            else:
                return f"Tool {tool_name} not found."
        else:
            return response

execute

class WeatherAPI:
    async def execute(self, location):
        # Fetch the weather and return it
        return f"The weather in {location} is sunny."

class CalendarTool:
    async def execute(self, event_name, date):
        # Add an event to the calendar
        return f"Event '{event_name}' added to your calendar on {date}."

{
  "action": "call_tool",
  "tool_name": "weather_api",
  "parameters": {
    "location": "New York"
  }
}

from livekit import api
from livekit.protocol import sip as proto_sip

async def transfer_call(participant_identity: str, transfer_to: str):
    """Transfer a SIP call to another number using SIP REFER."""
    # Initialize LiveKit API client
    livekit_api = api.LiveKitAPI(
        url=os.getenv('LIVEKIT_URL'),
        api_key=os.getenv('LIVEKIT_API_KEY'),
        api_secret=os.getenv('LIVEKIT_API_SECRET')
    )

    # Create transfer request
    transfer_request = proto_sip.TransferSIPParticipantRequest(
        participant_identity=participant_identity,
        room_name="your_room_name",  # Replace with the actual room name
        transfer_to=transfer_to,  # e.g., "tel:+15105550100"
        play_dialtone=True
    )

    # Execute transfer
    await livekit_api.sip.transfer_sip_participant(transfer_request)

class SessionManager:
    async def start_transfer(self):
        # Put customer on hold
        self.start_hold()

        # Create supervisor room
        supervisor_room_name = self.customer_room.name + "-supervisor"
        self.supervisor_room = rtc.Room()

        # Connect to supervisor room
        await self.supervisor_room.connect(os.getenv("LIVEKIT_URL"), token.to_jwt())

        # Dial supervisor
        await self.lkapi.sip.create_sip_participant(
            api.CreateSIPParticipantRequest(
                sip_trunk_id="your_sip_trunk_id",
                sip_call_to="supervisor_number",
                room_name=supervisor_room_name,
                participant_identity="supervisor-sip",
                wait_until_answered=True,
            )
        )

    async def merge_calls(self):
        # Move supervisor to customer room
        await self.lkapi.room.move_participant(
            api.MoveParticipantRequest(
                room=self.supervisor_room.name,
                identity="supervisor-sip",
                destination_room=self.customer_room.name,
            )
        )

        self.stop_hold()
        await self.customer_session.say("You are on the line with my supervisor.")

def _setup_event_handlers(self, room: rtc.Room):
    @room.on("sip_dtmf_received")
    def handle_dtmf(dtmf_event: rtc.SipDTMF):
        digit = dtmf_event.digit
        identity = dtmf_event.participant.identity

        department_numbers = {
            "1": ("BILLING_PHONE_NUMBER", "Billing"),
            "2": ("TECH_SUPPORT_PHONE_NUMBER", "Tech Support"),
            "3": ("CUSTOMER_SERVICE_PHONE_NUMBER", "Customer Service")
        }

        if digit in department_numbers:
            env_var, dept_name = department_numbers[digit]
            transfer_number = f"tel:{os.getenv(env_var)}"
            asyncio.create_task(self._handle_transfer(identity, transfer_number, dept_name))

update_options

update_options

session.tts.update_options(
    voice=hume.VoiceByName(name="Colton Rivers", provider=hume.VoiceProvider.hume),
    description="The voice exudes calm, serene, and peaceful qualities, like a gentle stream flowing through a quiet forest.",
    speed=2,
)

class RimeAgent(Agent):
    def __init__(self):
        super().__init__(
            tts=rime.TTS(model="mistv2", speaker="abbie"),
            # ... other config
        )

    @function_tool
    async def switch_to_elevenlabs(self):
        """Switch to ElevenLabs TTS voice"""
        return ElevenLabsAgent()

    @function_tool
    async def switch_to_cartesia(self):
        """Switch to Cartesia TTS voice"""
        return CartesiaAgent()

@ctx.room.on("participant_attributes_changed")
def on_participant_attributes_changed(changed_attributes: dict[str, str], participant: rtc.Participant):
    if "voice" in changed_attributes:
        voice_id = participant.attributes.get("voice")
        # Update TTS options dynamically
        tts._opts.voice = voice_id
        asyncio.create_task(agent.say("How do I sound now?", allow_interruptions=True))

update_options

async def _switch_language(self, language_code: str):
    if self.tts is not None:
        self.tts.update_options(language=language_code)

    if self.stt is not None:
        self.stt.update_options(language=deepgram_language)

    await self.session.say(self.greetings[language_code])

update_options

update_options

session.tts.update_options(
    voice="Joanna",  # Swap this out for whichever AWS Polly voice you like
    language="en-US"  # Optional: specify the language if you want
)

class LanguageAgent:
    def __init__(self, tts):
        self.tts = tts

    async def switch_voice(self, voice_name: str):
        """Dynamically switch the TTS voice."""
        self.tts.update_options(voice=voice_name)
        await self.session.say(f"Voice switched to {voice_name}.")

async def switch_language_and_voice(self, language_code: str, voice_name: str):
    self.tts.update_options(language=language_code, voice=voice_name)
    await self.session.say(f"Language switched to {language_code} and voice to {voice_name}.")

if user_input == "Change voice to Joanna":
    await agent.switch_voice("Joanna")

language_code

voice

speech_engine

if language_code not in self.aws_languages:
    raise ValueError(f"Invalid language code: {language_code}")

if voice not in self.aws_voice:
    raise ValueError(f"Invalid voice: {voice}")

if speech_engine not in self.aws_speech_engine:
    raise ValueError(f"Invalid speech engine: {speech_engine}")

await self.session.say(f"Language switched to {self.language_names[language_code]} with voice {voice} using {speech_engine}.")

aws_voice = self.aws_voice.get(voice, "default_voice")
aws_speech_engine = self.aws_speech_engine.get(speech_engine, "standard")

logging.error(f"Error switching language to {language_code}, voice: {voice}, engine: {speech_engine}: {e}")

self.session.say

async def _switch_language(self, language_code: str, voice: str, speech_engine: str):
    """Helper Function To Switch Languages"""
    try:
        if language_code == self.current_language:
            await self.session.say("You are already using this language.")
            return

        # Validate inputs
        if language_code not in self.aws_languages:
            raise ValueError(f"Invalid language code: {language_code}")

        if voice not in self.aws_voice:
            raise ValueError(f"Invalid voice: {voice}")

        if speech_engine not in self.aws_speech_engine:
            raise ValueError(f"Invalid speech engine: {speech_engine}")

        # Update TTS options
        if self.tts is not None:
            self.tts.update_options(language_code=language_code, voice=voice, speech_engine=speech_engine)

        # Update STT options
        if self.stt is not None:
            aws_languages = self.aws_languages.get(language_code)
            aws_voice = self.aws_voice.get(voice)
            aws_speech_engine = self.aws_speech_engine.get(speech_engine)
            self.stt.update_options(language_code=aws_languages)

        # Update current language
        self.current_language = language_code

        # Confirmation message
        await self.session.say(f"Language switched to {self.language_names[language_code]} with voice {voice} using {speech_engine}.")

    except Exception as e:
        logging.error(f"Error switching language to {language_code}, voice: {voice}, engine: {speech_engine}: {e}")
        raise CustomException(e, sys)

IrishAgent

@function_tool

aws_voice = self.aws_voice.get(voice, "default_voice")
aws_speech_engine = self.aws_speech_engine.get(speech_engine, "standard")

switch_to_en_gb

@function_tool
async def switch_language(self, language_code: str, voice: str, speech_engine: str):
    await self._switch_language(language_code, voice, speech_engine)

stt_node

await self.session.say(f"Language switched to {self.language_names[language_code]} with voice {voice} using {speech_engine}.")

aws_languages

aws_voice

@function_tool
async def switch_language(self, language_code: str, voice: str, speech_engine: str):
    """Switch to a specified language, voice, and speech engine."""
    await self._switch_language(language_code, voice, speech_engine)

stt_node

stt_node

IrishAgent

stt_node

stt_node

async def stt_node(
    self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
) -> AsyncIterable[rtc.AudioFrame]:
    """Custom STT Node with Content Filtering"""
    # First, call the default STT processing to grab speech events
    async for event in super().stt_node(audio, model_settings):
        if hasattr(event, 'alternatives') and event.alternatives:
            # Pull out the transcribed text from the first alternative
            original_text = event.alternatives[0].text

            # Here’s where you can filter or mask the content
            filtered_text = content_optimizer.process(original_text)

            # Swap the original text with your filtered version
            event.alternatives[0].text = filtered_text

        # Send the updated event back down the pipeline
        yield event

stt_node

alternatives

ContentOptimizer

stt_node

ContentOptimizer

from livekit.plugins import silero

vad = silero.VAD.load(
    min_speech_duration=2.0,  # At least 2 seconds of speech to start
    min_silence_duration=1.5,  # Wait 1.5 seconds of silence before ending
    max_buffered_speech=10.0  # Buffer up to 10 seconds of speech
)

update_options

class MultiLanguageAgent(Agent):
    def __init__(self):
        super().__init__(
            stt=aws.STT(language="en-US", region="us-east-1"),
            tts=aws.TTS(region="us-east-1", api_key=os.getenv("AWS_ACCESS_KEY_ID"), api_secret=os.getenv("AWS_SECRET_ACCESS_KEY"))
        )
        self.current_language = "en-US"
        self.language_mappings = {
            "en-US": {"voice": "Joanna", "stt_language": "en-US"},
            "fr-FR": {"voice": "Celine", "stt_language": "fr-FR"},
            "de-DE": {"voice": "Hans", "stt_language": "de-DE"},
            "hi-IN": {"voice": "Aditi", "stt_language": "hi-IN"}
        }

    async def switch_language(self, language_code: str):
        if language_code not in self.language_mappings:
            raise ValueError(f"Unsupported language: {language_code}")

        # Update TTS options
        language_settings = self.language_mappings[language_code]
        self.tts.update_options(language=language_code, voice=language_settings["voice"])

        # Update STT options
        self.stt.update_options(language=language_settings["stt_language"])

        self.current_language = language_code
        await self.session.say(f"Language switched to {language_code}.")

@function_tool
async def switch_to_french(self):
    await self.switch_language("fr-FR")

turn_detection=multilingual()

turn_detection=multilingual()

turn_detection=multilingual()

AgentSession

stt

vad

realtime_llm

manual

allow_interruptions

min_interruption_duration

min_interruption_words

min_endpointing_delay

max_endpointing_delay

preemptive_generation=True

video_sampler

user_away_timeout

agent_false_interruption_timeout

max_tool_steps

prewarm

WorkerOptions

session = AgentSession(
    vad=silero.VAD.load(),
    stt=deepgram.STT(model="nova-3"),
    llm=openai.LLM(model="gpt-4o-mini"),
    tts=openai.TTS(voice="ash"),
    preemptive_generation=True,
    turn_detection=MultilingualModel(),
    max_tool_steps=10,
    user_away_timeout=30.0,
)

AgentSession

LANGFUSE_PUBLIC_KEY

LANGFUSE_SECRET_KEY

LANGFUSE_HOST

<https://cloud.langfuse.com>

setup_langfuse()

from livekit.agents.telemetry import set_tracer_provider
import base64
import os

def setup_langfuse(metadata=None, host=None, public_key=None, secret_key=None):
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.export import BatchSpanProcessor

    public_key = public_key or os.getenv("LANGFUSE_PUBLIC_KEY")
    secret_key = secret_key or os.getenv("LANGFUSE_SECRET_KEY")
    host = host or os.getenv("LANGFUSE_HOST")

    if not public_key or not secret_key or not host:
        raise ValueError("LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, and LANGFUSE_HOST must be set")

    langfuse_auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
    os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = f"{host.rstrip('/')}/api/public/otel"
    os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {langfuse_auth}"

    trace_provider = TracerProvider()
    trace_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    set_tracer_provider(trace_provider, metadata=metadata)
    return trace_provider

setup_langfuse()

async def entrypoint(ctx: JobContext):
    # Setup Langfuse tracing
    trace_provider = setup_langfuse(metadata={"langfuse.session.id": ctx.room.name})

    # Optional: flush traces on shutdown
    async def flush_trace():
        trace_provider.force_flush()

    ctx.add_shutdown_callback(flush_trace)

    # Start the agent session
    session = AgentSession(vad=silero.VAD.load())
    await session.start(agent=Kelly(), room=ctx.room)

python langfuse_trace.py

ctx.room

ctx.room.name

ParticipantConnected

ParticipantDisconnected

TrackPublished

TrackUnpublished

RoomMetadataChanged

ActiveSpeakersChanged

# Example of unique room names
room_name = f"user_{user_id}_{uuid.uuid4()}"

{
  "dispatch_rule": {
    "rule": {
      "dispatchRuleIndividual": {
        "roomPrefix": "call-"
      }
    }
  }
}

lk sip dispatch-rule create dispatch-rule.json

from livekit import api

lk_api = api.LiveKitAPI(
    url=os.getenv("LIVEKIT_URL"),
    api_key=os.getenv("LIVEKIT_API_KEY"),
    api_secret=os.getenv("LIVEKIT_API_SECRET"),
)

inbound_trunk_request = api.CreateSIPInboundTrunkRequest(
    trunk=api.SIPInboundTrunkInfo(
        name="My Inbound Trunk",
        numbers=["+1234567890"],  # Don’t forget to replace this with your actual numbers
        krisp_enabled=True,       # This turns on noise cancellation for clearer calls
    ),
)
inbound_trunk = await lk_api.sip.create_inbound_trunk(inbound_trunk_request)

dispatch_rule_request = api.CreateSIPDispatchRuleRequest(
    trunk_ids=[inbound_trunk.sip_trunk_id],
    rule=api.SIPDispatchRule(
        dispatch_rule_individual=api.SIPDispatchRuleIndividual(room_prefix="call-")
    ),
)
dispatch_rule = await lk_api.sip.create_sip_dispatch_rule(dispatch_rule_request)

import asyncio
import os
from livekit import api

async def setup_sip_isolation():
    lk_api = api.LiveKitAPI(
        url=os.getenv("LIVEKIT_URL"),
        api_key=os.getenv("LIVEKIT_API_KEY"),
        api_secret=os.getenv("LIVEKIT_API_SECRET"),
    )

    # Create inbound trunk
    inbound_trunk = await lk_api.sip.create_inbound_trunk(
        api.CreateSIPInboundTrunkRequest(
            trunk=api.SIPInboundTrunkInfo(name="Inbound", numbers=["+1234567890"])
        )
    )

    # Create dispatch rule
    dispatch_rule = await lk_api.sip.create_sip_dispatch_rule(
        api.CreateSIPDispatchRuleRequest(
            trunk_ids=[inbound_trunk.sip_trunk_id],
            rule=api.SIPDispatchRule(
                dispatch_rule_individual=api.SIPDispatchRuleIndividual(room_prefix="user-call")
            ),
        )
    )
    print(f"Dispatch rule created: {dispatch_rule.sip_dispatch_rule_id}")

    await asyncio.Future()  # Keep the service running

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

ctx.room.metadata

room_name = f"session_{uuid.uuid4()}"

setup_langfuse(
    metadata={
        "langfuse.session.id": ctx.room.name,
        "langfuse.user.id": participant.identity,
        "langfuse.call.type": "SIP" if is_sip_call else "WEB"
    })

async def shutdown():
    transcription_logger.close()
    caller_info.close()
    await ctx.disconnect()

ctx.add_shutdown_callback(shutdown)

logging.error(f"Error in entrypoint: {e}", exc_info=True)

import uuid
room_name = f"session_{uuid.uuid4()}"

AgentSession

async def start_session(room_name):
    session = AgentSession(
        use_tts_aligned_transcript=True,
        user_away_timeout=30,
        max_tool_steps=10
    )

    agent = IrishAgent(
        llm_instance=llm,
        tool=[info_tool],
        vad=vad
    )

    await session.start(agent=agent, room=room_name, room_input_options=RoomInputOptions(close_on_disconnect=False))

@ctx.room.on("disconnected")
def on_room_disconnected():
    print(f"Room {ctx.room.name} disconnected. Cleaning up resources.")
    # Perform cleanup tasks here

dispatch_rule_request = api.CreateSIPDispatchRuleRequest(
    trunk_ids=[inbound_trunk.sip_trunk_id],
    rule=api.SIPDispatchRule(
        dispatch_rule_individual=api.SIPDispatchRuleIndividual(room_prefix="call-")
    ),
)
dispatch_rule = await lk_api.sip.create_sip_dispatch_rule(dispatch_rule_request)

active_sessions = {}

def add_session(session_id, session):
    active_sessions[session_id] = session

def remove_session(session_id):
    if session_id in active_sessions:
        del active_sessions[session_id]

import uuid

# Generate a unique room name
unique_room_name = f"session_{uuid.uuid4()}"
ctx.room.name = unique_room_name

active_sessions = {}

def add_session(session_id, session):
    active_sessions[session_id] = session

def remove_session(session_id):
    if session_id in active_sessions:
        del active_sessions[session_id]

add_session(unique_room_name, session)

remove_session(unique_room_name)

on_room_disconnected

@ctx.room.on("disconnected")
def on_room_disconnected():
    print(f"🏠  Room {ctx.room.name} disconnected - cleaning up resources.")
    remove_session(ctx.room.name)
    if current_call_id and transcription_logger:
        transcription_logger.close_transcription()
        caller_info.end_call(current_call_id)
        transcription_logger.remove_txt()
        print(f"✅  Call {current_call_id} ended and transcription saved to MongoDB")

import uuid

def generate_auth_token():
    return str(uuid.uuid4())

auth_token = generate_auth_token()
room_name = f"sip_{auth_token}"

participant_request = api.CreateSIPParticipantRequest(
    room_name=room_name,
    sip_trunk_id=os.getenv("SIP_TRUNK_ID"),
    sip_call_to=caller_number,
    participant_identity=auth_token,
    metadata={"auth_token": auth_token}
)

active_sessions[auth_token] = room_name

del active_sessions[auth_token]

import uuid

def generate_auth_token():
    """Generate a unique authentication token."""
    return str(uuid.uuid4())

auth_token = generate_auth_token()
room_name = f"sip_{auth_token}"

from livekit import api

participant_request = api.CreateSIPParticipantRequest(
    room_name=room_name,
    sip_trunk_id=os.getenv("SIP_TRUNK_ID"),
    sip_call_to="+1234567890",  # Just swap this with the caller's number
    participant_identity=auth_token,
    metadata={"auth_token": auth_token}
)

# Use the LiveKit API to create the participant
lk_api = api.LiveKitAPI(
    url=os.getenv("LIVEKIT_URL"),
    api_key=os.getenv("LIVEKIT_API_KEY"),
    api_secret=os.getenv("LIVEKIT_API_SECRET"),
)
await lk_api.sip.create_sip_participant(participant_request)

active_sessions = {}

# Add the session to the active sessions
active_sessions[auth_token] = room_name

@ctx.room.on("disconnected")
def on_room_disconnected():
    """Handle room disconnection and clean up resources."""
    if auth_token in active_sessions:
        print(f"🏠  Room {active_sessions[auth_token]} disconnected. Cleaning up resources.")
        del active_sessions[auth_token]

import uuid
import os
from livekit import api

# Step 1: Generate a unique auth token
def generate_auth_token():
    return str(uuid.uuid4())

# Step 2: Assign a unique room name
auth_token = generate_auth_token()
room_name = f"sip_{auth_token}"

# Step 3: Create a SIP participant
participant_request = api.CreateSIPParticipantRequest(
    room_name=room_name,
    sip_trunk_id=os.getenv("SIP_TRUNK_ID"),
    sip_call_to="+1234567890",  # Just swap this with the caller's number
    participant_identity=auth_token,
    metadata={"auth_token": auth_token}
)

lk_api = api.LiveKitAPI(
    url=os.getenv("LIVEKIT_URL"),
    api_key=os.getenv("LIVEKIT_API_KEY"),
    api_secret=os.getenv("LIVEKIT_API_SECRET"),
)

# Step 4: Track active sessions
active_sessions = {}
active_sessions[auth_token] = room_name

# Step 5: Handle session cleanup
@ctx.room.on("disconnected")
def on_room_disconnected():
    if auth_token in active_sessions:
        print(f"🏠  Room {active_sessions[auth_token]} disconnected. Cleaning up resources.")
        del active_sessions[auth_token]

# Step 6: Create the participant
await lk_api.sip.create_sip_participant(participant_request)

import uuid

# Generate a unique auth token and room name
auth_token = str(uuid.uuid4())
room_name = f"sip_{auth_token}"
ctx.room.name = room_name

from livekit import api

participant_request = api.CreateSIPParticipantRequest(
    room_name=room_name,
    sip_trunk_id=os.getenv("SIP_TRUNK_ID"),
    sip_call_to="+1234567890",  # Swap in the caller's number here
    participant_identity=auth_token,
    metadata={"auth_token": auth_token}
)

lk_api = api.LiveKitAPI(
    url=os.getenv("LIVEKIT_URL"),
    api_key=os.getenv("LIVEKIT_API_KEY"),
    api_secret=os.getenv("LIVEKIT_API_SECRET"),
)
await lk_api.sip.create_sip_participant(participant_request)

active_sessions = {}

# Add the session to the active sessions
active_sessions[auth_token] = room_name

on_room_disconnected

@ctx.room.on("disconnected")
def on_room_disconnected():
    if auth_token in active_sessions:
        print(f"🏠  Room {active_sessions[auth_token]} disconnected. Cleaning up resources.")
        del active_sessions[auth_token]

async def entrypoint(ctx: JobContext):
    try:
        # Generate a unique auth token and room name
        auth_token = str(uuid.uuid4())
        room_name = f"sip_{auth_token}"
        ctx.room.name = room_name

        # Set up Langfuse
        setup_langfuse(
            metadata={
                "langfuse.session.id": ctx.room.name,
                "langfuse.session": ctx.room.metadata
            }
        )

        print("🚀  Starting LiveKit agent...")
        await ctx.connect()

        # Create SIP participant
        participant_request = api.CreateSIPParticipantRequest(
            room_name=room_name,
            sip_trunk_id=os.getenv("SIP_TRUNK_ID"),
            sip_call_to="+1234567890",
            participant_identity=auth_token,
            metadata={"auth_token": auth_token}
        )

        lk_api = api.LiveKitAPI(
            url=os.getenv("LIVEKIT_URL"),
            api_key=os.getenv("LIVEKIT_API_KEY"),
            api_secret=os.getenv("LIVEKIT_API_SECRET"),
        )
        await lk_api.sip.create_sip_participant(participant_request)

        # Track active sessions
        active_sessions[auth_token] = room_name

        # Handle room disconnect
        @ctx.room.on("disconnected")
        def on_room_disconnected():
            if auth_token in active_sessions:
                print(f"🏠  Room {active_sessions[auth_token]} disconnected. Cleaning up resources.")
                del active_sessions[auth_token]

        # Continue with the rest of your session setup...

    except Exception as e:
        print(f"❌  Error: {e}")
        raise

call-

rule = api.SIPDispatchRule(
    dispatch_rule_individual=api.SIPDispatchRuleIndividual(
        room_prefix='call-',
    )
)

request = api.CreateSIPDispatchRuleRequest(
    dispatch_rule=api.SIPDispatchRuleInfo(
        rule=rule,
        name='My dispatch rule',
        trunk_ids=[],
        room_config=api.RoomConfiguration(
            agents=[api.RoomAgentDispatch(
                agent_name="inbound-agent",
                metadata="job dispatch metadata",
            )]
        )
    )
)

dispatch = await lkapi.sip.create_sip_dispatch_rule(request)
print("created dispatch", dispatch)

trunk_ids

room_prefix

ctx.room.name

ctx.room.metadata

setup_langfuse(
    metadata={
        "langfuse.session.id": ctx.room.name,
        "langfuse.sesssion": ctx.room.metadata,
    }
)

"langfuse.sesssion"

"langfuse.session"

await ctx.connect()

lkapi = api.LiveKitAPI(
    url=os.getenv("LIVEKIT_URL"),
    api_key=os.getenv("LIVEKIT_API_KEY"),
    api_secret=os.getenv("LIVEKIT_API_SECRET"),
)

# Create a dispatch rule to place each caller in a separate room
rule = api.SIPDispatchRule(
    dispatch_rule_individual=api.SIPDispatchRuleIndividual(
        room_prefix='call-',
    )
)

request = api.CreateSIPDispatchRuleRequest(
    dispatch_rule=api.SIPDispatchRuleInfo(
        rule=rule,
        name='My dispatch rule',
        trunk_ids=["your_trunk_id"],  # Replace with actual trunk IDs
        room_config=api.RoomConfiguration(
            agents=[api.RoomAgentDispatch(
                agent_name="inbound-agent",
                metadata="job dispatch metadata",
            )]
        )
    )
)

dispatch = await lkapi.sip.create_sip_dispatch_rule(request)
print("created dispatch", dispatch)
await lkapi.aclose()

# Set up Langfuse
setup_langfuse(
    metadata={
        "langfuse.session.id": ctx.room.name,
        "langfuse.session": ctx.room.metadata,
    }
)

print("🚀  Starting LiveKit agent...")
await ctx.connect()

dispatch_rule_individual

room_prefix

call-

call-12345

call-67890

room_config

call-abc123

abc123

trunk_ids

from livekit import api

lk_api = api.LiveKitAPI(
    url=os.getenv("LIVEKIT_URL"),
    api_key=os.getenv("LIVEKIT_API_KEY"),
    api_secret=os.getenv("LIVEKIT_API_SECRET"),
)

# List all SIP trunks
trunks = await lk_api.sip.list_sip_trunks()
for trunk in trunks.items:
    print(f"Trunk Name: {trunk.name}, Trunk ID: {trunk.sip_trunk_id}")

sip_trunk_id

trunk_ids

"your_trunk_id"

rule = api.SIPDispatchRule(
    dispatch_rule_individual=api.SIPDispatchRuleIndividual(
        room_prefix='call-',
    )
)

request = api.CreateSIPDispatchRuleRequest(
    dispatch_rule=api.SIPDispatchRuleInfo(
        rule=rule,
        name='My dispatch rule',
        trunk_ids=["trunk_12345"],  # Just replace this with your real trunk ID
        room_config=api.RoomConfiguration(
            agents=[api.RoomAgentDispatch(
                agent_name="inbound-agent",
                metadata="job dispatch metadata",
            )]
        )
    )
)

dispatch = await lk_api.sip.create_sip_dispatch_rule(request)
print("Created dispatch rule:", dispatch)

dispatch_rules = await lk_api.sip.list_sip_dispatch_rule()
for rule in dispatch_rules.items:
    print(f"Dispatch Rule Name: {rule.name}, Trunk IDs: {rule.trunk_ids}")

call-

call-abc123

call-def456

CreateSIPInboundTrunkRequest

from livekit import api

lk_api = api.LiveKitAPI(
    url=os.getenv("LIVEKIT_URL"),
    api_key=os.getenv("LIVEKIT_API_KEY"),
    api_secret=os.getenv("LIVEKIT_API_SECRET"),
)

# Create a new SIP trunk
inbound_trunk_request = api.CreateSIPInboundTrunkRequest(
    trunk=api.SIPInboundTrunkInfo(
        name="My Inbound Trunk",  # Give your trunk a name
        numbers=["+1234567890"],  # Swap this out for your actual phone numbers
        krisp_enabled=True  # Turn on noise cancellation for clearer calls
    )
)

inbound_trunk = await lk_api.sip.create_inbound_trunk(inbound_trunk_request)
print(f"Created SIP trunk: {inbound_trunk.sip_trunk_id}")

sip_trunk_id

rule = api.SIPDispatchRule(
    dispatch_rule_individual=api.SIPDispatchRuleIndividual(
        room_prefix='call-',
    )
)

request = api.CreateSIPDispatchRuleRequest(
    dispatch_rule=api.SIPDispatchRuleInfo(
        rule=rule,
        name='My dispatch rule',
        trunk_ids=[inbound_trunk.sip_trunk_id],  # Here’s where you use the trunk ID
        room_config=api.RoomConfiguration(
            agents=[api.RoomAgentDispatch(
                agent_name="inbound-agent",
                metadata="job dispatch metadata",
            )]
        )
    )
)

dispatch = await lk_api.sip.create_sip_dispatch_rule(request)
print("Created dispatch rule:", dispatch)

trunks = await lk_api.sip.list_sip_trunks()
for trunk in trunks.items:
    print(f"Trunk Name: {trunk.name}, Trunk ID: {trunk.sip_trunk_id}")

+1234567890

room_name = "my-room"
agent_name = "test-agent"

create_token_with_agent_dispatch

"my_participant"

room_name

RoomConfiguration(
    agents=[
        RoomAgentDispatch(
            agent_name="test-agent",
            metadata='{"user_id": "12345"}'
        )
    ],
)

test-agent

user_id

to_jwt()

token = AccessToken()
    .with_identity("my_participant")
    .with_grants(VideoGrants(room_join=True, room=room_name))
    .with_room_config(RoomConfiguration(...))
    .to_jwt()

return token

room_name

agent_name

metadata

import asyncio
from livekit import api

async def handle_inbound_call(caller_id):
    # Generate a unique room name for the inbound call
    room_name = f"inbound-call-{caller_id}"
    agent_name = "test-agent"

    # Initialize LiveKit API client
    lkapi = api.LiveKitAPI(
        url=os.getenv("LIVEKIT_URL"),
        api_key=os.getenv("LIVEKIT_API_KEY"),
        api_secret=os.getenv("LIVEKIT_API_SECRET"),
    )

    # Create an agent dispatch for the room
    dispatch = await lkapi.agent_dispatch.create_dispatch(
        api.CreateAgentDispatchRequest(
            agent_name=agent_name,
            room=room_name,
            metadata=f'{{"caller_id": "{caller_id}"}}'
        )
    )
    print("Created dispatch for inbound call:", dispatch)

    # List all dispatches in the room
    dispatches = await lkapi.agent_dispatch.list_dispatch(room_name=room_name)
    print(f"There are {len(dispatches)} dispatches in {room_name}")

    await lkapi.aclose()

# Simulate handling an inbound call
asyncio.run(handle_inbound_call("+1234567890"))

inbound-call-+1234567890

handle_inbound_call

uuid

import uuid

# Generate a unique ID for the call
unique_id = str(uuid.uuid4())
room_name = f"inbound-call-{unique_id}"

import asyncio
import uuid
from livekit import api

async def handle_inbound_call():
    # Generate a unique ID for the call
    unique_id = str(uuid.uuid4())
    room_name = f"inbound-call-{unique_id}"
    agent_name = "test-agent"

    # Initialize LiveKit API client
    lkapi = api.LiveKitAPI(
        url=os.getenv("LIVEKIT_URL"),
        api_key=os.getenv("LIVEKIT_API_KEY"),
        api_secret=os.getenv("LIVEKIT_API_SECRET"),
    )

    # Create an agent dispatch for the room
    dispatch = await lkapi.agent_dispatch.create_dispatch(
        api.CreateAgentDispatchRequest(
            agent_name=agent_name,
            room=room_name,
            metadata=f'{{"unique_id": "{unique_id}"}}'
        )
    )
    print("Created dispatch for inbound call:", dispatch)

    # List all dispatches in the room
    dispatches = await lkapi.agent_dispatch.list_dispatch(room_name=room_name)
    print(f"There are {len(dispatches)} dispatches in {room_name}")

    await lkapi.aclose()

# Simulate handling an inbound call
asyncio.run(handle_inbound_call())

inbound-call-abc123

generate_auth_token()

ctx.room.name

ctx.room.metadata

start_transcription

AgentSession

use_tts_aligned_transcript

user_away_timeout

max_tool_steps

on_participant_disconnected

on_room_disconnected

ctx.room.name

ctx.room.metadata

handle_inbound_call

sid

# Example in Python
room_sid = ctx.room.sid
print(f"Room SID: {room_sid}")

sid

class LangfuseTracker:
    def __init__(self):
        self.logs = []

    def log(self, node_type, input_data, output_data):
        log_entry = {
            "node_type": node_type,
            "input": input_data,
            "output": output_data
        }
        self.logs.append(log_entry)
        print(f"Logged {node_type}: {log_entry}")

from livekit.agents import Agent

class CustomAgent(Agent):
    def __init__(self, langfuse_tracker):
        super().__init__()
        self.langfuse_tracker = langfuse_tracker

    async def stt_node(self, audio, model_settings):
        async for event in super().stt_node(audio, model_settings):
            if hasattr(event, 'alternatives') and event.alternatives:
                input_audio = audio  # Log raw audio if needed
                output_text = event.alternatives[0].text
                self.langfuse_tracker.log("STT", input_audio, output_text)
            yield event

    async def llm_node(self, chat_ctx, tools, model_settings=None):
        async for chunk in super().llm_node(chat_ctx, tools, model_settings):
            input_text = chat_ctx.messages[-1].content  # Last user message
            output_text = chunk.delta.content if hasattr(chunk, 'delta') else str(chunk)
            self.langfuse_tracker.log("LLM", input_text, output_text)
            yield chunk

    async def tts_node(self, text, model_settings):
        async for audio_frame in super().tts_node(text, model_settings):
            input_text = text
            output_audio = audio_frame  # Log raw audio frames if needed
            self.langfuse_tracker.log("TTS", input_text, output_audio)
            yield audio_frame

langfuse_tracker = LangfuseTracker()
agent = CustomAgent(langfuse_tracker=langfuse_tracker)

LangfuseTracker

stt_node

llm_node

tts_node

LangfuseTracker

stt_node

llm_node

tts_node

class LangfuseTracker:
    def __init__(self):
        self.logs = []

    def log(self, node_type, input_data, output_data):
        log_entry = {
            "node_type": node_type,
            "input": input_data,
            "output": output_data
        }
        self.logs.append(log_entry)
        print(f"Logged {node_type}: {log_entry}")

from livekit.agents import Agent

class CustomAgent(Agent):
    def __init__(self, langfuse_tracker):
        super().__init__()
        self.langfuse_tracker = langfuse_tracker

    async def stt_node(self, audio, model_settings):
        async for event in super().stt_node(audio, model_settings):
            if hasattr(event, 'alternatives') and event.alternatives:
                input_audio = audio  # Log raw audio if needed
                output_text = event.alternatives[0].text
                self.langfuse_tracker.log("STT", input_audio, output_text)
            yield event

    async def llm_node(self, chat_ctx, tools, model_settings=None):
        async for chunk in super().llm_node(chat_ctx, tools, model_settings):
            input_text = chat_ctx.messages[-1].content  # Last user message
            output_text = chunk.delta.content if hasattr(chunk, 'delta') else str(chunk)
            self.langfuse_tracker.log("LLM", input_text, output_text)
            yield chunk

    async def tts_node(self, text, model_settings):
        async for audio_frame in super().tts_node(text, model_settings):
            input_text = text
            output_audio = audio_frame  # Log raw audio frames if needed
            self.langfuse_tracker.log("TTS", input_text, output_audio)
            yield audio_frame

import asyncio
from livekit import api

async def entrypoint(ctx):
    try:
        # Initialize Langfuse tracker
        langfuse_tracker = LangfuseTracker()

        # Create the custom agent
        agent = CustomAgent(langfuse_tracker=langfuse_tracker)

        # Set up the session
        session = AgentSession(
            use_tts_aligned_transcript=True,
            user_away_timeout=30,
            max_tool_steps=10
        )

        # Start the session
        await session.start(agent=agent, room=ctx.room, room_input_options=RoomInputOptions(close_on_disconnect=False))

        print("✅  Session started")

    except Exception as e:
        print(f"❌  Error: {e}")
        raise

stt_node

llm_node

tts_node

LangfuseTracker

LanguageSwitchingAgent

class TrackedLanguageSwitchingAgent(LanguageSwitchingAgent):
    def __init__(self, langfuse_tracker, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.langfuse_tracker = langfuse_tracker

    async def stt_node(self, audio, model_settings):
        async for event in super().stt_node(audio, model_settings):
            if hasattr(event, 'alternatives') and event.alternatives:
                input_audio = audio  # Log raw audio if needed
                output_text = event.alternatives[0].text
                self.langfuse_tracker.log("STT", input_audio, output_text)
            yield event

    async def llm_node(self, chat_ctx, tools, model_settings=None):
        async for chunk in super().llm_node(chat_ctx, tools, model_settings):
            input_text = chat_ctx.messages[-1].content  # Last user message
            output_text = chunk.delta.content if hasattr(chunk, 'delta') else str(chunk)
            self.langfuse_tracker.log("LLM", input_text, output_text)
            yield chunk

    async def tts_node(self, text, model_settings):
        async for audio_frame in super().tts_node(text, model_settings):
            input_text = text
            output_audio = audio_frame  # Log raw audio frames if needed
            self.langfuse_tracker.log("TTS", input_text, output_audio)
            yield audio_frame

async def entrypoint(ctx):
    try:
        # Initialize Langfuse tracker
        langfuse_tracker = LangfuseTracker()

        # Create the tracked language-switching agent
        agent = TrackedLanguageSwitchingAgent(
            langfuse_tracker=langfuse_tracker,
            # Pass other arguments required by your LanguageSwitchingAgent
            tools=[info_tool],
            llm_instance=llm,
            vad=vad
        )

        # Set up the session
        session = AgentSession(
            use_tts_aligned_transcript=True,
            user_away_timeout=30,
            max_tool_steps=10
        )

        # Start the session
        await session.start(agent=agent, room=ctx.room, room_input_options=RoomInputOptions(close_on_disconnect=False))

        print("✅  Session started")

    except Exception as e:
        print(f"❌  Error: {e}")
        raise