"""FastAPI web server for Meeting Room — REST API, WebSocket, and SSE endpoints.""" from __future__ import annotations import asyncio import logging import uuid from contextlib import asynccontextmanager from typing import Any from fastapi import APIRouter, FastAPI, HTTPException, WebSocket, WebSocketDisconnect from fastapi.responses import RedirectResponse, StreamingResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from meeting_room.api_client import APIClient from meeting_room.config import load_config, load_discussion_config, find_workspace_config from meeting_room.engine import DiscussionEngine from meeting_room.events import AsyncEventBridge, EventEmitter from meeting_room.models import DiscussionConfig, ProviderConfig from meeting_room.scenario_loader import load_scenario from meeting_room.tools import ToolRegistry logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Pydantic models for API request/response # --------------------------------------------------------------------------- class StartDiscussionRequest(BaseModel): """Request body for starting a discussion.""" scenario: str # scenario file path or name config: str | None = None # optional config path override save: bool = False class InjectMessageRequest(BaseModel): """Request body for injecting a message.""" role_id: str content: str class SessionResponse(BaseModel): """Response for session creation.""" id: str status: str class SessionDetailResponse(BaseModel): """Response for session detail.""" id: str status: str scenario_name: str | None = None message_count: int = 0 # --------------------------------------------------------------------------- # Session store # --------------------------------------------------------------------------- class Session: """Holds state for a single discussion session.""" def __init__( self, session_id: str, config: dict, discussion_config: DiscussionConfig, ) -> None: self.id = session_id self.config = config self.discussion_config = discussion_config self.status: str = "created" # created | running | completed | error self.scenario_name: str | None = None self.messages: list[dict] = [] self.event_bus = EventEmitter() self.bridge = AsyncEventBridge(self.event_bus) self._engine_task: asyncio.Task | None = None def build_engine(self) -> DiscussionEngine: providers = { name: ProviderConfig(**vals) for name, vals in self.config.get("providers", {}).items() } api_client = APIClient(providers=providers) tool_registry = ToolRegistry(workdir=self.discussion_config.defaults.workdir) return DiscussionEngine( config=self.discussion_config, api_client=api_client, tool_registry=tool_registry, events=self.event_bus, ) class SessionStore: """In-memory session store.""" def __init__(self) -> None: self._sessions: dict[str, Session] = {} def create(self, config: dict, discussion_config: DiscussionConfig) -> Session: session_id = uuid.uuid4().hex[:12] session = Session(session_id, config, discussion_config) self._sessions[session_id] = session return session def get(self, session_id: str) -> Session | None: return self._sessions.get(session_id) def list_sessions(self) -> list[Session]: return list(self._sessions.values()) # --------------------------------------------------------------------------- # Application factory # --------------------------------------------------------------------------- store = SessionStore() @asynccontextmanager async def lifespan(app: FastAPI): """Startup/shutdown lifecycle.""" logger.info("Meeting Room server started") yield # Cleanup: cancel any running engine tasks for session in store.list_sessions(): if session._engine_task and not session._engine_task.done(): session._engine_task.cancel() logger.info("Meeting Room server stopped") app = FastAPI( title="Meeting Room", version="0.2.0", description="Multi-agent discussion framework — REST API and WebSocket", lifespan=lifespan, ) # Serve static files (web UI) import os _static_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static") app.mount("/static", StaticFiles(directory=_static_dir), name="static") @app.get("/") async def index(): """Redirect to the web UI.""" from fastapi.responses import RedirectResponse return RedirectResponse(url="/static/index.html") # --------------------------------------------------------------------------- # Helper # --------------------------------------------------------------------------- def _load_config_or_default(config_path: str | None) -> dict: """Load config from path, workspace, or package default.""" if config_path: return load_config(config_path) ws_config = find_workspace_config() if ws_config: return load_config(ws_config) return load_config() # --------------------------------------------------------------------------- # REST API (mounted at /api) # --------------------------------------------------------------------------- api = APIRouter(prefix="/api") @api.post("/sessions", response_model=SessionResponse) async def create_session(config: str | None = None): """Create a new discussion session.""" raw_config = _load_config_or_default(config) discussion_config = load_discussion_config() session = store.create(raw_config, discussion_config) return SessionResponse(id=session.id, status=session.status) @api.get("/sessions", response_model=list[SessionDetailResponse]) async def list_sessions(): """List all sessions.""" sessions = store.list_sessions() return [ SessionDetailResponse( id=s.id, status=s.status, scenario_name=s.scenario_name, message_count=len(s.messages), ) for s in sessions ] @api.get("/sessions/{session_id}", response_model=SessionDetailResponse) async def get_session(session_id: str): """Get session details.""" session = store.get(session_id) if not session: raise HTTPException(status_code=404, detail="Session not found") return SessionDetailResponse( id=session.id, status=session.status, scenario_name=session.scenario_name, message_count=len(session.messages), ) @api.post("/sessions/{session_id}/start", response_model=SessionDetailResponse) async def start_discussion(session_id: str, body: StartDiscussionRequest): """Start a discussion in a session.""" session = store.get(session_id) if not session: raise HTTPException(status_code=404, detail="Session not found") if session.status == "running": raise HTTPException(status_code=409, detail="Discussion already running") # Load scenario import os scenario_path = body.scenario if not os.path.isabs(scenario_path): scenario_path = os.path.join(os.getcwd(), scenario_path) if not os.path.exists(scenario_path): scenario_path = os.path.join( os.path.dirname(os.path.abspath(__file__)), "scenarios", body.scenario ) if not os.path.exists(scenario_path): raise HTTPException(status_code=404, detail=f"Scenario not found: {body.scenario}") scenario_data = load_scenario(scenario_path) session.scenario_name = scenario_data.get("name", "Untitled") # Rebuild config if overridden if body.config: raw_config = _load_config_or_default(body.config) discussion_config = load_discussion_config() session.config = raw_config session.discussion_config = discussion_config # Start the bridge loop = asyncio.get_event_loop() session.bridge.start(loop) # Subscribe a handler to collect messages from meeting_room.events import AGENT_MESSAGE, BOSS_TURN, AGENT_ERROR def collect_message(event): session.messages.append({ "role_id": event.data.get("role_id", ""), "name": event.data.get("name", ""), "content": event.data.get("content", event.data.get("error", "")), }) session.event_bus.on(AGENT_MESSAGE, collect_message) session.event_bus.on(BOSS_TURN, collect_message) session.event_bus.on(AGENT_ERROR, collect_message) # Run engine in a thread engine = session.build_engine() session.status = "running" async def run_engine(): try: await asyncio.to_thread(engine.run, scenario_data) session.status = "completed" except Exception as exc: logger.exception("Engine error in session %s", session.id) session.status = "error" finally: session.bridge.stop() session._engine_task = asyncio.create_task(run_engine()) return SessionDetailResponse( id=session.id, status=session.status, scenario_name=session.scenario_name, message_count=0, ) @api.post("/sessions/{session_id}/inject", response_model=dict) async def inject_message(session_id: str, body: InjectMessageRequest): """Inject a message from a Boss/CSO role.""" session = store.get(session_id) if not session: raise HTTPException(status_code=404, detail="Session not found") if session.status != "running": raise HTTPException(status_code=409, detail="Session is not running") engine = session.build_engine() msg = engine.inject_message(body.role_id, body.content) session.messages.append({ "role_id": msg.role_id, "name": msg.name, "content": msg.content, }) return {"role_id": msg.role_id, "name": msg.name, "content": msg.content} @api.get("/sessions/{session_id}/messages") async def get_messages(session_id: str): """Get all messages in a session.""" session = store.get(session_id) if not session: raise HTTPException(status_code=404, detail="Session not found") return session.messages @api.get("/roles") async def list_roles(config: str | None = None): """List available roles and model bindings.""" raw_config = _load_config_or_default(config) return raw_config.get("roles", {}) @api.get("/providers") async def list_providers(config: str | None = None): """List configured providers.""" raw_config = _load_config_or_default(config) return raw_config.get("providers", {}) # Mount API router app.include_router(api) # --------------------------------------------------------------------------- # WebSocket # --------------------------------------------------------------------------- @app.websocket("/ws/{session_id}") async def websocket_endpoint(websocket: WebSocket, session_id: str): """WebSocket endpoint for real-time event streaming and message injection.""" session = store.get(session_id) if not session: await websocket.close(code=4004, reason="Session not found") return await websocket.accept() queue = session.bridge.subscribe() try: # Task: forward events from queue to WebSocket async def send_events(): try: while True: event = await queue.get() await websocket.send_json(event) except asyncio.CancelledError: pass send_task = asyncio.create_task(send_events()) # Listen for incoming messages (inject) try: while True: data = await websocket.receive_json() if data.get("type") == "inject": role_id = data.get("role_id", "boss") content = data.get("content", "") engine = session.build_engine() msg = engine.inject_message(role_id, content) session.messages.append({ "role_id": msg.role_id, "name": msg.name, "content": msg.content, }) except WebSocketDisconnect: pass finally: send_task.cancel() session.bridge.unsubscribe(queue) except Exception: session.bridge.unsubscribe(queue) # --------------------------------------------------------------------------- # SSE # --------------------------------------------------------------------------- @app.get("/events/{session_id}") async def sse_endpoint(session_id: str): """SSE endpoint for read-only event streaming.""" session = store.get(session_id) if not session: raise HTTPException(status_code=404, detail="Session not found") async def event_generator(): queue = session.bridge.subscribe() try: while True: event = await queue.get() # SSE format import json yield f"data: {json.dumps(event)}\n\n" except asyncio.CancelledError: pass finally: session.bridge.unsubscribe(queue) return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", }, )