Coverage for src / mcp_server_langgraph / core / feature_flags.py: 100%

1""" 

2Feature Flag Management System 

3 

4Enables gradual rollouts, A/B testing, and safe feature deployment. 

5All flags are configurable via environment variables for different environments. 

6""" 

7 

8from typing import Any 

9 

10from pydantic import Field 

11from pydantic_settings import BaseSettings, SettingsConfigDict 

12 

13 

14class FeatureFlags(BaseSettings): 

15 """ 

16 Feature flags for controlling system behavior 

17 

18 All flags can be overridden via environment variables with FF_ prefix. 

19 Example: FF_ENABLE_PYDANTIC_AI_ROUTING=false 

20 """ 

21 

22 # Pydantic AI Features 

23 enable_pydantic_ai_routing: bool = Field( 

24 default=True, 

25 description="Use Pydantic AI for type-safe routing decisions with confidence scoring", 

26 ) 

27 

28 enable_pydantic_ai_responses: bool = Field( 

29 default=True, 

30 description="Use Pydantic AI for structured response generation with validation", 

31 ) 

32 

33 pydantic_ai_confidence_threshold: float = Field( 

34 default=0.7, 

35 ge=0.0, 

36 le=1.0, 

37 description="Minimum confidence score to trust Pydantic AI routing (0.0-1.0)", 

38 ) 

39 

40 # LLM Features 

41 enable_llm_fallback: bool = Field( 

42 default=True, 

43 description="Enable automatic fallback to alternative models on failure", 

44 ) 

45 

46 enable_streaming_responses: bool = Field( 

47 default=True, 

48 description="Enable streaming responses for real-time output", 

49 ) 

50 

51 llm_timeout_seconds: int = Field( 

52 default=60, 

53 ge=10, 

54 le=300, 

55 description="Maximum time to wait for LLM responses (10-300 seconds)", 

56 ) 

57 

58 # Authorization Features 

59 enable_openfga: bool = Field( 

60 default=True, 

61 description="Enable OpenFGA for fine-grained authorization checks", 

62 ) 

63 

64 openfga_strict_mode: bool = Field( 

65 default=False, 

66 description="Fail-closed on OpenFGA errors (strict) vs fail-open (permissive)", 

67 ) 

68 

69 openfga_cache_ttl_seconds: int = Field( 

70 default=60, 

71 ge=0, 

72 le=3600, 

73 description="Cache authorization check results for N seconds (0=disabled)", 

74 ) 

75 

76 # Keycloak Features 

77 enable_keycloak: bool = Field( 

78 default=True, 

79 description="Enable Keycloak integration for authentication", 

80 ) 

81 

82 enable_token_refresh: bool = Field( 

83 default=True, 

84 description="Enable automatic token refresh for Keycloak tokens", 

85 ) 

86 

87 keycloak_role_sync: bool = Field( 

88 default=True, 

89 description="Sync Keycloak roles/groups to OpenFGA on authentication", 

90 ) 

91 

92 openfga_sync_on_login: bool = Field( 

93 default=True, 

94 description="Synchronize user roles to OpenFGA on every login", 

95 ) 

96 

97 # Observability Features 

98 enable_langsmith: bool = Field( 

99 default=False, 

100 description="Enable LangSmith tracing for LLM-specific observability", 

101 ) 

102 

103 enable_detailed_logging: bool = Field( 

104 default=True, 

105 description="Include detailed context in logs (may increase log volume)", 

106 ) 

107 

108 enable_trace_sampling: bool = Field( 

109 default=False, 

110 description="Sample traces rather than capturing all (reduces overhead)", 

111 ) 

112 

113 trace_sample_rate: float = Field( 

114 default=0.1, 

115 ge=0.0, 

116 le=1.0, 

117 description="Percentage of traces to sample when sampling enabled (0.0-1.0)", 

118 ) 

119 

120 # Performance Optimizations 

121 enable_response_caching: bool = Field( 

122 default=False, 

123 description="Cache LLM responses for identical inputs (experimental)", 

124 ) 

125 

126 cache_ttl_seconds: int = Field( 

127 default=300, 

128 ge=60, 

129 le=86400, 

130 description="How long to cache responses (60s-24h)", 

131 ) 

132 

