def create_app(
engine,
model: str,
*,
agent=None,
bus=None,
engine_name: str = "",
agent_name: str = "",
channel_bridge=None,
config=None,
memory_backend=None,
speech_backend=None,
agent_manager=None,
agent_scheduler=None,
api_key: str = "",
webhook_config: dict | None = None,
cors_origins: list[str] | None = None,
) -> FastAPI:
"""Create and configure the FastAPI application.
Parameters
----------
engine:
The inference engine to use for completions.
model:
Default model name.
agent:
Optional agent instance for agent-mode completions.
bus:
Optional event bus for telemetry.
channel_bridge:
Optional channel bridge for multi-platform messaging.
config:
Optional JarvisConfig for other settings.
"""
app = FastAPI(
title="OpenJarvis API",
description="OpenAI-compatible API server for OpenJarvis",
version="0.1.0",
)
from fastapi.middleware.cors import CORSMiddleware
_origins = (
cors_origins
if cors_origins is not None
else [
"http://localhost:5173",
"http://127.0.0.1:5173",
"http://localhost:5174",
"http://127.0.0.1:5174",
# Tauri 2 production webview origins:
# macOS / Linux / iOS -> tauri://localhost
# Windows / Android -> http://tauri.localhost (default),
# https://tauri.localhost when
# windows.useHttpsScheme is enabled
"tauri://localhost",
"http://tauri.localhost",
"https://tauri.localhost",
]
)
app.add_middleware(
CORSMiddleware,
allow_origins=_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Store dependencies in app state
app.state.engine = engine
app.state.model = model
app.state.agent = agent
app.state.bus = bus
app.state.engine_name = engine_name
app.state.agent_name = agent_name or (
getattr(agent, "agent_id", None) if agent else None
)
app.state.channel_bridge = channel_bridge
app.state.config = config
app.state.memory_backend = memory_backend
app.state.speech_backend = speech_backend
app.state.agent_manager = agent_manager
app.state.agent_scheduler = agent_scheduler
app.state.session_start = time.time()
# Exposed so WebSocket handlers can authenticate the handshake (the HTTP
# AuthMiddleware never sees WS upgrade requests). Empty = auth disabled.
app.state.api_key = api_key
# Wire up trace store if traces are enabled
app.state.trace_store = None
try:
from openjarvis.core.config import load_config
from openjarvis.traces.store import TraceStore
cfg = config if config is not None else load_config()
if cfg.traces.enabled:
_trace_store = TraceStore(db_path=cfg.traces.db_path)
app.state.trace_store = _trace_store
_bus = getattr(app.state, "bus", None)
if _bus is not None:
_trace_store.subscribe_to_bus(_bus)
except Exception:
pass # traces are optional; don't block server startup
# Wire up external analytics if enabled (PostHog) — never block startup.
# Note: we do NOT fire app_opened here. The frontend owns that event
# because "server started" (this code path) is not the same as "user
# opened the app" — the server can run headless via cron, daemons,
# or test suites.
app.state.analytics_client = None
app.state.analytics_bridge = None
try:
from openjarvis.analytics import (
AnalyticsClient,
EventBridge,
is_analytics_enabled,
)
from openjarvis.core.config import load_config
_cfg = config if config is not None else load_config()
if is_analytics_enabled(_cfg.analytics):
_client = AnalyticsClient(_cfg.analytics)
app.state.analytics_client = _client
_bus_ref = getattr(app.state, "bus", None)
if _bus_ref is not None:
_bridge = EventBridge(_bus_ref, _client)
_bridge.start()
app.state.analytics_bridge = _bridge
@app.on_event("shutdown")
async def _shutdown_analytics() -> None:
bridge = getattr(app.state, "analytics_bridge", None)
if bridge is not None:
try:
bridge.stop()
except Exception:
pass
client = getattr(app.state, "analytics_client", None)
if client is not None:
try:
client.shutdown()
except Exception:
pass
except Exception as _exc:
logger.debug("Analytics init skipped: %s", _exc)
app.include_router(router)
app.include_router(dashboard_router)
app.include_router(comparison_router)
app.include_router(create_connectors_router())
app.include_router(create_digest_router())
app.include_router(upload_router)
app.include_router(research_router)
app.include_router(analytics_router)
include_all_routes(app)
# Restore SendBlue channel bindings from database on startup
_restore_sendblue_bindings(app)
# Add security headers middleware
try:
from openjarvis.server.middleware import create_security_middleware
middleware_cls = create_security_middleware()
if middleware_cls is not None:
app.add_middleware(middleware_cls)
except Exception as exc:
logger.debug("Security middleware init skipped: %s", exc)
# API key authentication middleware
if api_key:
try:
from openjarvis.server.auth_middleware import AuthMiddleware
app.add_middleware(AuthMiddleware, api_key=api_key)
except Exception as exc:
logger.debug("Auth middleware init skipped: %s", exc)
# Mount webhook routes (always — SendBlue may be configured dynamically)
if webhook_config:
try:
from openjarvis.server.webhook_routes import (
create_webhook_router,
)
webhook_router = create_webhook_router(
bridge=channel_bridge,
twilio_auth_token=webhook_config.get("twilio_auth_token", ""),
bluebubbles_password=webhook_config.get("bluebubbles_password", ""),
whatsapp_verify_token=webhook_config.get("whatsapp_verify_token", ""),
whatsapp_app_secret=webhook_config.get("whatsapp_app_secret", ""),
)
app.include_router(webhook_router)
except Exception as exc:
logger.debug("Webhook routes init skipped: %s", exc)
# Serve static frontend assets if the static/ directory exists
static_dir = pathlib.Path(__file__).parent / "static"
if static_dir.is_dir():
assets_dir = static_dir / "assets"
if assets_dir.is_dir():
app.mount(
"/assets",
_NoCacheStaticFiles(directory=assets_dir),
name="static-assets",
)
@app.get("/{full_path:path}")
async def spa_catch_all(full_path: str):
"""Serve static files directly, fall back to index.html for SPA routes."""
if full_path:
candidate = (static_dir / full_path).resolve()
# Path traversal prevention
resolved_root = static_dir.resolve()
if candidate.is_relative_to(resolved_root) and candidate.is_file():
return FileResponse(candidate, headers=_NO_CACHE_HEADERS)
return FileResponse(
static_dir / "index.html",
headers=_NO_CACHE_HEADERS,
)
return app