Coverage for session_mgmt_mcp/config.py: 93.68%
205 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-01 05:22 -0700
1#!/usr/bin/env python3
2"""Configuration Management for Session Management MCP Server.
4Loads configuration from pyproject.toml and environment variables with sensible defaults.
5"""
7import os
8from dataclasses import dataclass, field
9from pathlib import Path
10from typing import Any
12try:
13 import tomllib # Python 3.11+
14except ImportError:
15 try:
16 import tomli as tomllib # fallback for older Python versions
17 except ImportError:
18 tomllib = None
21@dataclass
22class DatabaseConfig:
23 """Database configuration."""
25 path: str = "~/.claude/data/reflection.duckdb"
26 connection_timeout: int = 30
27 query_timeout: int = 120
28 max_connections: int = 10
30 # Multi-project settings
31 enable_multi_project: bool = True
32 auto_detect_projects: bool = True
33 project_groups_enabled: bool = True
35 # Search settings
36 enable_full_text_search: bool = True
37 search_index_update_interval: int = 3600 # seconds
38 max_search_results: int = 100
41@dataclass
42class SearchConfig:
43 """Search and indexing configuration."""
45 enable_semantic_search: bool = True
46 embedding_model: str = "all-MiniLM-L6-v2"
47 embedding_cache_size: int = 1000
49 # Advanced search settings
50 enable_faceted_search: bool = True
51 max_facet_values: int = 50
52 enable_search_suggestions: bool = True
53 suggestion_limit: int = 10
55 # Full-text search
56 enable_stemming: bool = True
57 enable_fuzzy_matching: bool = True
58 fuzzy_threshold: float = 0.8
61@dataclass
62class TokenOptimizationConfig:
63 """Token optimization settings."""
65 enable_optimization: bool = True
66 default_max_tokens: int = 4000
67 default_chunk_size: int = 2000
69 # Optimization strategies
70 preferred_strategy: str = "auto" # auto, truncate_old, summarize_content, etc.
71 enable_response_chunking: bool = True
72 enable_duplicate_filtering: bool = True
74 # Usage tracking
75 track_usage: bool = True
76 usage_retention_days: int = 90
79@dataclass
80class SessionConfig:
81 """Session management configuration."""
83 auto_checkpoint_interval: int = 1800 # seconds (30 minutes)
84 enable_auto_commit: bool = True
85 commit_message_template: str = "checkpoint: Session checkpoint - {timestamp}"
87 # Session permissions
88 enable_permission_system: bool = True
89 default_trusted_operations: list[str] = field(
90 default_factory=lambda: ["git_commit", "uv_sync", "file_operations"],
91 )
93 # Session cleanup
94 auto_cleanup_old_sessions: bool = True
95 session_retention_days: int = 365
98@dataclass
99class IntegrationConfig:
100 """External integrations configuration."""
102 # Crackerjack integration
103 enable_crackerjack: bool = True
104 crackerjack_command: str = "crackerjack"
106 # Git integration
107 enable_git_integration: bool = True
108 git_auto_stage: bool = False
110 # Global workspace
111 global_workspace_path: str = "~/Projects/claude"
112 enable_global_toolkits: bool = True
115@dataclass
116class LoggingConfig:
117 """Logging configuration."""
119 level: str = "INFO"
120 format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
122 # File logging
123 enable_file_logging: bool = True
124 log_file_path: str = "~/.claude/logs/session-mgmt.log"
125 log_file_max_size: int = 10 * 1024 * 1024 # 10MB
126 log_file_backup_count: int = 5
128 # Performance logging
129 enable_performance_logging: bool = False
130 log_slow_queries: bool = True
131 slow_query_threshold: float = 1.0 # seconds
134@dataclass
135class SecurityConfig:
136 """Security and privacy settings."""
138 # Data privacy
139 anonymize_paths: bool = False
140 exclude_sensitive_patterns: list[str] = field(
141 default_factory=lambda: [
142 r"password\s*=\s*['\"][^'\"]+['\"]",
143 r"api[_-]?key\s*=\s*['\"][^'\"]+['\"]",
144 r"secret\s*=\s*['\"][^'\"]+['\"]",
145 r"token\s*=\s*['\"][^'\"]+['\"]",
146 ],
147 )
149 # Access control
150 enable_rate_limiting: bool = True
151 max_requests_per_minute: int = 100
153 # Input validation
154 max_query_length: int = 10000
155 max_content_length: int = 1000000 # 1MB
158@dataclass
159class SessionMgmtConfig:
160 """Main configuration container."""
162 database: DatabaseConfig = field(default_factory=DatabaseConfig)
163 search: SearchConfig = field(default_factory=SearchConfig)
164 token_optimization: TokenOptimizationConfig = field(
165 default_factory=TokenOptimizationConfig,
166 )
167 session: SessionConfig = field(default_factory=SessionConfig)
168 integration: IntegrationConfig = field(default_factory=IntegrationConfig)
169 logging: LoggingConfig = field(default_factory=LoggingConfig)
170 security: SecurityConfig = field(default_factory=SecurityConfig)
172 # MCP Server settings
173 server_host: str = "localhost"
174 server_port: int = 3000
175 enable_websockets: bool = True
177 # Development settings
178 debug: bool = False
179 enable_hot_reload: bool = False
182class ConfigLoader:
183 """Loads configuration from pyproject.toml and environment variables."""
185 def __init__(self, project_root: Path | None = None) -> None:
186 self.project_root = project_root or self._find_project_root()
187 self._config_cache: SessionMgmtConfig | None = None
189 def _find_project_root(self) -> Path:
190 """Find the project root by looking for pyproject.toml."""
191 current = Path.cwd()
193 # Check if we're already in the session-mgmt-mcp directory
194 if (current / "pyproject.toml").exists(): 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true
195 return current
197 # Look for pyproject.toml in parent directories
198 for parent in current.parents: 198 ↛ 214line 198 didn't jump to line 214 because the loop on line 198 didn't complete
199 if (parent / "pyproject.toml").exists(): 199 ↛ 198line 199 didn't jump to line 198 because the condition on line 199 was always true
200 # Check if this is the session-mgmt-mcp project
201 try:
202 with open(parent / "pyproject.toml", "rb") as f:
203 if tomllib: 203 ↛ 198line 203 didn't jump to line 198
204 toml_data = tomllib.load(f)
205 if ( 205 ↛ 198line 205 didn't jump to line 198
206 toml_data.get("project", {}).get("name")
207 == "session-mgmt-mcp"
208 ):
209 return parent
210 except Exception:
211 pass
213 # Fallback to current directory
214 return current
216 def load_config(self, reload: bool = False) -> SessionMgmtConfig:
217 """Load configuration from pyproject.toml and environment variables."""
218 if self._config_cache and not reload:
219 return self._config_cache
221 config = SessionMgmtConfig()
223 # Load from pyproject.toml
224 pyproject_path = self.project_root / "pyproject.toml"
225 if pyproject_path.exists() and tomllib:
226 try:
227 with open(pyproject_path, "rb") as f:
228 toml_data = tomllib.load(f)
229 self._apply_toml_config(config, toml_data)
230 except Exception as e:
231 print(f"Warning: Failed to load pyproject.toml: {e}")
233 # Override with environment variables
234 self._apply_env_config(config)
236 # Expand user paths
237 self._expand_paths(config)
239 # Validate configuration
240 self._validate_config(config)
242 self._config_cache = config
243 return config
245 def _get_tool_config(self, toml_data: dict[str, Any]) -> dict[str, Any]:
246 """Extract tool configuration from TOML data."""
247 tool_config = toml_data.get("tool", {}).get("session-mgmt-mcp", {})
248 if not tool_config:
249 # Also check tool.session_mgmt_mcp (underscore variant)
250 tool_config = toml_data.get("tool", {}).get("session_mgmt_mcp", {})
251 return tool_config
253 def _apply_section_config(
254 self,
255 config: SessionMgmtConfig,
256 section_name: str,
257 section_config: dict[str, Any],
258 ) -> None:
259 """Apply configuration for a specific section."""
260 if not hasattr(config, section_name): 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 return
263 section_obj = getattr(config, section_name)
264 for key, value in section_config.items():
265 if hasattr(section_obj, key): 265 ↛ 264line 265 didn't jump to line 264 because the condition on line 265 was always true
266 setattr(section_obj, key, value)
268 def _apply_server_config(
269 self,
270 config: SessionMgmtConfig,
271 tool_config: dict[str, Any],
272 ) -> None:
273 """Apply server-level configuration."""
274 server_keys = [
275 "server_host",
276 "server_port",
277 "enable_websockets",
278 "debug",
279 "enable_hot_reload",
280 ]
282 for key in server_keys:
283 if key in tool_config and hasattr(config, key):
284 setattr(config, key, tool_config[key])
286 def _apply_toml_config(
287 self, config: SessionMgmtConfig, toml_data: dict[str, Any]
288 ) -> None:
289 """Apply configuration from pyproject.toml."""
290 tool_config = self._get_tool_config(toml_data)
291 if not tool_config: 291 ↛ 292line 291 didn't jump to line 292 because the condition on line 291 was never true
292 return
294 # Define config sections that map to config object attributes
295 config_sections = [
296 "database",
297 "search",
298 "token_optimization",
299 "session",
300 "integration",
301 "logging",
302 "security",
303 ]
305 # Apply section configs
306 for section_name in config_sections:
307 if section_name in tool_config:
308 self._apply_section_config(
309 config,
310 section_name,
311 tool_config[section_name],
312 )
314 # Apply server-level config
315 self._apply_server_config(config, tool_config)
317 def _apply_env_config(self, config: SessionMgmtConfig) -> None:
318 """Apply configuration from environment variables."""
319 env_mappings = {
320 # Database
321 "SESSION_MGMT_DB_PATH": ("database", "path"),
322 "SESSION_MGMT_DB_TIMEOUT": ("database", "connection_timeout", int),
323 "SESSION_MGMT_ENABLE_MULTI_PROJECT": (
324 "database",
325 "enable_multi_project",
326 bool,
327 ),
328 # Search
329 "SESSION_MGMT_ENABLE_SEMANTIC_SEARCH": (
330 "search",
331 "enable_semantic_search",
332 bool,
333 ),
334 "SESSION_MGMT_EMBEDDING_MODEL": ("search", "embedding_model"),
335 # Token optimization
336 "SESSION_MGMT_ENABLE_OPTIMIZATION": (
337 "token_optimization",
338 "enable_optimization",
339 bool,
340 ),
341 "SESSION_MGMT_MAX_TOKENS": (
342 "token_optimization",
343 "default_max_tokens",
344 int,
345 ),
346 # Session
347 "SESSION_MGMT_AUTO_CHECKPOINT": (
348 "session",
349 "auto_checkpoint_interval",
350 int,
351 ),
352 "SESSION_MGMT_ENABLE_AUTO_COMMIT": ("session", "enable_auto_commit", bool),
353 # Logging
354 "SESSION_MGMT_LOG_LEVEL": ("logging", "level"),
355 "SESSION_MGMT_ENABLE_FILE_LOGGING": (
356 "logging",
357 "enable_file_logging",
358 bool,
359 ),
360 # Server
361 "SESSION_MGMT_HOST": ("server_host",),
362 "SESSION_MGMT_PORT": ("server_port", int),
363 "SESSION_MGMT_DEBUG": ("debug", bool),
364 }
366 for env_var, mapping in env_mappings.items():
367 value = os.getenv(env_var)
368 if value is not None:
369 try:
370 # Parse value if type converter provided
371 if len(mapping) > 2:
372 converter = mapping[2]
373 if converter is bool:
374 value = value.lower() in ("true", "1", "yes", "on")
375 elif converter is int: 375 ↛ 386line 375 didn't jump to line 386 because the condition on line 375 was always true
376 value = int(value)
377 elif len(mapping) == 2 and not isinstance(mapping[1], str):
378 # This is a top-level attribute with type (e.g., ("server_port", int))
379 converter = mapping[1]
380 if converter is bool:
381 value = value.lower() in ("true", "1", "yes", "on")
382 elif converter is int: 382 ↛ 386line 382 didn't jump to line 386 because the condition on line 382 was always true
383 value = int(value)
385 # Set value on config object
386 if len(mapping) == 3 or (
387 len(mapping) == 2 and isinstance(mapping[1], str)
388 ):
389 # Nested attribute (e.g., database.path) with optional type
390 section_name, attr_name = mapping[0], mapping[1]
391 section = getattr(config, section_name)
392 setattr(section, attr_name, value)
393 elif len(mapping) >= 1: 393 ↛ 366line 393 didn't jump to line 366 because the condition on line 393 was always true
394 # Top-level attribute - mapping[0] should be string
395 attr_name = mapping[0]
396 if isinstance(attr_name, str): 396 ↛ 366line 396 didn't jump to line 366 because the condition on line 396 was always true
397 setattr(config, attr_name, value)
399 except (ValueError, AttributeError) as e:
400 print(
401 f"Warning: Invalid environment variable {env_var}={value}: {e}",
402 )
404 def _expand_paths(self, config: SessionMgmtConfig) -> None:
405 """Expand user paths in configuration."""
406 # Database path
407 config.database.path = os.path.expanduser(config.database.path)
409 # Log file path
410 config.logging.log_file_path = os.path.expanduser(config.logging.log_file_path)
412 # Global workspace path
413 config.integration.global_workspace_path = os.path.expanduser(
414 config.integration.global_workspace_path,
415 )
417 def _validate_config(self, config: SessionMgmtConfig) -> None:
418 """Validate configuration values."""
419 # Validate port range
420 if not (1024 <= config.server_port <= 65535):
421 print(
422 f"Warning: Invalid server port {config.server_port}, using default 3000",
423 )
424 config.server_port = 3000
426 # Validate token limits
427 if config.token_optimization.default_max_tokens < 100:
428 print("Warning: max_tokens too low, setting to 100")
429 config.token_optimization.default_max_tokens = 100
431 # Validate log level
432 valid_log_levels = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
433 if config.logging.level.upper() not in valid_log_levels:
434 print(f"Warning: Invalid log level {config.logging.level}, using INFO")
435 config.logging.level = "INFO"
437 # Create log directory if needed
438 if config.logging.enable_file_logging:
439 log_path = Path(config.logging.log_file_path)
440 log_path.parent.mkdir(parents=True, exist_ok=True)
442 def get_example_config(self) -> str:
443 """Get example pyproject.toml configuration."""
444 return """
445# Example session-mgmt-mcp configuration in pyproject.toml
447[tool.session-mgmt-mcp]
448# Server settings
449debug = false
450server_host = "localhost"
451server_port = 3000
452enable_websockets = true
454[tool.session-mgmt-mcp.database]
455# Database configuration
456path = "~/.claude/data/reflection.duckdb"
457connection_timeout = 30
458query_timeout = 120
459enable_multi_project = true
460auto_detect_projects = true
461enable_full_text_search = true
463[tool.session-mgmt-mcp.search]
464# Search and embedding settings
465enable_semantic_search = true
466embedding_model = "all-MiniLM-L6-v2"
467enable_faceted_search = true
468max_facet_values = 50
469enable_search_suggestions = true
471[tool.session-mgmt-mcp.token_optimization]
472# Token optimization settings
473enable_optimization = true
474default_max_tokens = 4000
475default_chunk_size = 2000
476preferred_strategy = "auto"
477enable_response_chunking = true
478track_usage = true
480[tool.session-mgmt-mcp.session]
481# Session management
482auto_checkpoint_interval = 1800 # 30 minutes
483enable_auto_commit = true
484enable_permission_system = true
485default_trusted_operations = ["git_commit", "uv_sync", "file_operations"]
486session_retention_days = 365
488[tool.session-mgmt-mcp.integration]
489# External integrations
490enable_crackerjack = true
491enable_git_integration = true
492global_workspace_path = "~/Projects/claude"
493enable_global_toolkits = true
495[tool.session-mgmt-mcp.logging]
496# Logging configuration
497level = "INFO"
498enable_file_logging = true
499log_file_path = "~/.claude/logs/session-mgmt.log"
500enable_performance_logging = false
501log_slow_queries = true
503[tool.session-mgmt-mcp.security]
504# Security settings
505anonymize_paths = false
506enable_rate_limiting = true
507max_requests_per_minute = 100
508max_query_length = 10000
509"""
512# Global config instance
513_config_loader: ConfigLoader | None = None
516def get_config(reload: bool = False) -> SessionMgmtConfig:
517 """Get the global configuration instance."""
518 global _config_loader
520 if _config_loader is None:
521 _config_loader = ConfigLoader()
523 return _config_loader.load_config(reload=reload)
526def reload_config() -> SessionMgmtConfig:
527 """Force reload configuration from files."""
528 return get_config(reload=True)