Coverage for src / agent / config / manager.py: 81%

1# Copyright 2025-2026 Microsoft Corporation 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15"""Configuration file manager for loading, saving, and managing agent settings.""" 

16 

17import json 

18import os 

19from pathlib import Path 

20from typing import Any 

21 

22from pydantic import ValidationError 

23 

24from .schema import AgentSettings 

25 

26 

27class ConfigurationError(Exception): 

28 """Raised when configuration operations fail.""" 

29 

30 pass 

31 

32 

33def get_config_path() -> Path: 

34 """Get the path to the configuration file. 

35 

36 Returns: 

37 Path to ~/.osdu-agent/settings.json 

38 """ 

39 return Path.home() / ".osdu-agent" / "settings.json" 

40 

41 

42def load_config(config_path: Path | None = None) -> AgentSettings: 

43 """Load configuration from JSON file. 

44 

45 Args: 

46 config_path: Optional path to config file. Defaults to ~/.osdu-agent/settings.json 

47 

48 Returns: 

49 AgentSettings instance loaded from file, or default settings if file doesn't exist 

50 

51 Raises: 

52 ConfigurationError: If file exists but is invalid JSON or fails validation 

53 

54 Note: 

55 This function loads configuration from file only. For environment variable 

56 merging, use load_config_with_env() or manually apply merge_with_env(). 

57 

58 Example: 

59 >>> settings = load_config() 

60 >>> settings.providers.enabled 

61 ['local'] 

62 """ 

63 if config_path is None: 

64 config_path = get_config_path() 

65 

66 # Return defaults if file doesn't exist 

67 if not config_path.exists(): 

68 return AgentSettings() 

69 

70 try: 

71 with open(config_path) as f: 

72 data = json.load(f) 

73 

74 # Validate and load into Pydantic model 

75 return AgentSettings(**data) 

76 

77 except json.JSONDecodeError as e: 

78 raise ConfigurationError(f"Invalid JSON in configuration file {config_path}: {e}") from e 

79 except ValidationError as e: 

80 raise ConfigurationError(f"Configuration validation failed for {config_path}:\n{e}") from e 

81 except Exception as e: 

82 raise ConfigurationError(f"Failed to load configuration from {config_path}: {e}") from e 

83 

84 

85def load_config_with_env(config_path: Path | None = None) -> AgentSettings: 

86 """Load configuration from JSON file and merge with environment variables. 

87 

88 Merges configuration from three sources (in precedence order): 

89 1. Environment variables (highest priority) 

90 2. Configuration file (~/.osdu-agent/settings.json) 

91 3. Default values (lowest priority) 

92 

93 Args: 

94 config_path: Optional path to config file. Defaults to ~/.osdu-agent/settings.json 

95 

96 Returns: 

97 AgentSettings instance with merged configuration 

98 

99 Raises: 

100 ConfigurationError: If file exists but is invalid JSON or fails validation 

101 

102 Example: 

103 >>> settings = load_config_with_env() 

104 >>> settings.providers.enabled # From file or env (LLM_PROVIDER) 

105 ['openai'] 

106 """ 

107 # Load base configuration 

108 settings = load_config(config_path) 

109 

110 # Apply environment variable overrides 

111 env_overrides = merge_with_env(settings) 

112 if env_overrides: 

113 merged_dict = deep_merge(settings.model_dump(), env_overrides) 

114 settings = AgentSettings(**merged_dict) 

115 

116 return settings 

117 

118 

119def save_config(settings: AgentSettings, config_path: Path | None = None) -> None: 

120 """Save configuration to JSON file with minimal formatting. 

121 

122 Uses progressive disclosure: only saves enabled providers and non-null values. 

123 This creates cleaner, more user-friendly config files (~20 lines vs 100+ lines). 

124 

125 Sets restrictive permissions (0o600) on POSIX systems to protect API keys. 

126 

127 Args: 

128 settings: AgentSettings instance to save 

129 config_path: Optional path to config file. Defaults to ~/.osdu-agent/settings.json 

130 

131 Raises: 

132 ConfigurationError: If save operation fails 

133 

134 Example: 

135 >>> settings = AgentSettings() 

136 >>> settings.providers.enabled = ["openai"] 

137 >>> settings.providers.openai.api_key = "sk-..." 

138 >>> save_config(settings) 

139 # Creates minimal config with only openai (not all 6 providers) 

140 """ 

