Spaces:
Runtime error
Runtime error
| """ | |
| 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 | |
| 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)}") | |
| 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} | |
| 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)) | |
| 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)) | |
| async def health_check(): | |
| """Health check endpoint.""" | |
| return { | |
| "status": "healthy", | |
| "transport": "livekit", | |
| "active_sessions": len(active_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") | |