Amazon Security Lake MCP Server

import os from typing import Optional import structlog from pydantic import Field, model_validator from pydantic_settings import BaseSettings logger = structlog.get_logger(__name__) class Settings(BaseSettings): aws_profile: Optional[str] = Field( default=None, description="AWS profile to use for authentication" ) aws_region: str = Field( default="us-east-1", description="AWS region for Athena and Security Lake" ) security_lake_database: Optional[str] = Field( default=None, description="Glue database name for Security Lake (auto-discovered if not set)" ) athena_workgroup: str = Field( default="primary", description="Athena workgroup to use for queries" ) athena_output_location: Optional[str] = Field( default=None, description="S3 location for Athena query results (auto-discovered if not set)" ) auto_discover_resources: bool = Field( default=True, description="Automatically discover AWS resources if not explicitly configured" ) log_level: str = Field( default="INFO", description="Logging level" ) max_query_results: int = Field( default=1000, description="Maximum number of results to return per query" ) query_timeout_seconds: int = Field( default=300, description="Maximum time to wait for query completion" ) enable_query_caching: bool = Field( default=True, description="Enable caching of query results" ) cache_ttl_seconds: int = Field( default=3600, description="Time to live for cached results in seconds" ) model_config = { "env_prefix": "ASL_MCP_", "case_sensitive": False, "env_file": ".env" } @model_validator(mode='after') def discover_aws_resources(self): """Auto-discover AWS resources if not explicitly configured""" if not self.auto_discover_resources: # Validate that required fields are set when auto-discovery is disabled if not self.athena_output_location: raise ValueError("athena_output_location is required when auto_discover_resources is False") if not self.security_lake_database: raise ValueError("security_lake_database is required when auto_discover_resources is False") return self try: from ..aws.discovery import AWSResourceDiscovery discovery = AWSResourceDiscovery( aws_region=self.aws_region, aws_profile=self.aws_profile ) # Auto-discover Athena output location if not set if not self.athena_output_location: discovered_bucket = discovery.discover_security_lake_bucket() if discovered_bucket: self.athena_output_location = discovered_bucket logger.info("Auto-discovered Athena output location", location=discovered_bucket) else: logger.warning("Could not auto-discover Athena output location") # Auto-discover Security Lake database if not set if not self.security_lake_database: discovered_db = discovery.discover_security_lake_database() if discovered_db: self.security_lake_database = discovered_db logger.info("Auto-discovered Security Lake database", database=discovered_db) else: logger.warning("Could not auto-discover Security Lake database") # Set default as fallback self.security_lake_database = "amazon_security_lake_glue_db" except Exception as e: logger.warning("AWS resource discovery failed", error=str(e)) # Set defaults if discovery fails if not self.security_lake_database: self.security_lake_database = "amazon_security_lake_glue_db" # Final validation if not self.athena_output_location: raise ValueError( "athena_output_location could not be auto-discovered and was not provided. " "Please set ASL_MCP_ATHENA_OUTPUT_LOCATION environment variable or create an Athena results bucket." ) return self def get_settings() -> Settings: return Settings()