""" LiveKit FastAPI server for Pipecat. Provides HTTP endpoints for managing LiveKit-based voice AI sessions. """ import asyncio import json import os import subprocess import sys import uuid from typing import Dict, Any, Optional from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import uvicorn from loguru import logger from dotenv import load_dotenv # Add parent directory for imports sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from utils.session import get_max_session_duration from livekit_server.utils import LiveKitHelper # Load environment variables load_dotenv() app = FastAPI(title="Pipecat LiveKit Server", version="1.0.0") # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Store active bot processes active_bots: Dict[str, subprocess.Popen] = {} class BotStartRequest(BaseModel): """Request to start a new bot session.""" room_name: Optional[str] = None # Auto-generated if not provided body: Dict[str, Any] # Bot configuration (flow, config, metadata) class BotStartResponse(BaseModel): """Response with session details.""" room_name: str livekit_url: str participant_token: str # Token for the user/client to join bot_token: str # Token for the bot (internal use) session_id: str class RoomListResponse(BaseModel): """Response with list of active rooms.""" rooms: list @app.post("/", response_model=BotStartResponse) @app.post("/start", response_model=BotStartResponse) async def start_bot(request: BotStartRequest): """ Start a new bot session with LiveKit. Creates a LiveKit room, generates tokens, and starts the bot process. """ try: session_id = str(uuid.uuid4()) room_name = request.room_name or f"pipecat-{session_id[:8]}" logger.info(f"Starting LiveKit bot session: {session_id}") # Get max session duration max_session_duration = get_max_session_duration(request.body) # Add max_session_duration to bot_data bot_data = request.body.copy() if "max_session_duration" not in bot_data: bot_data["max_session_duration"] = max_session_duration # Initialize LiveKit helper livekit = LiveKitHelper() # Create room await livekit.create_room( room_name=room_name, empty_timeout=max_session_duration + 60, # Room stays alive a bit longer max_participants=2, # Bot + user ) # Create tokens bot_token = livekit.create_token( room_name=room_name, participant_identity="bot", ttl=max_session_duration + 300, can_publish=True, can_subscribe=True, ) participant_token = livekit.create_token( room_name=room_name, participant_identity=f"user-{session_id[:8]}", ttl=max_session_duration + 300, can_publish=True, can_subscribe=True, ) # Start bot process await start_bot_process( session_id=session_id, room_name=room_name, livekit_url=livekit.url, bot_token=bot_token, bot_data=bot_data, ) logger.info(f"Bot started successfully: {session_id} in room {room_name}") return BotStartResponse( room_name=room_name, livekit_url=livekit.url, participant_token=participant_token, bot_token=bot_token, session_id=session_id, ) except Exception as e: logger.error(f"Error starting bot: {e}") raise HTTPException(status_code=500, detail=f"Failed to start bot: {str(e)}") @app.delete("/bot/{session_id}") async def stop_bot(session_id: str): """Stop an active bot session.""" if session_id not in active_bots: raise HTTPException(status_code=404, detail="Bot session not found") process = active_bots[session_id] process.terminate() del active_bots[session_id] logger.info(f"Bot stopped: {session_id}") return {"status": "stopped", "session_id": session_id} @app.get("/rooms", response_model=RoomListResponse) async def list_rooms(): """List all active LiveKit rooms.""" try: livekit = LiveKitHelper() rooms = await livekit.list_rooms() return RoomListResponse(rooms=rooms) except Exception as e: logger.error(f"Error listing rooms: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.delete("/rooms/{room_name}") async def delete_room(room_name: str): """Delete a LiveKit room.""" try: livekit = LiveKitHelper() await livekit.delete_room(room_name) return {"status": "deleted", "room_name": room_name} except Exception as e: logger.error(f"Error deleting room: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health_check(): """Health check endpoint.""" return { "status": "healthy", "transport": "livekit", "active_sessions": len(active_bots), } @app.get("/bots") async def list_bots(): """List all active bot sessions.""" return { "bots": [ {"session_id": sid, "pid": proc.pid} for sid, proc in active_bots.items() if proc.poll() is None # Only running processes ] } async def start_bot_process( session_id: str, room_name: str, livekit_url: str, bot_token: str, bot_data: Dict[str, Any], ): """Start the bot runner process.""" try: project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) cmd = [ "uv", "run", "livekit_server/runner.py", "--session-id", session_id, "--room-name", room_name, "--livekit-url", livekit_url, "--token", bot_token, "--bot-data", json.dumps(bot_data), ] logger.info(f"Starting bot process: {' '.join(cmd[:6])}...") process = subprocess.Popen( cmd, cwd=project_root, env=os.environ.copy(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, universal_newlines=True, ) active_bots[session_id] = process logger.info(f"Bot process started with PID: {process.pid}") # Start log piping task asyncio.create_task(pipe_logs(process, session_id)) except Exception as e: logger.error(f"Failed to start bot process: {e}") raise async def pipe_logs(process: subprocess.Popen, session_id: str): """Pipe subprocess logs to main logger.""" try: while True: line = process.stdout.readline() if not line: if process.poll() is not None: break await asyncio.sleep(0.1) continue logger.info(f"[{session_id[:8]}] {line.strip()}") except Exception as e: logger.error(f"Error piping logs for {session_id}: {e}") finally: # Clean up when process ends if session_id in active_bots: del active_bots[session_id] logger.info(f"Bot session ended: {session_id}") if __name__ == "__main__": host = os.getenv("HOST", "0.0.0.0") port = int(os.getenv("PORT", "8000")) logger.info("Starting Pipecat LiveKit Server") logger.info(f"Server will run at: http://{host}:{port}") logger.info("Available endpoints:") logger.info(" POST /start - Start bot session") logger.info(" DELETE /bot/{id} - Stop bot session") logger.info(" GET /rooms - List active rooms") logger.info(" DELETE /rooms/{n} - Delete room") logger.info(" GET /health - Health check") logger.info(" GET /bots - List active bots") uvicorn.run(app, host=host, port=port, log_level="info")