141 if config_path is None: 

142 config_path = get_config_path() 

143 

144 # Create directory if it doesn't exist 

145 config_path.parent.mkdir(parents=True, exist_ok=True) 

146 

147 try: 

148 # Set restrictive permissions before writing (POSIX only) 

149 old_umask = os.umask(0o077) if os.name != "nt" else None 

150 try: 

151 # Write with minimal formatting (progressive disclosure) 

152 json_str = settings.model_dump_json_minimal() 

153 with open(config_path, "w") as f: 

154 f.write(json_str) 

155 

156 # Set restrictive permissions on POSIX systems (user read/write only) 

157 if os.name != "nt": # Not Windows 

158 os.chmod(config_path, 0o600) 

159 finally: 

160 if old_umask is not None: 

161 os.umask(old_umask) 

162 

163 except Exception as e: 

164 raise ConfigurationError(f"Failed to save configuration to {config_path}: {e}") from e 

165 

166 

167def merge_with_env(settings: AgentSettings) -> dict[str, Any]: 

168 """Merge configuration file settings with environment variable overrides. 

169 

170 Environment variables take precedence over file settings. 

171 Returns a dictionary that can be used to update provider configs. 

172 

173 Args: 

174 settings: AgentSettings instance from file 

175 

176 Returns: 

177 Dictionary of environment variable overrides 

178 

179 Example: 

180 >>> settings = load_config() 

181 >>> env_overrides = merge_with_env(settings) 

182 >>> # Apply overrides to settings if needed 

183 """ 

184 env_overrides: dict[str, Any] = {} 

185 

186 # Check if LLM_PROVIDER environment variable is set 

187 llm_provider = os.getenv("LLM_PROVIDER") 

188 

189 # If LLM_PROVIDER is set and not already in enabled list, add it 

190 if llm_provider and llm_provider not in settings.providers.enabled: 

191 # Add provider to enabled list (env takes precedence) 

192 enabled_list = [llm_provider] + settings.providers.enabled 

193 env_overrides.setdefault("providers", {})["enabled"] = enabled_list 

194 

195 # OpenAI overrides 

196 if os.getenv("OPENAI_API_KEY"): 

197 env_overrides.setdefault("providers", {}).setdefault("openai", {})["api_key"] = os.getenv( 

198 "OPENAI_API_KEY" 

199 ) 

200 if os.getenv("AGENT_MODEL") and llm_provider == "openai": 

201 env_overrides.setdefault("providers", {}).setdefault("openai", {})["model"] = os.getenv( 

202 "AGENT_MODEL" 

203 ) 

204 

205 # Anthropic overrides 

206 if os.getenv("ANTHROPIC_API_KEY"): 

207 env_overrides.setdefault("providers", {}).setdefault("anthropic", {})["api_key"] = ( 

208 os.getenv("ANTHROPIC_API_KEY") 

209 ) 

210 if os.getenv("AGENT_MODEL") and llm_provider == "anthropic": 

211 env_overrides.setdefault("providers", {}).setdefault("anthropic", {})["model"] = os.getenv( 

212 "AGENT_MODEL" 

213 ) 

214 

215 # Azure OpenAI overrides 

216 if os.getenv("AZURE_OPENAI_ENDPOINT"): 

217 env_overrides.setdefault("providers", {}).setdefault("azure", {})["endpoint"] = os.getenv( 

218 "AZURE_OPENAI_ENDPOINT" 

219 ) 

220 if os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"): 

221 env_overrides.setdefault("providers", {}).setdefault("azure", {})["deployment"] = os.getenv( 

222 "AZURE_OPENAI_DEPLOYMENT_NAME" 

223 ) 

