SIPDispatchRuleCallee

from livekit import api
from livekit.protocol import sip

async def create_callee_dispatch_rule():
    # Initialize the LiveKit API client
    livekit_api = api.LiveKitAPI(
        url="your-livekit-url",
        api_key="your-api-key", 
        api_secret="your-api-secret"
    )
    
    try:
        # Create a SIPDispatchRuleCallee for automatic room creation
        dispatch_rule_info = sip.SIPDispatchRuleInfo(
            rule=sip.SIPDispatchRule(
                dispatch_rule_callee=sip.SIPDispatchRuleCallee(
                    room_prefix="number-",  # Optional prefix for room names
                    pin="",                 # Optional PIN protection
                    randomize=False         # Set to True for unique rooms per caller
                )
            ),
            name="Callee Dispatch Rule",
            metadata="Routes calls based on called number",
            trunk_ids=[],  # Empty means all trunks
            hide_phone_number=False
        )
        
        # Create the dispatch rule
        request = sip.CreateSIPDispatchRuleRequest(
            dispatch_rule=dispatch_rule_info
        )
        
        response = await livekit_api.sip.create_sip_dispatch_rule(request)
        print(f"Created dispatch rule: {response.sip_dispatch_rule_id}")
        
        return response
        
    finally:
        await livekit_api.aclose()

# Usage example
import asyncio

async def main():
    # Let's create that dispatch rule!
    result = await create_callee_dispatch_rule()
    print(f"Dispatch rule created with ID: {result.sip_dispatch_rule_id}")

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

True

+15551234567

room_prefix="number-"

number-+15551234567

SIPDispatchRuleCallee

import asyncio
from livekit import api

class CalleeRoomHandler:
    def __init__(self, api_key: str, api_secret: str, host: str):
        self.livekit_api = api.LiveKitAPI(host, api_key, api_secret)
    
    async def connect_to_callee_room(self, called_number: str, room_prefix: str = "number-"):
        """Connect to a room created by SIPDispatchRuleCallee"""
        
        room_name = f"{room_prefix}{called_number}"
        
        try:
            # First, let's check if the room exists
            rooms = await self.livekit_api.room.list_rooms(
                api.ListRoomsRequest()
            )
            
            room_exists = any(room.name == room_name for room in rooms.rooms)
            
            if room_exists:
                print(f"Found callee room: {room_name}")
                
                # Now, let's see who’s in the room
                participants = await self.livekit_api.room.list_participants(
                    api.ListParticipantsRequest(room=room_name)
                )
                
                print(f"Participants in {room_name}: {len(participants.participants)}")
                
                return room_name
            else:
                print(f"Room {room_name} does not exist yet")
                return None
                
        except Exception as e:
            print(f"Oops! There was an error connecting to the callee room: {e}")
            raise

# Here’s a quick example of how to use this
async def main():
    handler = CalleeRoomHandler(
        api_key="your_api_key",
        api_secret="your_api_secret", 
        host="your_livekit_host"
    )
    
    # Let’s connect to a room using the called number
    called_number = "+15105550123"
    await handler.connect_to_callee_room(called_number)

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

number-+15105550123

list_rooms

list_participants

SIPDispatchRuleCallee

SIPDispatchRuleCallee

import asyncio
from livekit import api, rtc
from typing import List

class SIPRoomManager:
    def __init__(self, room_prefix: str = "number-", randomize: bool = False):
        self.room_prefix = room_prefix
        self.randomize = randomize
        self.active_rooms = {}

    def get_room_name(self, phone_number: str) -> str:
        return f"{self.room_prefix}{phone_number}"

    async def connect_to_callee_rooms(self, phone_numbers: List[str], identity_prefix: str = "bot"):
        tasks = []
        for phone_number in phone_numbers:
            room_name = self.get_room_name(phone_number)
            identity = f"{identity_prefix}-{phone_number.replace('+', '').replace('-', '')}"
            task = asyncio.create_task(self.connect_and_handle_room(room_name, identity))
            tasks.append(task)
        await asyncio.gather(*tasks)

    async def connect_and_handle_room(self, room_name: str, identity: str):
        room = rtc.Room()

        @room.on("track_subscribed")
        def on_track_subscribed(track, publication, participant):
            print(f"Track subscribed in {room_name}: {publication.sid}")

        token = (
            api.AccessToken()
            .with_identity(identity)
            .with_grants(api.VideoGrants(room_join=True, room=room_name))
            .to_jwt()
        )

        try:
            await room.connect("your-livekit-url", token)
            print(f"Connected to room: {room_name}")
            self.active_rooms[room_name] = room
            await self.publish_tracks_to_room(room)
            await self.monitor_room(room, room_name)
        except Exception as e:
            print(f"Oops, failed to connect to room {room_name}: {e}")

    async def publish_tracks_to_room(self, room):
        video_source = rtc.VideoSource(width=1280, height=720)
        video_track = rtc.LocalVideoTrack.create_video_track("video", video_source)
        audio_source = rtc.AudioSource(sample_rate=48000, num_channels=2)
        audio_track = rtc.LocalAudioTrack.create_audio_track("audio", audio_source)

        await room.local_participant.publish_track(video_track)
        await room.local_participant.publish_track(audio_track)
        print(f"Published tracks to room: {room.name}")

    async def monitor_room(self, room, room_name):
        try:
            while room.isconnected():
                await asyncio.sleep(1)
        except Exception as e:
            print(f"Room {room_name} monitoring error: {e}")
        finally:
            if room_name in self.active_rooms:
                del self.active_rooms[room_name]

# Example usage
async def main():
    room_manager = SIPRoomManager(room_prefix="number-", randomize=True)
    phone_numbers = ["+15551234567", "+15559876543"]
    await room_manager.connect_to_callee_rooms(phone_numbers, "sip-bot")

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