133 enable_request_batching: bool = Field( 

134 default=False, 

135 description="Batch multiple requests to reduce LLM API calls (experimental)", 

136 ) 

137 

138 max_batch_size: int = Field( 

139 default=10, 

140 ge=1, 

141 le=100, 

142 description="Maximum number of requests to batch together", 

143 ) 

144 

145 # Agent Behavior 

146 max_agent_iterations: int = Field( 

147 default=10, 

148 ge=1, 

149 le=50, 

150 description="Maximum iterations for agent loops before stopping", 

151 ) 

152 

153 enable_agent_memory: bool = Field( 

154 default=True, 

155 description="Enable conversation memory/checkpointing for stateful agents", 

156 ) 

157 

158 memory_max_messages: int = Field( 

159 default=100, 

160 ge=10, 

161 le=1000, 

162 description="Maximum messages to retain in conversation history", 

163 ) 

164 

165 # Security Features 

166 enable_rate_limiting: bool = Field( 

167 default=True, 

168 description="Enable rate limiting to prevent abuse", 

169 ) 

170 

171 rate_limit_requests_per_minute: int = Field( 

172 default=60, 

173 ge=1, 

174 le=1000, 

175 description="Maximum requests per minute per user", 

176 ) 

177 

178 enable_input_validation: bool = Field( 

179 default=True, 

180 description="Validate and sanitize all user inputs", 

181 ) 

182 

183 max_input_length: int = Field( 

184 default=10000, 

185 ge=100, 

186 le=100000, 

187 description="Maximum input length in characters", 

188 ) 

189 

190 # Experimental Features 

191 enable_experimental_features: bool = Field( 

192 default=False, 

193 description="Master switch for all experimental features", 

194 ) 

195 

196 enable_multi_agent_collaboration: bool = Field( 

197 default=False, 

198 description="Enable multiple agents working together (experimental)", 

199 ) 

200 

201 enable_tool_reflection: bool = Field( 

202 default=False, 

203 description="Enable agents to reflect on tool usage effectiveness (experimental)", 

204 ) 

205 

206 model_config = SettingsConfigDict( 

207 env_prefix="FF_", # All flags use FF_ prefix in environment 

208 env_file=".env", 

209 env_file_encoding="utf-8", 

210 case_sensitive=False, 

211 extra="ignore", # Ignore extra fields from environment 

212 ) 

213 

214 def is_feature_enabled(self, feature_name: str) -> bool: 

215 """ 

216 Check if a feature is enabled 

217 

218 Args: 

219 feature_name: Name of the feature flag (e.g., 'enable_pydantic_ai_routing') 

220 

221 Returns: 

222 True if feature is enabled, False otherwise 

223 """ 

224 return getattr(self, feature_name, False) 

225 

226 def get_feature_value(self, feature_name: str, default: Any | None = None) -> Any: 

227 """ 

228 Get feature flag value with fallback 

229 

230 Args: 

231 feature_name: Name of the feature flag 

232 default: Default value if flag not found 

233 

234 Returns: 

235 Feature flag value or default 

236 """ 

237 return getattr(self, feature_name, default) 

238 

239 def should_use_experimental(self, feature_name: str) -> bool: 

240 """ 

241 Check if experimental feature should be enabled 

242 

243 Requires both the master experimental switch and individual feature flag. 

244 

245 Args: 

246 feature_name: Name of the experimental feature 

247 

248 Returns: 

249 True if both master switch and feature are enabled 

250 """ 

251 if not self.enable_experimental_features: 

252 return False 

253 

254 return self.is_feature_enabled(feature_name) 

255 

256 

257# Global feature flags instance 

258feature_flags = FeatureFlags() 

259 

260 

261def get_feature_flags() -> FeatureFlags: 

262 """Get the global feature flags instance""" 

263 return feature_flags 

264 

265 

266def is_enabled(feature_name: str) -> bool: 

267 """ 

268 Convenience function to check if a feature is enabled 

269 

270 Args: 

271 feature_name: Name of the feature flag 

272 

273 Returns: 

274 True if enabled, False otherwise 

275 

276 Example: 

277 if is_enabled('enable_pydantic_ai_routing'): 

278 # Use Pydantic AI routing 

279 """ 

280 return feature_flags.is_feature_enabled(feature_name)