64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
"""FastAPI application factory for FusionAGI Dvādaśa API."""
|
|
|
|
from typing import Any
|
|
|
|
from fusionagi.api.dependencies import SessionStore, default_orchestrator, set_app_state
|
|
from fusionagi.api.routes import router as api_router
|
|
|
|
|
|
def create_app(
|
|
adapter: Any = None,
|
|
cors_origins: list[str] | None = None,
|
|
) -> Any:
|
|
"""Create FastAPI app with Dvādaśa routes.
|
|
|
|
Args:
|
|
adapter: Optional LLMAdapter for head/Witness LLM calls.
|
|
cors_origins: Optional list of CORS allowed origins (e.g. ["*"] or ["https://example.com"]).
|
|
If None, no CORS middleware is added.
|
|
"""
|
|
try:
|
|
from fastapi import FastAPI
|
|
except ImportError as e:
|
|
raise ImportError("Install with: pip install fusionagi[api]") from e
|
|
|
|
app = FastAPI(
|
|
title="FusionAGI Dvādaśa API",
|
|
description="12-headed multi-agent orchestration API",
|
|
version="0.1.0",
|
|
)
|
|
app.state.llm_adapter = adapter
|
|
from fusionagi.api.dependencies import set_default_adapter
|
|
set_default_adapter(adapter)
|
|
|
|
@app.on_event("startup")
|
|
async def startup():
|
|
"""Initialize orchestrator and session store."""
|
|
if getattr(app.state, "_dvadasa_ready", False):
|
|
return
|
|
adapter_inner = getattr(app.state, "llm_adapter", None)
|
|
orch, bus = default_orchestrator(adapter_inner)
|
|
store = SessionStore()
|
|
set_app_state(orch, bus, store)
|
|
app.state._dvadasa_ready = True
|
|
|
|
app.include_router(api_router, prefix="/v1", tags=["dvadasa"])
|
|
|
|
if cors_origins is not None:
|
|
try:
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=cors_origins,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
except ImportError:
|
|
pass # CORS optional
|
|
|
|
return app
|
|
|
|
|
|
# Default app instance for uvicorn/gunicorn
|
|
app = create_app()
|