on_participant_connected_handler

participant.attributes['sip.phoneNumber']

AgentSession

ctx.wait_for_participant()

session.start(ctx.room)

participant.attributes['sip.phoneNumber']

stt

tts

import fetch from 'node-fetch';
import { defineAgent } from '@livekit/agents';
import { ParticipantKind, TrackKind } from '@livekit/rtc-node';
import { voice, inference } from '@livekit/agents/voice'; // adjust imports per SDK

export default defineAgent({
  entry: async (ctx) => {
    // Connect to the room; auto-subscribe audio only
    await ctx.connect(undefined, { autoSubscribe: false });
    await ctx.waitForParticipant(); // ensure a participant will join if needed

    // Wait for the SIP participant (this returns the first participant)
    const participant = await ctx.waitForParticipant();
    if (participant.kind !== ParticipantKind.SIP) return;

    const phone = participant.attributes['sip.phoneNumber'];
    if (!phone) throw new Error('No SIP phone number found');

    // Fetch caller context from your external API
    const resp = await fetch(`<https://your-api.com/context?number=${encodeURIComponent(phone)}`>);
    const data = await resp.json();

    const sttModel = data.stt_model || 'deepgram/nova-2-general';
    const ttsModel = data.tts_model || 'cartesia/sonic-3';
    const voiceName = data.voice || 'default-voice';
    const language = data.language || 'en';

    // Construct AgentSession with dynamic STT/TTS
    const session = new voice.AgentSession({
      stt: new inference.STT({ model: sttModel, language }),
      tts: new inference.TTS({ model: ttsModel, voice: voiceName, language }),
      // add other options like VAD, turn detection if desired
    });

    // Start using the room from ctx
    await session.start(ctx.room);

    // Send an initial prompt (optional)
    await session.generateReply({ instructions: `Hello — I will use ${language} for speech.` });

    // Keep session alive and stream audio until participant disconnects
    const audioTrack = participant.tracks.get(TrackKind.AUDIO)?.track;
    if (audioTrack) {
      const audioStream = new ctx.rtc.AudioStream(audioTrack);
      for await (const _chunk of audioStream) { /* keep reading */ }
    }
  },
});

import aiohttp
import asyncio
from livekit import agents, rtc
from livekit.agents import JobContext, AgentSession
from livekit.agents import inference as inference_mod  # adjust per SDK layout

async def fetch_caller_context(phone_number: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(f'<https://your-api.com/context?number={phone_number}>') as resp:
            return await resp.json()

@agents.room.on('participant_connected')
async def on_participant_connected(room: rtc.Room, participant: rtc.Participant):
    # spawn a background task so handler returns quickly
    asyncio.create_task(handle_participant(room, participant))

async def handle_participant(room: rtc.Room, participant: rtc.Participant):
    if participant.kind != rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
        return

    phone = participant.attributes.get('sip.phoneNumber')
    if not phone:
        return

    try:
        ctx_data = await fetch_caller_context(phone)
        stt_model = ctx_data.get('stt_model', 'deepgram/nova-2-general')
        tts_model = ctx_data.get('tts_model', 'cartesia/sonic-3')
        voice_name = ctx_data.get('voice', 'default-voice')
        language = ctx_data.get('language', 'en')

        # Build AgentSession with dynamic parameters
        session = AgentSession(
            stt=inference_mod.STT(model=stt_model, language=language),
            tts=inference_mod.TTS(model=tts_model, voice=voice_name, language=language),
            # add vad/turn_detection if needed
        )

        # Start session bound to the room
        await session.start(room)

        await session.generate_reply(instructions=f"Hello — I'll use {language} for speech.")

        # Subscribe to audio to keep session alive (simple loop)
        audio_track = participant.tracks.get(rtc.TrackKind.KIND_AUDIO).track
        async for _ in rtc.AudioStream(audio_track):
            pass

    except Exception as e:
        # handle/log error
        print('Error handling participant:', e)

async def entrypoint(ctx: JobContext):
    # If using event handlers above, entrypoint can be minimal
    pass

if __name__ == '__main__':
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint, worker_type=agents.WorkerType.ROOM))

on_participant_connected

ctx.wait_for_participant()

AgentSession