def create_app(
engine,
model: str,
*,
agent=None,
bus=None,
engine_name: str = "",
agent_name: str = "",
channel_bridge=None,
config=None,
memory_backend=None,
memory_service=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 DiapasonConfig for other settings.
"""
@asynccontextmanager
async def _lifespan(application: FastAPI):
prewarm_task = asyncio.create_task(_prewarm_local_model(application))
heartbeat_task = asyncio.create_task(_mesh_heartbeat(application))
from diapason.mesh.discovery import run_discovery
from diapason.server.prechauffage import entretenir_le_prefixe
discovery_task = asyncio.create_task(run_discovery())
# 20/09/2026 : le préfixe du chat (identité + trousse) coûte 24 s à
# froid et meurt avec le runner d'Ollama après 30 min de silence ;
# le rejouer toutes les dix minutes le garde chaud pour la question
# qui suivra la pause. Voir server/prechauffage.py.
prefixe_task = asyncio.create_task(entretenir_le_prefixe(application))
try:
yield
finally:
for task in (prewarm_task, heartbeat_task, discovery_task, prefixe_task):
if not task.done():
task.cancel()
with suppress(asyncio.CancelledError):
await task
bridge = getattr(application.state, "analytics_bridge", None)
if bridge is not None:
try:
bridge.stop()
except Exception:
pass
client = getattr(application.state, "analytics_client", None)
if client is not None:
try:
client.shutdown()
except Exception:
pass
service = getattr(application.state, "memory_service", None)
if service is not None:
try:
service.stop()
except Exception:
pass
conversations = getattr(application.state, "conversations_store", None)
if conversations is not None:
try:
conversations.close()
except Exception:
pass
app = FastAPI(
title="Diapason API",
description="OpenAI-compatible API server for Diapason",
version="1.0.0",
lifespan=_lifespan,
)
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.memory_service = memory_service
app.state.speech_backend = speech_backend
app.state.agent_manager = agent_manager
app.state.agent_scheduler = agent_scheduler
from diapason.actions import LightningActionService
app.state.lightning_actions = LightningActionService(config)
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.
#
# We deliberately do NOT subscribe the trace store to the bus. The chat
# endpoints persist through a TraceCollector that calls store.save()
# directly (mirroring system/orchestrator.py), and the collector ALSO
# publishes TRACE_COMPLETE. A store subscribed to that same bus would
# therefore save every agent trace twice — the second INSERT hitting the
# UNIQUE constraint on trace_id (a 500 on every completion). Keeping the
# collector the single writer is what makes the dual code path safe; only
# the telemetry store is bus-subscribed (see system/builder.py).
app.state.trace_store = None
try:
from diapason.core.config import load_config
from diapason.traces.store import TraceStore
cfg = config if config is not None else load_config()
if cfg.traces.enabled:
app.state.trace_store = TraceStore(db_path=cfg.traces.db_path)
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 diapason.analytics import (
AnalyticsClient,
EventBridge,
is_analytics_enabled,
)
from diapason.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
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(create_dictation_router())
app.include_router(create_config_router())
# Les conversations du chat. Sur l'app localhost UNIQUEMENT, jamais sur
# la sous-app lan : ce sont des transcriptions privées. Avant ce magasin
# (16 sept. 2026), l'historique vivait dans le localStorage du bundle,
# cloisonné par origine — la fenêtre principale et le mini-panneau
# montraient chacun le sien.
conversations_store = ConversationsStore()
app.state.conversations_store = conversations_store
app.include_router(create_conversations_router(conversations_store))
app.include_router(create_screen_share_router())
app.include_router(create_trigger_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 diapason.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 diapason.server.auth_middleware import (
AuthMiddleware,
RateLimitMiddleware,
)
_rate = getattr(config, "security", None)
if _rate is not None:
app.add_middleware(
RateLimitMiddleware,
requests_per_minute=_rate.rate_limit_rpm,
burst_size=_rate.rate_limit_burst,
enabled=_rate.rate_limit_enabled,
)
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 diapason.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