|
|
from __future__ import annotations |
|
|
|
|
|
import logging |
|
|
import os |
|
|
import time |
|
|
from contextlib import asynccontextmanager |
|
|
from typing import Any, Dict |
|
|
|
|
|
from fastapi import FastAPI |
|
|
from fastapi.responses import RedirectResponse |
|
|
|
|
|
|
|
|
|
|
|
try: |
|
|
from .middleware import attach_middlewares |
|
|
except Exception: |
|
|
try: |
|
|
from .middlewares import attach_middlewares |
|
|
except Exception: |
|
|
def attach_middlewares(app: FastAPI) -> None: |
|
|
logging.getLogger("uvicorn.error").warning( |
|
|
"attach_middlewares not found; continuing without custom middlewares." |
|
|
) |
|
|
|
|
|
|
|
|
from .routers import health, plan, chat |
|
|
|
|
|
|
|
|
try: |
|
|
from .ui import router as ui_router |
|
|
HAS_UI = True |
|
|
except Exception: |
|
|
HAS_UI = False |
|
|
|
|
|
TAGS_METADATA = [ |
|
|
{"name": "Health", "description": "Liveness / readiness probes and basic service metadata."}, |
|
|
{"name": "Planning", "description": "AI plan generation for Matrix Guardian (/v1/plan)."}, |
|
|
{"name": "Chat", "description": "Lightweight RAG/Q&A about Matrix System (/v1/chat)."}, |
|
|
{"name": "UI", "description": "Minimal web UI (Home, Chat, Dev) if enabled."}, |
|
|
] |
|
|
|
|
|
@asynccontextmanager |
|
|
async def lifespan(app: FastAPI): |
|
|
app.state.started_at = time.time() |
|
|
app.state.version = os.getenv("APP_VERSION", "1.0.0") |
|
|
logging.getLogger("uvicorn.error").info( |
|
|
"matrix-ai starting (version=%s, port=%s)", app.state.version, os.getenv("PORT", "7860") |
|
|
) |
|
|
try: |
|
|
yield |
|
|
finally: |
|
|
uptime = time.time() - getattr(app.state, "started_at", time.time()) |
|
|
logging.getLogger("uvicorn.error").info( |
|
|
"matrix-ai shutting down (uptime=%.2fs)", uptime |
|
|
) |
|
|
|
|
|
def create_app() -> FastAPI: |
|
|
app = FastAPI( |
|
|
title="matrix-ai", |
|
|
version=os.getenv("APP_VERSION", "1.0.0"), |
|
|
description="AI planning microservice for the Matrix EcoSystem", |
|
|
openapi_tags=TAGS_METADATA, |
|
|
docs_url="/docs", |
|
|
redoc_url=None, |
|
|
lifespan=lifespan, |
|
|
) |
|
|
|
|
|
|
|
|
attach_middlewares(app) |
|
|
|
|
|
|
|
|
app.include_router(health.router, tags=["Health"]) |
|
|
app.include_router(plan.router, prefix="/v1", tags=["Planning"]) |
|
|
app.include_router(chat.router, prefix="/v1", tags=["Chat"]) |
|
|
|
|
|
|
|
|
if HAS_UI: |
|
|
app.include_router(ui_router, tags=["UI"]) |
|
|
else: |
|
|
|
|
|
@app.get("/", include_in_schema=False) |
|
|
async def root() -> Dict[str, Any]: |
|
|
return { |
|
|
"ok": True, |
|
|
"service": "matrix-ai", |
|
|
"version": app.version, |
|
|
"docs": "/docs", |
|
|
"endpoints": {"plan": "/v1/plan", "chat": "/v1/chat", "healthz": "/healthz"}, |
|
|
} |
|
|
|
|
|
@app.get("/home", include_in_schema=False) |
|
|
async def home_redirect(): |
|
|
return RedirectResponse(url="/docs", status_code=302) |
|
|
|
|
|
return app |
|
|
|
|
|
app = create_app() |
|
|
|