224 if os.getenv("AZURE_OPENAI_API_KEY"): 

225 env_overrides.setdefault("providers", {}).setdefault("azure", {})["api_key"] = os.getenv( 

226 "AZURE_OPENAI_API_KEY" 

227 ) 

228 if os.getenv("AZURE_OPENAI_VERSION"): 

229 env_overrides.setdefault("providers", {}).setdefault("azure", {})["api_version"] = ( 

230 os.getenv("AZURE_OPENAI_VERSION") 

231 ) 

232 

233 # Azure AI Foundry overrides 

234 if os.getenv("AZURE_PROJECT_ENDPOINT"): 

235 env_overrides.setdefault("providers", {}).setdefault("foundry", {})["project_endpoint"] = ( 

236 os.getenv("AZURE_PROJECT_ENDPOINT") 

237 ) 

238 if os.getenv("AZURE_MODEL_DEPLOYMENT"): 

239 env_overrides.setdefault("providers", {}).setdefault("foundry", {})["model_deployment"] = ( 

240 os.getenv("AZURE_MODEL_DEPLOYMENT") 

241 ) 

242 

243 # Gemini overrides 

244 if os.getenv("GEMINI_API_KEY"): 

245 env_overrides.setdefault("providers", {}).setdefault("gemini", {})["api_key"] = os.getenv( 

246 "GEMINI_API_KEY" 

247 ) 

248 if os.getenv("GEMINI_PROJECT_ID"): 

249 env_overrides.setdefault("providers", {}).setdefault("gemini", {})["project_id"] = ( 

250 os.getenv("GEMINI_PROJECT_ID") 

251 ) 

252 if os.getenv("GEMINI_LOCATION"): 

253 env_overrides.setdefault("providers", {}).setdefault("gemini", {})["location"] = os.getenv( 

254 "GEMINI_LOCATION" 

255 ) 

256 if os.getenv("GEMINI_USE_VERTEXAI"): 

257 env_overrides.setdefault("providers", {}).setdefault("gemini", {})["use_vertexai"] = ( 

258 os.getenv("GEMINI_USE_VERTEXAI", "false").lower() == "true" 

259 ) 

260 if os.getenv("AGENT_MODEL") and llm_provider == "gemini": 

261 env_overrides.setdefault("providers", {}).setdefault("gemini", {})["model"] = os.getenv( 

262 "AGENT_MODEL" 

263 ) 

264 

265 # Local provider overrides 

266 if os.getenv("LOCAL_BASE_URL"): 

267 env_overrides.setdefault("providers", {}).setdefault("local", {})["base_url"] = os.getenv( 

268 "LOCAL_BASE_URL" 

269 ) 

270 if os.getenv("LOCAL_MODEL"): 

271 env_overrides.setdefault("providers", {}).setdefault("local", {})["model"] = os.getenv( 

272 "LOCAL_MODEL" 

273 ) 

274 if os.getenv("AGENT_MODEL") and llm_provider == "local": 

275 env_overrides.setdefault("providers", {}).setdefault("local", {})["model"] = os.getenv( 

276 "AGENT_MODEL" 

277 ) 

278 

279 # Agent config overrides 

280 if os.getenv("AGENT_DATA_DIR"): 

281 env_overrides.setdefault("agent", {})["data_dir"] = os.getenv("AGENT_DATA_DIR") 

282 

283 # Note: AGENT_SKILLS environment variable removed 

284 # Skills now configured via settings.skills (plugins, disabled_bundled) 

285 

286 # Telemetry overrides 

287 if os.getenv("ENABLE_OTEL"): 

288 env_overrides.setdefault("telemetry", {})["enabled"] = ( 

289 os.getenv("ENABLE_OTEL", "false").lower() == "true" 

290 ) 

291 if os.getenv("ENABLE_SENSITIVE_DATA"): 

292 env_overrides.setdefault("telemetry", {})["enable_sensitive_data"] = ( 

293 os.getenv("ENABLE_SENSITIVE_DATA", "false").lower() == "true" 

294 ) 

