""" 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 @app.post("/", response_model=BotStartResponse) 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 @app.websocket("/twilio") async def twilio_websocket(websocket: WebSocket): """ Twilio Media Streams WebSocket endpoint. This handles incoming calls via Twilio's 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")