"""FastAPI application entry point."""
import asyncio
from contextlib import asynccontextmanager
from pathlib import Path
import logging
import structlog
from fastapi import Depends, FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from navi.api.deps import require_admin
from navi.api.routes import agents, api_tokens, auth, health, messages, sessions, webhooks
from navi.api.routes.admin import router as admin_router
from navi.api.websocket import router as ws_router
from navi.config import settings
from navi.core.container import create_container
from debug.eval.api import router as eval_router
structlog.configure(
wrapper_class=structlog.make_filtering_bound_logger(
getattr(logging, settings.log_level)
),
)
# Suppress noisy MCP SDK health-check chatter.
logging.getLogger("mcp.server.lowlevel.server").setLevel(logging.WARNING)
_base = Path(__file__).parent.parent
async def _wait_for_ui_server(host: str, port: int, timeout: float = 5.0) -> bool:
"""Poll until the internal UI MCP server is accepting TCP connections."""
deadline = asyncio.get_event_loop().time() + timeout
while asyncio.get_event_loop().time() < deadline:
try:
reader, writer = await asyncio.wait_for(
asyncio.open_connection(host, port), timeout=0.5
)
writer.close()
try:
await writer.wait_closed()
except Exception:
pass
return True
except Exception:
await asyncio.sleep(0.1)
return False
@asynccontextmanager
async def lifespan(app: FastAPI):
log = structlog.get_logger()
# Fail fast: credentialed CORS needs an explicit origin list — browsers reject
# "*" together with credentials, and an implicit empty list would silently
# lock out (or leave open) cookie-based access.
if settings.navi_auth_enabled and not settings.navi_allowed_origins_list:
raise RuntimeError(
"NAVI_ALLOWED_ORIGINS must be set (comma-separated list) when NAVI_AUTH_ENABLED=true. "
"Refusing to start with permissive CORS in auth mode."
)
ui_server_task: asyncio.Task | None = None
if settings.navi_ui_mcp_enabled:
from navi.mcp.ui_server import start_ui_server
ui_server_task = asyncio.create_task(
start_ui_server(settings.navi_ui_mcp_host, settings.navi_ui_mcp_port)
)
ready = await _wait_for_ui_server(
settings.navi_ui_mcp_host, settings.navi_ui_mcp_port
)
if ready:
log.info(
"startup.navi_ui_mcp_ready",
host=settings.navi_ui_mcp_host,
port=settings.navi_ui_mcp_port,
)
else:
log.warning(
"startup.navi_ui_mcp_not_ready",
host=settings.navi_ui_mcp_host,
port=settings.navi_ui_mcp_port,
)
container = await create_container()
app.state.container = container
# The navi_ui MCP server needs no orchestrator/session-store wiring: it
# returns component metadata in the tool result, and the webclient renders
# from the role="tool" message like content_publish does.
from navi.api.deps import set_container
set_container(container)
if not settings.navi_auth_enabled:
log.warning(
"startup.auth_disabled",
message="Authorization is disabled — the server is open to anyone with network access. "
"Use NAVI_AUTH_ENABLED=false only for trusted single-user/local deployments.",
)
from navi.content_store import ensure_tables
from navi.session_files import cleanup_loop
from navi.auth import _ensure_auth_tables
from navi.profiles._overrides import ensure_table, load_overrides
from navi.api.routes.health import _check_embed
from navi.core.scheduler import recall_scheduler_loop
from navi.core.pg_session_store import pending_sweep_loop
# Ensure auth tables first (navi_users is referenced by other DDL).
for attempt in range(1, 6):
try:
await _ensure_auth_tables()
await ensure_tables()
break
except Exception as e:
if attempt < 5:
log.warning("startup.ensure_tables_retry", attempt=attempt, error=str(e))
await asyncio.sleep(2)
else:
log.error("startup.ensure_tables_failed", error=str(e))
# Apply persisted profile overrides
try:
pool = await container.database.pool()
await ensure_table(pool)
overrides = await load_overrides(pool)
if overrides:
for pid, is_admin_only in overrides.items():
try:
profile = container.profile_registry.get(pid)
profile.is_admin_only = is_admin_only
except Exception:
pass
log.info("startup.profile_overrides_applied", count=len(overrides))
except Exception:
log.warning("startup.profile_overrides_failed", exc_info=True)
# Check embedding backend health
embed_status = await _check_embed()
if embed_status["ok"]:
log.info("startup.embed_ready", backend=embed_status["backend"])
else:
log.warning("startup.embed_unavailable", backend=embed_status["backend"], error=embed_status["error"])
# Start background tasks
cleanup_task = asyncio.create_task(cleanup_loop(container.session_store))
scheduler_task = asyncio.create_task(
recall_scheduler_loop(container.scheduler, container.session_store, container.orchestrator)
)
pending_sweep_task = asyncio.create_task(pending_sweep_loop(container.session_store))
yield
# Shutdown
scheduler_task.cancel()
try:
await scheduler_task
except asyncio.CancelledError:
pass
cleanup_task.cancel()
try:
await cleanup_task
except asyncio.CancelledError:
pass
pending_sweep_task.cancel()
try:
await pending_sweep_task
except asyncio.CancelledError:
pass
if settings.navi_ui_mcp_enabled and ui_server_task is not None:
ui_server_task.cancel()
try:
await ui_server_task
except asyncio.CancelledError:
pass
from navi.tools.ssh_exec import close_all_connections
close_all_connections()
await container.shutdown()
app = FastAPI(
title="Navi",
description="Modular agent system — REST API and WebSocket",
version="0.1.0",
lifespan=lifespan,
)
# Credentialed CORS requires an explicit origin list (browsers refuse "*"
# together with credentials). In no-auth mode there are no cookies to send,
# so a permissive policy is acceptable there.
if settings.navi_auth_enabled:
_cors_origins = settings.navi_allowed_origins_list
_cors_credentials = True
else:
_cors_origins = ["*"]
_cors_credentials = False
app.add_middleware(
CORSMiddleware,
allow_origins=_cors_origins,
allow_credentials=_cors_credentials,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(health.router)
app.include_router(auth.router)
app.include_router(api_tokens.router)
app.include_router(agents.router)
app.include_router(sessions.router)
app.include_router(messages.router)
app.include_router(ws_router)
app.include_router(webhooks.router)
app.include_router(admin_router)
# Eval endpoints spend LLM tokens and read session data — admin only.
# (With auth disabled require_admin resolves to the anonymous admin user.)
app.include_router(eval_router, dependencies=[Depends(require_admin)])
app.mount("/assets", StaticFiles(directory=str(_base / "webclient" / "dist" / "assets")), name="assets")
app.mount("/images", StaticFiles(directory=str(_base / "webclient" / "dist" / "images")), name="images")
app.mount("/content-viewers", StaticFiles(directory=str(_base / "webclient" / "dist" / "content-viewers")), name="content_viewers")
app.mount("/content", StaticFiles(directory=str(_base / "navi" / "content")), name="content")
@app.get("/api/sessions/{session_id}/files/{filename}", include_in_schema=False)
async def api_session_file_redirect(session_id: str, filename: str, download: bool = False):
target = f"/sessions/{session_id}/files/{filename}"
if download:
target += "?download=1"
return RedirectResponse(url=target, status_code=307)
@app.get("/", include_in_schema=False)
async def index() -> FileResponse:
return FileResponse(str(_base / "webclient" / "dist" / "index.html"), headers={"Cache-Control": "no-store"})
if not settings.navi_auth_enabled:
# Debug panels are a local-development tool — not exposed in auth mode.
@app.get("/debug", include_in_schema=False)
async def debug() -> FileResponse:
return FileResponse("debug/index.html", headers={"Cache-Control": "no-store"})
@app.get("/debug/eval", include_in_schema=False)
async def debug_eval() -> FileResponse:
return FileResponse("debug/eval/index.html", headers={"Cache-Control": "no-store"})
@app.get("/admin", include_in_schema=False)
async def admin_panel() -> FileResponse:
return FileResponse("admin/index.html", headers={"Cache-Control": "no-store"})
