Merge pull request #106 from kazmer97/fix/analyser-agent-config

Config access simplifications

Author: rstrahan

lib/idp_common_pkg/idp_common/__init__.py

@@ -2,11 +2,22 @@
 # SPDX-License-Identifier: MIT-0
 
 # Use true lazy loading for all submodules
+from typing import TYPE_CHECKING
+
 __version__ = "0.1.0"
 
 # Cache for lazy-loaded submodules
 _submodules = {}
 
+# Type hints are only evaluated during type checking, not at runtime
+if TYPE_CHECKING:
+    from .config import get_config as get_config
+    from .config.models import IDPConfig as IDPConfig
+    from .models import Document as Document
+    from .models import Page as Page
+    from .models import Section as Section
+    from .models import Status as Status
+
 
 def __getattr__(name):
     """Lazy load submodules only when accessed"""
@@ -65,6 +76,7 @@ def __getattr__(name):
     "reporting",
     "agents",
     "get_config",
+    "IDPConfig",
     "Document",
     "Page",
     "Section",

lib/idp_common_pkg/idp_common/agents/analytics/schema_provider.py

@@ -9,6 +9,7 @@
 from typing import Any, Dict, Generator, Optional
 
 from idp_common.config import get_config
+from idp_common.config.models import IDPConfig
 from idp_common.config.schema_constants import (
     SCHEMA_DESCRIPTION,
     SCHEMA_ITEMS,
@@ -211,7 +212,7 @@ def get_evaluation_tables_description() -> str:
 
 
 def get_dynamic_document_sections_description(
-    config: Optional[Dict[str, Any]] = None,
+    config: Optional[IDPConfig] = None,
 ) -> str:
     """
     Generate deployment-specific description of document sections tables based on actual configuration.
@@ -224,10 +225,10 @@ def get_dynamic_document_sections_description(
     """
     try:
         if config is None:
-            config = get_config()
+            config = get_config(as_model=True)
 
         # Get document classes from config
-        classes = config.get("classes", [])
+        classes = config.classes
 
         if not classes:
             logger.warning("No classes found in configuration")
@@ -251,7 +252,7 @@ def get_dynamic_document_sections_description(
         # Generate table list
         table_names = []
         for doc_class in classes:
-            class_name = doc_class.get("name", "Unknown")
+            class_name = doc_class.get(X_AWS_IDP_DOCUMENT_TYPE, "Unknown")
             # Apply exact table name transformation logic
             table_name = f"document_sections_{_get_table_suffix(class_name)}"
             table_names.append(table_name)
@@ -500,7 +501,7 @@ def _get_error_aware_fallback(
 """
 
 
-def get_database_overview(config: Optional[Dict[str, Any]] = None) -> str:
+def get_database_overview(config: Optional[IDPConfig] = None) -> str:
     """
     Get a fast, lightweight overview of available tables with brief descriptions.
     This is the first step in the two-step progressive disclosure system.
@@ -513,10 +514,10 @@ def get_database_overview(config: Optional[Dict[str, Any]] = None) -> str:
     """
     try:
         if config is None:
-            config = get_config()
+            config = get_config(as_model=True)
 
         # Get document classes from config
-        classes = config.get("classes", [])
+        classes = config.classes
 
         overview = """# Database Overview - Available Tables
 
@@ -570,9 +571,7 @@ def get_database_overview(config: Optional[Dict[str, Any]] = None) -> str:
         return """# Database Overview - Error Loading Configuration"""
 
 
-def get_table_info(
-    table_names: list[str], config: Optional[Dict[str, Any]] = None
-) -> str:
+def get_table_info(table_names: list[str], config: Optional[IDPConfig] = None) -> str:
     """
     Get detailed schema information for specific tables.
     This is the second step in the two-step progressive disclosure system.
@@ -620,7 +619,7 @@ def get_table_info(
 
 
 def _get_specific_document_sections_table_info(
-    table_suffix: str, config: Optional[Dict[str, Any]] = None
+    table_suffix: str, config: Optional[IDPConfig] = None
 ) -> str:
     """
     Get detailed information for a specific document sections table.
@@ -634,9 +633,9 @@ def _get_specific_document_sections_table_info(
     """
     try:
         if config is None:
-            config = get_config()
+            config = get_config(as_model=True)
 
-        classes = config.get("classes", [])
+        classes = config.classes
         table_name = f"document_sections_{table_suffix}"
 
         # Find the matching class for this table

lib/idp_common_pkg/idp_common/agents/error_analyzer/agent.py

@@ -6,13 +6,14 @@
 """
 
 import logging
-from typing import Any, Dict
+from typing import Optional
 
 import boto3
 import strands
 
+from idp_common.config import get_config
+
 from ..common.strands_bedrock_model import create_strands_bedrock_model
-from .config import get_error_analyzer_config
 from .tools import (
     cloudwatch_document_logs,
     cloudwatch_logs,
@@ -29,9 +30,7 @@
 
 
 def create_error_analyzer_agent(
-    config: Dict[str, Any] = None,
-    session: boto3.Session = None,
-    pattern_config: Dict[str, Any] = None,
+    session: Optional[boto3.Session] = None,
     **kwargs,
 ) -> strands.Agent:
     """
@@ -44,7 +43,7 @@ def create_error_analyzer_agent(
         pattern_config: Pattern configuration containing agents section
         **kwargs: Additional arguments
     """
-    config = get_error_analyzer_config(pattern_config)
+    config = get_config(as_model=True)
 
     # Create session if not provided
     if session is None:
@@ -63,9 +62,11 @@ def create_error_analyzer_agent(
         xray_performance_analysis,
     ]
     bedrock_model = create_strands_bedrock_model(
-        model_id=config["model_id"], boto_session=session
+        model_id=config.agents.error_analyzer.model_id, boto_session=session
     )
 
     return strands.Agent(
-        tools=tools, system_prompt=config["system_prompt"], model=bedrock_model
+        tools=tools,
+        system_prompt=config.agents.error_analyzer.system_prompt,
+        model=bedrock_model,
     )

lib/idp_common_pkg/idp_common/agents/error_analyzer/config.py

@@ -6,99 +6,11 @@
 """
 
 import logging
-from typing import Any, Dict, List
-
-from ..common.config import configure_logging, get_environment_config
+from typing import Any, Dict
 
 logger = logging.getLogger(__name__)
 
 
-def get_error_analyzer_config(pattern_config: Dict[str, Any] = None) -> Dict[str, Any]:
-    """
-    Builds complete error analyzer configuration from environment and patterns.
-    Get error analyzer configuration with defaults and overrides.
-
-    Returns:
-        Dict containing complete error analyzer configuration
-    """
-    from ... import get_config
-
-    # Start with base environment and context limits
-    config = get_environment_config(["CLOUDWATCH_LOG_GROUP_PREFIX", "AWS_STACK_NAME"])
-    config.update(get_context_limits())
-
-    # Load and apply agent configuration
-    full_config = get_config()
-    agent_config = full_config.get("agents", {}).get("error_analyzer", {})
-
-    if not agent_config:
-        raise ValueError("error_analyzer configuration not found")
-
-    # Apply agent settings with defaults
-    config.update(
-        {
-            "model_id": agent_config.get(
-                "model_id", "anthropic.claude-3-sonnet-20240229-v1:0"
-            ),
-            "system_prompt": agent_config.get("system_prompt"),
-            "error_patterns": get_default_error_patterns(),
-            "aws_capabilities": get_aws_service_capabilities(),
-        }
-    )
-
-    # Apply parameters with type conversion
-    params = agent_config.get("parameters", {})
-    config["max_log_events"] = safe_int_conversion(params.get("max_log_events"), 5)
-    config["time_range_hours_default"] = safe_int_conversion(
-        params.get("time_range_hours_default"), 24
-    )
-
-    # Apply UI overrides for context limits - UI config takes precedence
-    if pattern_config and "max_log_events" in pattern_config:
-        config["max_log_events"] = safe_int_conversion(
-            pattern_config["max_log_events"], config["max_log_events"]
-        )
-
-    # Validate required fields
-    if not config.get("system_prompt"):
-        raise ValueError("system_prompt is required")
-
-    configure_logging(
-        log_level=config.get("log_level"),
-        strands_log_level=config.get("strands_log_level"),
-    )
-
-    return config
-
-
-def get_default_error_patterns() -> List[str]:
-    """Returns standard error patterns for CloudWatch log filtering."""
-    return [
-        "ERROR",
-        "CRITICAL",
-        "FATAL",
-        "Exception",
-        "Traceback",
-        "Failed",
-        "Timeout",
-        "AccessDenied",
-        "ThrottlingException",
-    ]
-
-
-def get_context_limits() -> Dict[str, int]:
-    """Returns default resource and context size constraints."""
-    return {
-        "max_log_events": 5,
-        "max_log_message_length": 400,
-        "max_events_per_log_group": 5,
-        "max_log_groups": 20,
-        "max_stepfunction_timeline_events": 3,
-        "max_stepfunction_error_length": 400,
-        "time_range_hours_default": 24,
-    }
-
-
 def get_aws_service_capabilities() -> Dict[str, Any]:
     """Returns AWS service integration metadata and descriptions."""
     return {
@@ -161,12 +73,3 @@ def truncate_message(message: str, max_length: int = 200) -> str:
     if len(message) <= max_length:
         return message
     return message[:max_length] + "... [truncated]"
-
-
-def get_config_with_fallback() -> Dict[str, Any]:
-    """Gets error analyzer config with graceful fallback to defaults."""
-    try:
-        return get_error_analyzer_config()
-    except Exception as e:
-        logger.warning(f"Failed to load config, using defaults: {e}")
-        return get_context_limits()

lib/idp_common_pkg/idp_common/agents/error_analyzer/tools/stepfunction_tool.py

@@ -11,10 +11,11 @@
 import boto3
 from strands import tool
 
+from idp_common.config import get_config
+
 from ..config import (
     create_error_response,
     create_response,
-    get_config_with_fallback,
 )
 
 logger = logging.getLogger(__name__)
@@ -98,8 +99,11 @@ def _analyze_execution_timeline(events: List[Dict[str, Any]]) -> Dict[str, Any]:
         return {"error": "No execution events available"}
 
     # Cache config values once
-    config = get_config_with_fallback()
-    max_timeline_events = config.get("max_stepfunction_timeline_events", 3)
+    config = get_config(as_model=True)
+
+    max_timeline_events = (
+        config.agents.error_analyzer.parameters.max_stepfunction_timeline_events
+    )
 
     timeline = []
     failure_point = None

lib/idp_common_pkg/idp_common/agents/error_analyzer/tools/xray_tool.py

@@ -13,10 +13,11 @@
 import boto3
 from strands import tool
 
+from idp_common.config import get_config
+
 from ..config import (
     create_error_response,
     create_response,
-    get_config_with_fallback,
     safe_int_conversion,
 )
 
@@ -81,8 +82,10 @@ def _analyze_trace_segments(segments: List[Dict[str, Any]]) -> Dict[str, Any]:
     error_segments = []
     slow_segments = []
 
-    config = get_config_with_fallback()
-    slow_threshold = config.get("xray_slow_segment_threshold_ms", 5000)
+    config = get_config(as_model=True)
+    slow_threshold = (
+        config.agents.error_analyzer.parameters.xray_slow_segment_threshold_ms
+    )
 
     for segment in segments:
         segment_doc = _parse_segment_document(segment.get("Document", {}))
@@ -338,9 +341,13 @@ def _analyze_service_performance(
         if not services:
             return {"services_found": 0, "message": "No service map data available"}
 
-        config = get_config_with_fallback()
-        error_rate_threshold = config.get("xray_error_rate_threshold", 0.05)
-        response_time_threshold = config.get("xray_response_time_threshold_ms", 10000)
+        config = get_config(as_model=True)
+        error_rate_threshold = (
+            config.agents.error_analyzer.parameters.xray_error_rate_threshold
+        )
+        response_time_threshold = (
+            config.agents.error_analyzer.parameters.xray_response_time_threshold_ms
+        )
 
         service_analysis = []
         high_error_services = []