feat: Lazy initialization (#1492)

* refactor: Remove proactive imports from top-level __init__.py

Branch: LazyImports

Signed-off-by: Gabe Goodhart <[email protected]>

* feat: Add client_mode to settings

This allows client-side apps such as the CLI to use the settings without
all of the server-side initialization and validation.

Branch: LazySettings

Signed-off-by: Gabe Goodhart <[email protected]>

* feat: Create a lazy wrapper around the `settings` singleton

The underlying object will only be instantiated when an attribute is
accessed. This will enable non-server uses of the library to carefully set
up settings before triggering the creation of the singleton.

NOTE: We still need to plumb-out all of the import-time settings use!

Branch: LazySettings

Signed-off-by: Gabe Goodhart <[email protected]>

* fix: Fix warning skipping for settings in client mode

Branch: LazyImports

Signed-off-by: Gabe Goodhart <[email protected]>

* test: Re-clear cache after testing LRU with the Settings class mocked

Branch: LazyImports

Signed-off-by: Gabe Goodhart <[email protected]>

* fix: Fix flake8 docstring issues

Also remove the ability to give an arbitrary get_settings function to the
lazy wrapper. This for some reason was not passing flake8 even with the
docstring fixed.

Branch: LazyImports

Signed-off-by: Gabe Goodhart <[email protected]>

* fix: Correct type annotation for get_settings kwargs parameter

The **kwargs type hint was incorrectly specified as Dict[str, Any],
which implies each value is a Dict. The correct annotation is Any,
indicating each value in kwargs can be of any type.

Signed-off-by: Mihai Criveti <[email protected]>

---------

Signed-off-by: Gabe Goodhart <[email protected]>
Signed-off-by: Mihai Criveti <[email protected]>
Co-authored-by: Mihai Criveti <[email protected]>

Author: gabe-l-hart

mcpgateway/__init__.py

@@ -15,15 +15,3 @@
 __url__ = "https://ibm.github.io/mcp-context-forge/"
 __download_url__ = "https://github.com/IBM/mcp-context-forge"
 __packages__ = ["mcpgateway"]
-
-from mcpgateway import reverse_proxy, wrapper, translate
-
-# Export main components for easier imports
-__all__ = [
-    "__version__",
-    "__author__",
-    "__license__",
-    "reverse_proxy",
-    "wrapper",
-    "translate",
-]

mcpgateway/config.py

@@ -152,6 +152,7 @@ class Settings(BaseSettings):
     app_name: str = "MCP_Gateway"
     host: str = "127.0.0.1"
     port: PositiveInt = Field(default=4444, ge=1, le=65535)
+    client_mode: bool = False
     docs_allow_basic_auth: bool = False  # Allow basic auth for docs
     database_url: str = "sqlite:///./mcp.db"
 
@@ -448,28 +449,30 @@ def validate_secrets(cls, v: Any, info: ValidationInfo) -> SecretStr:
             value = str(v)
 
         # Check for default/weak secrets
-        weak_secrets = ["my-test-key", "my-test-salt", "changeme", "secret", "password"]
-        if value.lower() in weak_secrets:
-            logger.warning(f"🔓 SECURITY WARNING - {field_name}: Default/weak secret detected! Please set a strong, unique value for production.")
+        if not info.data.get("client_mode"):
+            weak_secrets = ["my-test-key", "my-test-salt", "changeme", "secret", "password"]
+            if value.lower() in weak_secrets:
+                logger.warning(f"🔓 SECURITY WARNING - {field_name}: Default/weak secret detected! Please set a strong, unique value for production.")
 
-        # Check minimum length
-        if len(value) < 32:
-            logger.warning(f"⚠️  SECURITY WARNING - {field_name}: Secret should be at least 32 characters long. Current length: {len(value)}")
+            # Check minimum length
+            if len(value) < 32:
+                logger.warning(f"⚠️  SECURITY WARNING - {field_name}: Secret should be at least 32 characters long. Current length: {len(value)}")
 
-        # Basic entropy check (at least 10 unique characters)
-        if len(set(value)) < 10:
-            logger.warning(f"🔑 SECURITY WARNING - {field_name}: Secret has low entropy. Consider using a more random value.")
+            # Basic entropy check (at least 10 unique characters)
+            if len(set(value)) < 10:
+                logger.warning(f"🔑 SECURITY WARNING - {field_name}: Secret has low entropy. Consider using a more random value.")
 
         # Always return SecretStr to keep it secret-safe
         return v if isinstance(v, SecretStr) else SecretStr(value)
 
     @field_validator("basic_auth_password")
     @classmethod
-    def validate_admin_password(cls, v: str | SecretStr) -> SecretStr:
+    def validate_admin_password(cls, v: str | SecretStr, info: ValidationInfo) -> SecretStr:
         """Validate admin password meets security requirements.
 
         Args:
             v: The admin password value to validate.
+            info: ValidationInfo containing field data.
 
         Returns:
             SecretStr: The validated admin password value, wrapped as SecretStr.
@@ -480,35 +483,37 @@ def validate_admin_password(cls, v: str | SecretStr) -> SecretStr:
         else:
             value = v
 
-        if value == "changeme":  # nosec B105 - checking for default value
-            logger.warning("🔓 SECURITY WARNING: Default admin password detected! Please change the BASIC_AUTH_PASSWORD immediately.")
+        if not info.data.get("client_mode"):
+            if value == "changeme":  # nosec B105 - checking for default value
+                logger.warning("🔓 SECURITY WARNING: Default admin password detected! Please change the BASIC_AUTH_PASSWORD immediately.")
 
-        # Note: We can't access password_min_length here as it's not set yet during validation
-        # Using default value of 8 to match the field default
-        min_length = 8  # This matches the default in password_min_length field
-        if len(value) < min_length:
-            logger.warning(f"⚠️  SECURITY WARNING: Admin password should be at least {min_length} characters long. Current length: {len(value)}")
+            # Note: We can't access password_min_length here as it's not set yet during validation
+            # Using default value of 8 to match the field default
+            min_length = 8  # This matches the default in password_min_length field
+            if len(value) < min_length:
+                logger.warning(f"⚠️  SECURITY WARNING: Admin password should be at least {min_length} characters long. Current length: {len(value)}")
 
-        # Check password complexity
-        has_upper = any(c.isupper() for c in value)
-        has_lower = any(c.islower() for c in value)
-        has_digit = any(c.isdigit() for c in value)
-        has_special = bool(re.search(r'[!@#$%^&*(),.?":{}|<>]', value))
+            # Check password complexity
+            has_upper = any(c.isupper() for c in value)
+            has_lower = any(c.islower() for c in value)
+            has_digit = any(c.isdigit() for c in value)
+            has_special = bool(re.search(r'[!@#$%^&*(),.?":{}|<>]', value))
 
-        complexity_score = sum([has_upper, has_lower, has_digit, has_special])
-        if complexity_score < 3:
-            logger.warning("🔐 SECURITY WARNING: Admin password has low complexity. Should contain at least 3 of: uppercase, lowercase, digits, special characters")
+            complexity_score = sum([has_upper, has_lower, has_digit, has_special])
+            if complexity_score < 3:
+                logger.warning("🔐 SECURITY WARNING: Admin password has low complexity. Should contain at least 3 of: uppercase, lowercase, digits, special characters")
 
         # Always return SecretStr to keep it secret-safe
         return v if isinstance(v, SecretStr) else SecretStr(value)
 
     @field_validator("allowed_origins")
     @classmethod
-    def validate_cors_origins(cls, v: Any) -> set[str] | None:
+    def validate_cors_origins(cls, v: Any, info: ValidationInfo) -> set[str] | None:
         """Validate CORS allowed origins.
 
         Args:
             v: The set of allowed origins to validate.
+            info: ValidationInfo containing field data.
 
         Returns:
             set: The validated set of allowed origins.
@@ -522,35 +527,38 @@ def validate_cors_origins(cls, v: Any) -> set[str] | None:
             raise ValueError("allowed_origins must be a set or list of strings")
 
         dangerous_origins = ["*", "null", ""]
-        for origin in v:
-            if origin in dangerous_origins:
-                logger.warning(f"🌐 SECURITY WARNING: Dangerous CORS origin '{origin}' detected. Consider specifying explicit origins instead of wildcards.")
+        if not info.data.get("client_mode"):
+            for origin in v:
+                if origin in dangerous_origins:
+                    logger.warning(f"🌐 SECURITY WARNING: Dangerous CORS origin '{origin}' detected. Consider specifying explicit origins instead of wildcards.")
 
-            # Validate URL format
-            if not origin.startswith(("http://", "https://")) and origin not in dangerous_origins:
-                logger.warning(f"⚠️  SECURITY WARNING: Invalid origin format '{origin}'. Origins should start with http:// or https://")
+                # Validate URL format
+                if not origin.startswith(("http://", "https://")) and origin not in dangerous_origins:
+                    logger.warning(f"⚠️  SECURITY WARNING: Invalid origin format '{origin}'. Origins should start with http:// or https://")
 
         return set({str(origin) for origin in v})
 
     @field_validator("database_url")
     @classmethod
-    def validate_database_url(cls, v: str) -> str:
+    def validate_database_url(cls, v: str, info: ValidationInfo) -> str:
         """Validate database connection string security.
 
         Args:
             v: The database URL to validate.
+            info: ValidationInfo containing field data.
 
         Returns:
             str: The validated database URL.
         """
         # Check for hardcoded passwords in non-SQLite databases
-        if not v.startswith("sqlite"):
-            if "password" in v and any(weak in v for weak in ["password", "123", "admin", "test"]):
-                logger.warning("Potentially weak database password detected. Consider using a stronger password.")
+        if not info.data.get("client_mode"):
+            if not v.startswith("sqlite"):
+                if "password" in v and any(weak in v for weak in ["password", "123", "admin", "test"]):
+                    logger.warning("Potentially weak database password detected. Consider using a stronger password.")
 
-        # Warn about SQLite in production
-        if v.startswith("sqlite"):
-            logger.info("Using SQLite database. Consider PostgreSQL or MySQL for production.")
+            # Warn about SQLite in production
+            if v.startswith("sqlite"):
+                logger.info("Using SQLite database. Consider PostgreSQL or MySQL for production.")
 
         return v
 
@@ -561,6 +569,9 @@ def validate_security_combinations(self) -> Self:
         Returns:
             Itself.
         """
+        if self.client_mode:
+            return self
+
         # Check for dangerous combinations - only log warnings, don't raise errors
         if not self.auth_required and self.mcpgateway_ui_enabled:
             logger.warning("🔓 SECURITY WARNING: Admin UI is enabled without authentication. Consider setting AUTH_REQUIRED=true for production.")
@@ -1526,9 +1537,12 @@ def log_summary(self) -> None:
 
 
 @lru_cache()
-def get_settings() -> Settings:
+def get_settings(**kwargs: Any) -> Settings:
     """Get cached settings instance.
 
+    Args:
+        **kwargs: Keyword arguments to pass to the Settings setup.
+
     Returns:
         Settings: A cached instance of the Settings class.
 
@@ -1543,7 +1557,7 @@ def get_settings() -> Settings:
     """
     # Instantiate a fresh Pydantic Settings object,
     # loading from env vars or .env exactly once.
-    cfg = Settings()
+    cfg = Settings(**kwargs)
     # Validate that transport_type is correct; will
     # raise if mis-configured.
     cfg.validate_transport()
@@ -1565,7 +1579,24 @@ def generate_settings_schema() -> dict[str, Any]:
     return Settings.model_json_schema(mode="validation")
 
 
-settings = get_settings()
+# Lazy "instance" of settings
+class LazySettingsWrapper:
+    """Lazily initialize settings singleton on getattr"""
+
+    def __getattr__(self, key: str) -> Any:
+        """Get the real settings object and forward to it
+
+        Args:
+            key: The key to fetch from settings
+
+        Returns:
+            Any: The value of the attribute on the settings
+        """
+        return getattr(get_settings(), key)
+
+
+settings = LazySettingsWrapper()
+
 
 if __name__ == "__main__":
     if "--schema" in sys.argv:

tests/unit/mcpgateway/test_config.py

@@ -108,16 +108,19 @@ def test_get_settings_is_lru_cached(mock_settings):
     """Constructor must run only once regardless of repeated calls."""
     get_settings.cache_clear()
 
-    inst1 = MagicMock()
-    inst1.validate_transport.return_value = None
-    inst1.validate_database.return_value = None
-
-    inst2 = MagicMock()
-    mock_settings.side_effect = [inst1, inst2]
-
-    assert get_settings() is inst1
-    assert get_settings() is inst1  # cached
-    assert mock_settings.call_count == 1
+    try:
+        inst1 = MagicMock()
+        inst1.validate_transport.return_value = None
+        inst1.validate_database.return_value = None
+
+        inst2 = MagicMock()
+        mock_settings.side_effect = [inst1, inst2]
+
+        assert get_settings() is inst1
+        assert get_settings() is inst1  # cached
+        assert mock_settings.call_count == 1
+    finally:
+        get_settings.cache_clear()
 
 
 # --------------------------------------------------------------------------- #