295 if os.getenv("OTLP_ENDPOINT"): 

296 env_overrides.setdefault("telemetry", {})["otlp_endpoint"] = os.getenv("OTLP_ENDPOINT") 

297 if os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING"): 

298 env_overrides.setdefault("telemetry", {})["applicationinsights_connection_string"] = ( 

299 os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING") 

300 ) 

301 

302 # Memory overrides 

303 if os.getenv("MEMORY_ENABLED"): 

304 env_overrides.setdefault("memory", {})["enabled"] = ( 

305 os.getenv("MEMORY_ENABLED", "true").lower() == "true" 

306 ) 

307 if os.getenv("MEMORY_TYPE"): 

308 env_overrides.setdefault("memory", {})["type"] = os.getenv("MEMORY_TYPE") 

309 if os.getenv("MEMORY_HISTORY_LIMIT"): 

310 try: 

311 env_overrides.setdefault("memory", {})["history_limit"] = int( 

312 os.getenv("MEMORY_HISTORY_LIMIT", "20") 

313 ) 

314 except ValueError: 

315 # Invalid value, fallback to default 

316 env_overrides.setdefault("memory", {})["history_limit"] = 20 

317 

318 # Mem0 overrides 

319 if os.getenv("MEM0_STORAGE_PATH"): 

320 env_overrides.setdefault("memory", {}).setdefault("mem0", {})["storage_path"] = os.getenv( 

321 "MEM0_STORAGE_PATH" 

322 ) 

323 if os.getenv("MEM0_API_KEY"): 

324 env_overrides.setdefault("memory", {}).setdefault("mem0", {})["api_key"] = os.getenv( 

325 "MEM0_API_KEY" 

326 ) 

327 if os.getenv("MEM0_ORG_ID"): 

328 env_overrides.setdefault("memory", {}).setdefault("mem0", {})["org_id"] = os.getenv( 

329 "MEM0_ORG_ID" 

330 ) 

331 if os.getenv("MEM0_USER_ID"): 

332 env_overrides.setdefault("memory", {}).setdefault("mem0", {})["user_id"] = os.getenv( 

333 "MEM0_USER_ID" 

334 ) 

335 if os.getenv("MEM0_PROJECT_ID"): 

336 env_overrides.setdefault("memory", {}).setdefault("mem0", {})["project_id"] = os.getenv( 

337 "MEM0_PROJECT_ID" 

338 ) 

339 

340 return env_overrides 

341 

342 

343def validate_config(settings: AgentSettings) -> list[str]: 

344 """Validate configuration and return list of errors. 

345 

346 Args: 

347 settings: AgentSettings instance to validate 

348 

349 Returns: 

350 List of validation error messages (empty if valid) 

351 

352 Example: 

353 >>> settings = load_config() 

354 >>> errors = validate_config(settings) 

355 >>> if errors: 

356 ... print("Configuration errors:", errors) 

357 """ 

358 return settings.validate_enabled_providers() 

359 

360 

361def migrate_from_env() -> AgentSettings: 

362 """Create settings.json from current environment variables. 

363 

364 Reads current .env file and environment variables, creates a new 

365 AgentSettings instance with those values. 

366 

367 Returns: 

368 AgentSettings instance populated from environment 

369 

370 Example: 

371 >>> settings = migrate_from_env() 

372 >>> save_config(settings) 

373 """ 

374 from dotenv import load_dotenv 

375 

376 # Load .env file if it exists 

377 load_dotenv() 

378 

379 # Determine which provider is configured 

380 llm_provider = os.getenv("LLM_PROVIDER", "local") 

381 agent_model = os.getenv("AGENT_MODEL") 

382 

383 # Build settings dict 

