Service

DashboardHTTPService - extends BlackboardHTTPService with WebSocket support.

Provides real-time dashboard capabilities by: 1. Mounting WebSocket endpoint at /ws 2. Serving static files for dashboard frontend 3. Integrating DashboardEventCollector with WebSocketManager 4. Supporting CORS for development mode (DASHBOARD_DEV=1)

Classes

DashboardHTTPService

DashboardHTTPService(orchestrator: Flock, websocket_manager: WebSocketManager | None = None, event_collector: DashboardEventCollector | None = None, *, use_v2: bool = False)

Bases: BlackboardHTTPService

HTTP service with WebSocket support for real-time dashboard.

Extends BlackboardHTTPService to add: - WebSocket endpoint at /ws for real-time event streaming - Static file serving for dashboard frontend - Integration with DashboardEventCollector - Optional CORS middleware for development

Initialize DashboardHTTPService.

Parameters:

Name	Type	Description	Default
orchestrator	Flock	Flock orchestrator instance	required
websocket_manager	WebSocketManager | None	Optional WebSocketManager (creates new if not provided)	None
event_collector	DashboardEventCollector | None	Optional DashboardEventCollector (creates new if not provided)	None

Source code in src/flock/dashboard/service.py

def __init__(
    self,
    orchestrator: Flock,
    websocket_manager: WebSocketManager | None = None,
    event_collector: DashboardEventCollector | None = None,
    *,
    use_v2: bool = False,
) -> None:
    """Initialize DashboardHTTPService.

    Args:
        orchestrator: Flock orchestrator instance
        websocket_manager: Optional WebSocketManager (creates new if not provided)
        event_collector: Optional DashboardEventCollector (creates new if not provided)
    """
    # Initialize base service
    super().__init__(orchestrator)

    # Initialize WebSocket manager and event collector
    self.websocket_manager = websocket_manager or WebSocketManager()
    self.event_collector = event_collector or DashboardEventCollector(
        store=self.orchestrator.store
    )
    self.use_v2 = use_v2

    # Integrate collector with WebSocket manager
    self.event_collector.set_websocket_manager(self.websocket_manager)

    # Graph assembler powers both dashboards by default
    self.graph_assembler: GraphAssembler | None = GraphAssembler(
        self.orchestrator.store, self.event_collector, self.orchestrator
    )

    # Configure CORS if DASHBOARD_DEV environment variable is set
    if os.environ.get("DASHBOARD_DEV") == "1":
        logger.info("DASHBOARD_DEV mode enabled - adding CORS middleware")
        self.app.add_middleware(
            CORSMiddleware,
            allow_origins=["*"],  # Allow all origins in dev mode
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )

    # IMPORTANT: Register API routes BEFORE static files!
    # Static file mount acts as catch-all and must be last
    self._register_control_routes()
    self._register_theme_routes()
    self._register_dashboard_routes()

    logger.info("DashboardHTTPService initialized")

Functions

start async

start() -> None

Start the dashboard service.

Note: For testing purposes. In production, use uvicorn.run(app).

Source code in src/flock/dashboard/service.py

async def start(self) -> None:
    """Start the dashboard service.

    Note: For testing purposes. In production, use uvicorn.run(app).
    """
    logger.info("DashboardHTTPService started")
    # Start heartbeat if there are clients
    if len(self.websocket_manager.clients) > 0:
        await self.websocket_manager.start_heartbeat()

stop async

stop() -> None

Stop the dashboard service and clean up resources.

Closes all WebSocket connections gracefully.

Source code in src/flock/dashboard/service.py

async def stop(self) -> None:
    """Stop the dashboard service and clean up resources.

    Closes all WebSocket connections gracefully.
    """
    logger.info("Stopping DashboardHTTPService")
    await self.websocket_manager.shutdown()
    logger.info("DashboardHTTPService stopped")

get_app

get_app() -> Any

Get FastAPI application instance.

Returns:

Type	Description
Any	FastAPI app for testing or custom server setup

Source code in src/flock/dashboard/service.py

def get_app(self) -> Any:
    """Get FastAPI application instance.

    Returns:
        FastAPI app for testing or custom server setup
    """
    return self.app

Functions