Spaces:
Runtime error
Runtime error
| """ | |
| LiveKit Cloud utilities. | |
| Provides helper functions for room and token management via LiveKit Cloud API. | |
| """ | |
| import os | |
| from datetime import timedelta | |
| from typing import Optional | |
| from loguru import logger | |
| from livekit.api import LiveKitAPI, AccessToken, VideoGrants, CreateRoomRequest | |
| class LiveKitHelper: | |
| """Helper class for LiveKit Cloud room and token management.""" | |
| def __init__( | |
| self, | |
| api_key: Optional[str] = None, | |
| api_secret: Optional[str] = None, | |
| url: Optional[str] = None, | |
| ): | |
| """ | |
| Initialize LiveKit helper. | |
| Args: | |
| api_key: LiveKit API key (defaults to LIVEKIT_API_KEY env var) | |
| api_secret: LiveKit API secret (defaults to LIVEKIT_API_SECRET env var) | |
| url: LiveKit Cloud URL (defaults to LIVEKIT_URL env var) | |
| """ | |
| self.api_key = api_key or os.getenv("LIVEKIT_API_KEY") | |
| self.api_secret = api_secret or os.getenv("LIVEKIT_API_SECRET") | |
| self.url = url or os.getenv("LIVEKIT_URL") | |
| if not self.api_key or not self.api_secret: | |
| raise ValueError( | |
| "LiveKit API key and secret are required. " | |
| "Set LIVEKIT_API_KEY and LIVEKIT_API_SECRET environment variables." | |
| ) | |
| if not self.url: | |
| raise ValueError( | |
| "LiveKit URL is required. Set LIVEKIT_URL environment variable." | |
| ) | |
| self._api: Optional[LiveKitAPI] = None | |
| def api(self) -> LiveKitAPI: | |
| """Get or create LiveKit API client.""" | |
| if self._api is None: | |
| self._api = LiveKitAPI( | |
| url=self.url, | |
| api_key=self.api_key, | |
| api_secret=self.api_secret, | |
| ) | |
| return self._api | |
| def create_token( | |
| self, | |
| room_name: str, | |
| participant_identity: str, | |
| ttl: int = 3600, | |
| can_publish: bool = True, | |
| can_subscribe: bool = True, | |
| can_publish_data: bool = True, | |
| ) -> str: | |
| """ | |
| Create an access token for a participant. | |
| Args: | |
| room_name: Name of the room to join | |
| participant_identity: Unique identity for the participant | |
| ttl: Token time-to-live in seconds (default: 1 hour) | |
| can_publish: Allow publishing audio/video | |
| can_subscribe: Allow subscribing to other participants | |
| can_publish_data: Allow publishing data messages | |
| Returns: | |
| JWT access token string | |
| """ | |
| token = ( | |
| AccessToken(api_key=self.api_key, api_secret=self.api_secret) | |
| .with_identity(participant_identity) | |
| .with_ttl(timedelta(seconds=ttl)) | |
| .with_grants( | |
| VideoGrants( | |
| room_join=True, | |
| room=room_name, | |
| can_publish=can_publish, | |
| can_subscribe=can_subscribe, | |
| can_publish_data=can_publish_data, | |
| ) | |
| ) | |
| ) | |
| return token.to_jwt() | |
| async def create_room( | |
| self, | |
| room_name: str, | |
| empty_timeout: int = 300, | |
| max_participants: int = 10, | |
| ) -> dict: | |
| """ | |
| Create a new LiveKit room. | |
| Args: | |
| room_name: Unique name for the room | |
| empty_timeout: Seconds to keep room alive when empty (default: 5 min) | |
| max_participants: Maximum number of participants | |
| Returns: | |
| Room info dictionary | |
| """ | |
| room = await self.api.room.create_room( | |
| CreateRoomRequest( | |
| name=room_name, | |
| empty_timeout=empty_timeout, | |
| max_participants=max_participants, | |
| ) | |
| ) | |
| logger.info(f"Created LiveKit room: {room_name}") | |
| return { | |
| "name": room.name, | |
| "sid": room.sid, | |
| "empty_timeout": room.empty_timeout, | |
| "max_participants": room.max_participants, | |
| "creation_time": room.creation_time, | |
| } | |
| async def delete_room(self, room_name: str) -> bool: | |
| """ | |
| Delete a LiveKit room. | |
| Args: | |
| room_name: Name of the room to delete | |
| Returns: | |
| True if successful | |
| """ | |
| await self.api.room.delete_room(room_name) | |
| logger.info(f"Deleted LiveKit room: {room_name}") | |
| return True | |
| async def list_rooms(self) -> list: | |
| """ | |
| List all active rooms. | |
| Returns: | |
| List of room info dictionaries | |
| """ | |
| rooms = await self.api.room.list_rooms() | |
| return [ | |
| { | |
| "name": room.name, | |
| "sid": room.sid, | |
| "num_participants": room.num_participants, | |
| "creation_time": room.creation_time, | |
| } | |
| for room in rooms | |
| ] | |
| async def list_participants(self, room_name: str) -> list: | |
| """ | |
| List participants in a room. | |
| Args: | |
| room_name: Name of the room | |
| Returns: | |
| List of participant info dictionaries | |
| """ | |
| participants = await self.api.room.list_participants(room_name) | |
| return [ | |
| { | |
| "identity": p.identity, | |
| "sid": p.sid, | |
| "state": p.state, | |
| "joined_at": p.joined_at, | |
| } | |
| for p in participants | |
| ] | |
| async def remove_participant(self, room_name: str, identity: str) -> bool: | |
| """ | |
| Remove a participant from a room. | |
| Args: | |
| room_name: Name of the room | |
| identity: Participant identity to remove | |
| Returns: | |
| True if successful | |
| """ | |
| await self.api.room.remove_participant(room_name, identity) | |
| logger.info(f"Removed participant {identity} from room {room_name}") | |
| return True | |
| async def close(self): | |
| """Close the API client.""" | |
| if self._api: | |
| await self._api.aclose() | |
| self._api = None | |
| # Singleton instance for convenience | |
| _helper: Optional[LiveKitHelper] = None | |
| def get_livekit_helper() -> LiveKitHelper: | |
| """Get or create the global LiveKit helper instance.""" | |
| global _helper | |
| if _helper is None: | |
| _helper = LiveKitHelper() | |
| return _helper | |