384 settings_dict: dict[str, Any] = { 

385 "version": "1.0", 

386 "providers": { 

387 "enabled": [llm_provider], 

388 "local": { 

389 "enabled": llm_provider == "local", 

390 "base_url": os.getenv( 

391 "LOCAL_BASE_URL", "http://localhost:12434/engines/llama.cpp/v1" 

392 ), 

393 "model": agent_model if llm_provider == "local" and agent_model else "ai/phi4", 

394 }, 

395 "openai": { 

396 "enabled": llm_provider == "openai", 

397 "api_key": os.getenv("OPENAI_API_KEY"), 

398 "model": agent_model if llm_provider == "openai" and agent_model else "gpt-5-mini", 

399 }, 

400 "anthropic": { 

401 "enabled": llm_provider == "anthropic", 

402 "api_key": os.getenv("ANTHROPIC_API_KEY"), 

403 "model": ( 

404 agent_model 

405 if llm_provider == "anthropic" and agent_model 

406 else "claude-haiku-4-5-20251001" 

407 ), 

408 }, 

409 "azure": { 

410 "enabled": llm_provider == "azure", 

411 "endpoint": os.getenv("AZURE_OPENAI_ENDPOINT"), 

412 "deployment": os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"), 

413 "api_version": os.getenv("AZURE_OPENAI_VERSION", "2025-03-01-preview"), 

414 "api_key": os.getenv("AZURE_OPENAI_API_KEY"), 

415 }, 

416 "foundry": { 

417 "enabled": llm_provider == "foundry", 

418 "project_endpoint": os.getenv("AZURE_PROJECT_ENDPOINT"), 

419 "model_deployment": os.getenv("AZURE_MODEL_DEPLOYMENT"), 

420 }, 

421 "gemini": { 

422 "enabled": llm_provider == "gemini", 

423 "api_key": os.getenv("GEMINI_API_KEY"), 

424 "model": ( 

425 agent_model 

426 if llm_provider == "gemini" and agent_model 

427 else "gemini-2.0-flash-exp" 

428 ), 

429 "use_vertexai": os.getenv("GEMINI_USE_VERTEXAI", "false").lower() == "true", 

430 "project_id": os.getenv("GEMINI_PROJECT_ID"), 

431 "location": os.getenv("GEMINI_LOCATION"), 

432 }, 

433 }, 

434 "agent": { 

435 "data_dir": os.getenv("AGENT_DATA_DIR", "~/.osdu-agent"), 

436 "log_level": "info", 

437 }, 

438 "telemetry": { 

439 "enabled": os.getenv("ENABLE_OTEL", "false").lower() == "true", 

440 "enable_sensitive_data": os.getenv("ENABLE_SENSITIVE_DATA", "false").lower() == "true", 

441 "otlp_endpoint": os.getenv("OTLP_ENDPOINT", "http://localhost:4317"), 

442 "applicationinsights_connection_string": os.getenv( 

443 "APPLICATIONINSIGHTS_CONNECTION_STRING" 

444 ), 

445 }, 

446 "memory": { 

447 "enabled": os.getenv("MEMORY_ENABLED", "true").lower() == "true", 

448 "type": os.getenv("MEMORY_TYPE", "in_memory"), 

449 "history_limit": int(os.getenv("MEMORY_HISTORY_LIMIT", "20")), 

450 "mem0": { 

451 "storage_path": os.getenv("MEM0_STORAGE_PATH"), 

452 "api_key": os.getenv("MEM0_API_KEY"), 

453 "org_id": os.getenv("MEM0_ORG_ID"), 

454 "user_id": os.getenv("MEM0_USER_ID"), 

455 "project_id": os.getenv("MEM0_PROJECT_ID"), 

456 }, 

457 }, 

458 } 

459 

460 return AgentSettings(**settings_dict) 

461 

462 

463def deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: 

464 """Deep merge two dictionaries, with override taking precedence. 

465 

466 Args: 

467 base: Base dictionary 

468 override: Override dictionary (takes precedence) 

469 

470 Returns: 

471 Merged dictionary 

472 """ 

473 result = base.copy() 

474 for key, value in override.items(): 

475 if key in result and isinstance(result[key], dict) and isinstance(value, dict): 

476 result[key] = deep_merge(result[key], value) 

477 else: 

478 result[key] = value 

479 return result