#!/usr/bin/env -S uv run --script
# /// script
# dependencies = [
# "fastmcp>=2.0.0",
# "livekit-api",
# ]
# requires-python = ">=3.9"
# ///
"""
LiveKit Admin MCP Server - Create and manage LiveKit rooms
"""
import logging
import os
from typing import Optional, List
from fastmcp import FastMCP
from livekit import api
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("livekit-admin-mcp")
# Create MCP server
mcp = FastMCP("LiveKit Admin")
# Configuration
LIVEKIT_URL = os.getenv("LIVEKIT_URL", "ws://127.0.0.1:7880")
LIVEKIT_API_KEY = os.getenv("LIVEKIT_API_KEY", "devkey")
LIVEKIT_API_SECRET = os.getenv("LIVEKIT_API_SECRET", "secret")
LIVEKIT_API_URL = LIVEKIT_URL.replace("ws://", "http://").replace("wss://", "https://")
def get_api_client():
"""Get LiveKit API client"""
return api.LiveKitAPI(LIVEKIT_API_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET)
@mcp.tool()
async def create_room(room_name: str, max_participants: int = 10, empty_timeout: int = 300) -> str:
"""Create a new LiveKit room"""
try:
lk_api = get_api_client()
room_config = api.CreateRoomRequest(
name=room_name,
empty_timeout=empty_timeout,
max_participants=max_participants,
)
room = await lk_api.room.create_room(room_config)
logger.info(f"Created room: {room_name}")
return f"Successfully created room '{room_name}' with max {max_participants} participants"
except Exception as e:
error_msg = f"Error creating room '{room_name}': {str(e)}"
logger.error(error_msg)
return error_msg
@mcp.tool()
async def list_rooms() -> str:
"""List all active LiveKit rooms"""
try:
lk_api = get_api_client()
rooms_response = await lk_api.room.list_rooms(api.ListRoomsRequest())
if not rooms_response.rooms:
return "No active rooms found"
result = f"Active rooms ({len(rooms_response.rooms)}):\n"
for room in rooms_response.rooms:
result += f" • {room.name}\n"
result += f" - Participants: {room.num_participants}\n"
result += f" - Created: {room.creation_time}\n"
result += f" - Max participants: {room.max_participants}\n"
result += "\n"
return result
except Exception as e:
error_msg = f"Error listing rooms: {str(e)}"
logger.error(error_msg)
return error_msg
@mcp.tool()
async def get_room_info(room_name: str) -> str:
"""Get detailed information about a specific room"""
try:
lk_api = get_api_client()
# Get room list to find our room
rooms_response = await lk_api.room.list_rooms(api.ListRoomsRequest())
target_room = None
for room in rooms_response.rooms:
if room.name == room_name:
target_room = room
break
if not target_room:
return f"Room '{room_name}' not found"
# Get participants
participants = await lk_api.room.list_participants(
api.ListParticipantsRequest(room=room_name)
)
result = f"Room: {room_name}\n"
result += f"Participants: {len(participants.participants)}/{target_room.max_participants}\n"
result += f"Created: {target_room.creation_time}\n"
result += f"Empty timeout: {target_room.empty_timeout}s\n\n"
if participants.participants:
result += "Participants:\n"
for p in participants.participants:
result += f" • {p.identity} ({p.name or 'unnamed'})\n"
result += f" - State: {p.state}\n"
result += f" - Joined: {p.joined_at}\n"
if p.tracks:
result += f" - Tracks: {len(p.tracks)}\n"
result += "\n"
else:
result += "No participants\n"
return result
except Exception as e:
error_msg = f"Error getting room info for '{room_name}': {str(e)}"
logger.error(error_msg)
return error_msg
@mcp.tool()
async def delete_room(room_name: str) -> str:
"""Delete a LiveKit room"""
try:
lk_api = get_api_client()
await lk_api.room.delete_room(api.DeleteRoomRequest(room=room_name))
logger.info(f"Deleted room: {room_name}")
return f"Successfully deleted room '{room_name}'"
except Exception as e:
error_msg = f"Error deleting room '{room_name}': {str(e)}"
logger.error(error_msg)
return error_msg
@mcp.tool()
async def create_participant_token(room_name: str, identity: str, participant_name: str = "") -> str:
"""Create an access token for a participant to join a room"""
try:
# Create access token
token = api.AccessToken(LIVEKIT_API_KEY, LIVEKIT_API_SECRET)
token.with_identity(identity)
if participant_name:
token.with_name(participant_name)
token.with_grants(api.VideoGrants(
room_join=True,
room=room_name,
can_publish=True,
can_subscribe=True,
))
jwt_token = token.to_jwt()
logger.info(f"Created token for {identity} in room {room_name}")
result = f"Access token created for '{identity}' in room '{room_name}'\n"
result += f"Token: {jwt_token}\n\n"
result += f"Connection details:\n"
result += f" URL: {LIVEKIT_URL}\n"
result += f" Room: {room_name}\n"
result += f" Identity: {identity}\n"
return result
except Exception as e:
error_msg = f"Error creating token: {str(e)}"
logger.error(error_msg)
return error_msg
@mcp.tool()
async def remove_participant(room_name: str, identity: str) -> str:
"""Remove a participant from a room"""
try:
lk_api = get_api_client()
await lk_api.room.remove_participant(
api.RoomParticipantIdentity(room=room_name, identity=identity)
)
logger.info(f"Removed participant {identity} from room {room_name}")
return f"Successfully removed '{identity}' from room '{room_name}'"
except Exception as e:
error_msg = f"Error removing participant '{identity}' from room '{room_name}': {str(e)}"
logger.error(error_msg)
return error_msg
# Entry point
if __name__ == "__main__":
logger.info("Starting LiveKit Admin MCP Server...")
logger.info(f"LiveKit URL: {LIVEKIT_URL}")
logger.info("")
logger.info("Available tools:")
logger.info(" - create_room(room_name, max_participants=10, empty_timeout=300)")
logger.info(" - list_rooms()")
logger.info(" - get_room_info(room_name)")
logger.info(" - delete_room(room_name)")
logger.info(" - create_participant_token(room_name, identity, participant_name='')")
logger.info(" - remove_participant(room_name, identity)")
mcp.run()
