webapp/backend/ — FastAPI REST Backend

What is this & why does it exist?

The backend translates HTTP requests into agent pipeline executions. When a user clicks “Run” in the web UI, the backend receives the React Flow Orchestration" href="../../core_doc/graph/index.html">graph definition (nodes + edges), builds a LangGraph StateGraph from it using create_graph(), compiles and invokes it, and streams progress events (Start → Progress → Complete/Error) back to the browser via Server-Sent Events. The repository pattern abstracts SQLite access for flow persistence, and Alembic handles schema migrations.

Why Is It Designed This Way?

SSE was chosen over WebSockets because the communication is unidirectional (server → client) and SSE is simpler to implement with FastAPI’s StreamingResponse. The create_graph() function dynamically builds a StateGraph from React Flow node/edge definitions rather than hardcoding the pipeline because the web UI allows users to compose arbitrary agent topologies. The repository pattern decouples database access from route handlers, making it easy to swap SQLite for PostgreSQL. The extract_base_agent_key() function strips random suffixes from React Flow node IDs to match against the canonical ANALYST_CONFIG dictionary.

SSE Streaming Endpoint (routes/hedge_fund.py)

async def run(request_data: HedgeFundRequest, request: Request,
              db: Session = Depends(get_db)):
    # WHY: Hydrate API keys from DB when frontend doesn't send them
    if not request_data.api_keys:
        api_key_service = ApiKeyService(db)
        request_data.api_keys = api_key_service.get_api_keys_dict()

    portfolio = create_portfolio(request_data.initial_cash, ...)

    # WHY: Build graph from React Flow nodes/edges - not hardcoded
    graph = create_graph(
        graph_nodes=request_data.graph_nodes,
        graph_edges=request_data.graph_edges
    )
    graph = graph.compile()

    async def event_generator():
        progress_queue = asyncio.Queue()

        # WHY: Progress handler bridges sync agent code to async SSE stream
        def progress_handler(agent_name, ticker, status, analysis, timestamp):
            event = ProgressUpdateEvent(agent=agent_name, ticker=ticker,
                                        status=status, timestamp=timestamp)
            progress_queue.put_nowait(event)

        progress.register_handler(progress_handler)

        # WHY: Run graph in background task so we can yield SSE events concurrently
        run_task = asyncio.create_task(run_graph_async(graph=graph, ...))
        disconnect_task = asyncio.create_task(wait_for_disconnect())

        yield StartEvent().to_sse()

        while not run_task.done():
            # WHY: Detect client disconnect to cancel wasted work
            if disconnect_task.done():
                run_task.cancel()
                return
            try:
                event = await asyncio.wait_for(progress_queue.get(), timeout=1.0)
                yield event.to_sse()
            except asyncio.TimeoutError:
                pass  # Keep polling

        result = await run_task
        yield CompleteEvent(decisions=result["decisions"], ...).to_sse()

    return StreamingResponse(event_generator(), media_type="text/event-stream")