Coverage for src / mcp_server_langgraph / app.py: 72%

1""" 

2Main FastAPI Application 

3 

4Centralized FastAPI app that mounts all HTTP API routers for: 

5- API Key management 

6- Service Principal management 

7- GDPR compliance endpoints 

8- SCIM 2.0 provisioning 

9 

10This app can be run standalone via uvicorn or integrated into the MCP server. 

11 

12Usage: 

13 uvicorn mcp_server_langgraph.app:app --host 0.0.0.0 --port 8000 

14""" 

15 

16import os 

17from collections.abc import AsyncGenerator 

18from contextlib import asynccontextmanager 

19 

20from fastapi import FastAPI 

21from fastapi.middleware.cors import CORSMiddleware 

22 

23from mcp_server_langgraph.api import api_keys_router, gdpr_router, health_router, scim_router, service_principals_router 

24from mcp_server_langgraph.api.auth_request_middleware import AuthRequestMiddleware 

25from mcp_server_langgraph.api.error_handlers import register_exception_handlers 

26from mcp_server_langgraph.api.health import run_startup_validation_async 

27from mcp_server_langgraph.auth.factory import create_user_provider 

28from mcp_server_langgraph.auth.middleware import AuthMiddleware, set_global_auth_middleware 

29from mcp_server_langgraph.core.config import Settings, settings 

30from mcp_server_langgraph.middleware.rate_limiter import setup_rate_limiting 

31from mcp_server_langgraph.observability.telemetry import init_observability, logger 

32 

33 

34def create_app(settings_override: Settings | None = None, skip_startup_validation: bool = False) -> FastAPI: 

35 """ 

36 Create and configure the FastAPI application. 

37 ... 

38 """ 

39 # Use override settings if provided, otherwise use global settings 

40 config = settings_override if settings_override is not None else settings 

41 

42 # Initialize observability FIRST before any logging (OpenAI Codex Finding #3) 

43 # This prevents RuntimeError: "Observability not initialized" when logger is used 

44 init_observability(config) 

45 

46 # Validation: Verify logger is now usable (prevent regression) 

47 try: 

48 logger.debug("Observability initialized successfully") 

49 except RuntimeError as e: 

50 msg = ( 

51 "Observability initialization failed! Logger still raises RuntimeError. " 

52 f"Error: {e}. This is a critical bug - logging will fail throughout the app." 

53 ) 

54 raise RuntimeError(msg) 

55 

56 @asynccontextmanager 

57 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 

58 # Run startup validation to ensure all critical systems initialized correctly 

59 # This prevents the app from starting if any of the OpenAI Codex findings recur 

60 # Skip validation in unit tests (skip_startup_validation=True) to avoid DB dependency 

61 if not skip_startup_validation: 

62 try: 

63 await run_startup_validation_async() 

64 except Exception as e: 

65 try: 

66 logger.critical(f"Startup validation failed: {e}") 

67 except RuntimeError: 

68 pass # Graceful degradation if observability not initialized 

69 raise 

70 else: 

71 try: 

72 logger.debug("Skipping startup validation (test mode)") 

73 except RuntimeError: 

74 pass # Graceful degradation if observability not initialized 

75 

76 yield 

77 

78 app = FastAPI( 

79 title="MCP Server LangGraph API", 

80 version="2.8.0", 

81 description="Production-ready MCP server with LangGraph, OpenFGA, and multi-LLM support", 

82 docs_url="/api/docs", 

83 redoc_url="/api/redoc", 

84 openapi_url="/api/openapi.json", 

85 lifespan=lifespan, 

86 ) 

87 

88 # CORS middleware - use config (settings or override) for environment-aware defaults 

89 # get_cors_origins() provides localhost origins in dev, empty list in production 

90 cors_origins = config.get_cors_origins() 

91 if cors_origins: 

92 app.add_middleware( 

93 CORSMiddleware, 

94 allow_origins=cors_origins, 

95 allow_credentials=True, 

96 allow_methods=["*"], 

97 allow_headers=["*"], 

98 ) 

99 try: 

100 logger.info(f"CORS enabled for origins: {cors_origins}") 

101 except RuntimeError: 

102 pass # Graceful degradation if observability not initialized 

103 else: 

104 try: 

105 logger.info("CORS disabled (no allowed origins configured)") 

106 except RuntimeError: 

107 pass # Graceful degradation if observability not initialized 

108 

109 # Authentication middleware - intercepts requests and verifies JWT tokens 

110 # Sets request.state.user for authenticated requests 

111 # Use factory to create the correct user provider based on AUTH_PROVIDER setting 

112 user_provider = create_user_provider(config) 

113 auth_middleware = AuthMiddleware(secret_key=config.jwt_secret_key, settings=config, user_provider=user_provider) 

114 set_global_auth_middleware(auth_middleware) # Set global instance for dependency injection 

115 app.add_middleware(AuthRequestMiddleware, auth_middleware=auth_middleware) 

116 try: 

117 logger.info("Auth request middleware enabled") 

118 except RuntimeError: 

119 pass # Graceful degradation if observability not initialized 

120 

121 # Rate limiting - setup function registers middleware and exception handlers 

122 setup_rate_limiting(app) 

123 

124 # Register exception handlers 

125 register_exception_handlers(app) 

126 

127 # Include API routers 

128 app.include_router(health_router) # Health check first (doesn't require auth) 

129 app.include_router(api_keys_router) 

130 app.include_router(service_principals_router) 

131 app.include_router(gdpr_router) 

132 app.include_router(scim_router) 

133 

134 try: 

135 logger.info("FastAPI application created with all routers mounted") 

136 except RuntimeError: 

137 pass # Graceful degradation if observability not initialized 

138 

139 return app 

140 

141 

142# Create the application instance 

143# Skip validation when running under pytest to avoid DB dependency in unit tests 

144# This is detected via PYTEST_CURRENT_TEST environment variable set by pytest 

145# Also skip if TESTING env var is set (used by integration tests) 

146_is_pytest_session = os.getenv("PYTEST_CURRENT_TEST") is not None or os.getenv("TESTING") == "true" 

147app = create_app(skip_startup_validation=_is_pytest_session) 

148 

149 

150@app.get("/health") 

151async def health_check() -> dict[str, str]: 

152 """Health check endpoint""" 

153 return {"status": "healthy", "service": "mcp-server-langgraph"} 

154 

155 

156@app.get("/") 

157async def root() -> dict[str, str]: 

158 """Root endpoint with API information""" 

159 return { 

160 "service": "MCP Server LangGraph", 

161 "version": "2.8.0", 

162 "docs": "/api/docs", 

163 "health": "/health", 

164 }