nex_docus/backend/main.py

85 lines
1.9 KiB
Python

"""
FastAPI 主应用
"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from app.core.config import settings
from app.core.redis_client import init_redis, close_redis
from app.api.v1 import api_router
from app.mcp import create_mcp_http_app, get_mcp_session_manager, MCPHeaderAuthApp
try:
mcp_http_app = create_mcp_http_app()
mcp_session_manager = get_mcp_session_manager()
except RuntimeError:
mcp_http_app = None
mcp_session_manager = None
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理"""
# 启动时初始化 Redis
await init_redis()
if mcp_session_manager is not None:
async with mcp_session_manager.run():
yield
else:
yield
# 关闭时清理 Redis 连接
await close_redis()
# 创建 FastAPI 应用
app = FastAPI(
title=settings.APP_NAME,
version=settings.APP_VERSION,
description="NEX Docus - 团队协作文档管理平台",
debug=settings.DEBUG,
lifespan=lifespan,
)
# 配置 CORS
app.add_middleware(
CORSMiddleware,
allow_origins=settings.CORS_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 注册 API 路由
app.include_router(api_router, prefix="/api/v1")
# 挂载 MCP Streamable HTTP 入口
if mcp_http_app is not None:
app.mount("/mcp", MCPHeaderAuthApp(mcp_http_app))
@app.get("/")
async def root():
"""根路径"""
return {
"name": settings.APP_NAME,
"version": settings.APP_VERSION,
"status": "running"
}
@app.get("/health")
async def health_check():
"""健康检查"""
return {"status": "healthy"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"main:app",
host=settings.HOST,
port=settings.PORT,
reload=settings.DEBUG,
)