Spaces:
Runtime error
Runtime error
| """ | |
| Local FastAPI server for Pipecat development. | |
| """ | |
| import asyncio | |
| import json | |
| import os | |
| import subprocess | |
| import sys | |
| import uuid | |
| import time | |
| from typing import Dict, Any, Optional | |
| import aiohttp | |
| from fastapi import FastAPI, HTTPException, WebSocket | |
| 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 pipecat.transports.daily.utils import ( | |
| DailyRESTHelper, | |
| DailyRoomParams, | |
| DailyRoomProperties, | |
| ) | |
| # Load environment variables | |
| load_dotenv() | |
| app = FastAPI(title="Local Pipecat Server", version="1.0.0") | |
| # Add CORS middleware | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| class BotStartRequest(BaseModel): | |
| createDailyRoom: bool = True | |
| dailyRoomProperties: Optional[Dict[str, Any]] = None | |
| body: Dict[str, Any] | |
| class BotStartResponse(BaseModel): | |
| dailyRoom: str | |
| dailyToken: str | |
| session_id: Optional[str] = None | |
| async def start_bot(request: BotStartRequest): | |
| """Start a bot process locally.""" | |
| try: | |
| session_id = str(uuid.uuid4()) | |
| logger.info(f"🚀 Starting bot session: {session_id}") | |
| # Get max session duration with priority: API > ENV > Default | |
| max_session_duration = get_max_session_duration(request.body) | |
| # Add max_session_duration to bot_data if not already present | |
| bot_data = request.body.copy() | |
| if "max_session_duration" not in bot_data: | |
| bot_data["max_session_duration"] = max_session_duration | |
| # Create Daily room and token | |
| room_url, token = await create_daily_room_and_token(session_id, request) | |
| # Start bot process with piped logs | |
| await start_bot_process(session_id, room_url, token, bot_data) | |
| logger.info(f"✅ Bot started successfully for session: {session_id}") | |
| return BotStartResponse( | |
| dailyRoom=room_url, dailyToken=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 create_daily_room_and_token( | |
| session_id: str, request: BotStartRequest | |
| ) -> tuple[str, str]: | |
| """Create a Daily room and meeting token.""" | |
| daily_api_key = os.getenv("DAILY_API_KEY") | |
| if not daily_api_key: | |
| raise HTTPException( | |
| status_code=500, detail="DAILY_API_KEY environment variable is required" | |
| ) | |
| # Get max session duration with priority: API > ENV > Default | |
| max_session_duration = get_max_session_duration(request.body) | |
| async with aiohttp.ClientSession() as http_session: | |
| daily_helper = DailyRESTHelper( | |
| daily_api_key=daily_api_key, aiohttp_session=http_session | |
| ) | |
| enable_recording = None | |
| recordings_bucket = None | |
| recordings_template = None | |
| if request.dailyRoomProperties: | |
| enable_recording = request.dailyRoomProperties.get( | |
| "enable_recording", False | |
| ) | |
| recordings_bucket = request.dailyRoomProperties.get( | |
| "recordings_bucket", None | |
| ) | |
| recordings_template = request.dailyRoomProperties.get( | |
| "recordings_template", None | |
| ).replace(" ", "_") | |
| room_properties = DailyRoomProperties( | |
| exp=int(time.time() + max_session_duration), | |
| enable_chat=False, | |
| enable_prejoin_ui=False, | |
| start_video_off=True, | |
| eject_at_room_exp=True, | |
| enable_recording=enable_recording, | |
| recordings_bucket=recordings_bucket, | |
| recordings_template=recordings_template, # type: ignore[attr-defined] | |
| dialout_config={"max_idle_timeout_sec": 0}, | |
| ) | |
| room_name = f"pipecat-local-{session_id[:8]}" | |
| room_params = DailyRoomParams( | |
| name=room_name, privacy="public", properties=room_properties | |
| ) | |
| logger.info(f"🏠 Creating Daily room: {room_name}") | |
| room = await daily_helper.create_room(room_params) | |
| logger.info(f"🎫 Creating meeting token for room: {room.name}") | |
| token = await daily_helper.get_token( | |
| room_url=room.url, expiry_time=3600, eject_at_token_exp=True, owner=True | |
| ) | |
| logger.info(f"✅ Daily room created: {room.url}") | |
| return room.url, token | |
| async def start_bot_process( | |
| session_id: str, room_url: str, token: str, bot_data: Dict[str, Any] | |
| ): | |
| """Start the bot.py process with piped stdout/stderr.""" | |
| try: | |
| # Get project root (parent of local/) | |
| project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| cmd = [ | |
| "uv", | |
| "run", | |
| "local/runner.py", | |
| "--session-id", | |
| session_id, | |
| "--room-url", | |
| room_url, | |
| "--token", | |
| token, | |
| "--bot-data", | |
| json.dumps(bot_data), | |
| ] | |
| logger.info(f"🔧 Starting bot process in {project_root}") | |
| logger.info(f"🔧 Command: {' '.join(cmd)}") | |
| # Start process with logs piped to this process | |
| process = subprocess.Popen( | |
| cmd, | |
| cwd=project_root, | |
| env=os.environ.copy(), | |
| stdout=subprocess.PIPE, | |
| stderr=subprocess.STDOUT, # Merge stderr into stdout | |
| text=True, | |
| bufsize=1, | |
| universal_newlines=True, | |
| ) | |
| logger.info(f"✅ Bot process started with PID: {process.pid}") | |
| # Start task to pipe logs | |
| 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 | |
| # Log subprocess output with session prefix | |
| logger.info(f"[{session_id[:8]}] {line.strip()}") | |
| except Exception as e: | |
| logger.error(f"❌ Error piping logs for {session_id}: {e}") | |
| # Twilio WebSocket endpoint for Media Streams | |
| async def twilio_websocket(websocket: WebSocket): | |
| """ | |
| Twilio Media Streams WebSocket endpoint. | |
| This handles incoming calls via Twilio's <Stream> TwiML directive. | |
| """ | |
| try: | |
| from bot import twilio_websocket_handler | |
| await twilio_websocket_handler(websocket) | |
| except ImportError as e: | |
| logger.error(f"❌ Failed to import bot: {e}") | |
| await websocket.close(code=1011, reason="Server configuration error") | |
| if __name__ == "__main__": | |
| host = "127.0.0.1" | |
| port = 8000 | |
| logger.info("🚀 Starting Local Pipecat Server") | |
| logger.info(f"📍 Server will run at: http://{host}:{port}") | |
| logger.info("📋 Available endpoints:") | |
| logger.info(" POST / - Start bot session (Daily.co/WebRTC)") | |
| logger.info(" WS /twilio - Twilio Media Streams WebSocket") | |
| uvicorn.run(app, host=host, port=port, log_level="info") | |