feat: Add environment variable expansion in config files

docs/index.md

@@ -947,7 +947,7 @@ For servers requiring special setup:
    ```json
    {
      "name": "docker-server",
-     "source": "docker://myimage:tag", 
+     "source": "docker://myimage:tag",
      "prefix": "myimage",
      "command": "docker run -i myimage:tag",
      "env": {
@@ -956,6 +956,52 @@ For servers requiring special setup:
    }
    ```
 
+### Environment Variable Expansion
+
+Magg automatically expands environment variables in the `env`, `transport.headers`, and `transport.auth` fields when loading configuration. This allows you to keep secrets and sensitive data out of config files.
+
+**Supported Syntax:**
+- `${VAR}` - Expands to the value of VAR (or stays as-is if VAR doesn't exist)
+- `${VAR:-default}` - Expands to the value of VAR, or 'default' if VAR is unset/empty
+
+**Example Configuration:**
+```json
+{
+  "servers": {
+    "remote-api": {
+      "source": "https://example.com/api-server",
+      "uri": "https://api.example.com/mcp",
+      "transport": {
+        "auth": "Bearer ${API_TOKEN}",
+        "headers": {
+          "X-Environment": "${ENVIRONMENT:-production}"
+        }
+      }
+    },
+    "local-server": {
+      "source": "https://example.com/local-server",
+      "command": "python",
+      "args": ["server.py"],
+      "env": {
+        "DATABASE_URL": "${DATABASE_URL}",
+        "API_KEY": "${API_KEY}",
+        "LOG_LEVEL": "${LOG_LEVEL:-info}"
+      }
+    }
+  }
+}
+```
+
+**Set environment variables before running Magg:**
+```bash
+export API_TOKEN="secret_token_123"
+export DATABASE_URL="postgresql://localhost/mydb"
+export API_KEY="my_api_key"
+magg serve
+```
+
+This approach keeps secrets out of version control while maintaining readable configuration files.
+
 ### Debugging
 
 Enable debug logging to troubleshoot issues:

magg/settings.py

@@ -12,7 +12,7 @@
 from pydantic import field_validator, Field, model_validator, AnyUrl, BaseModel
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-from .util.system import get_project_root
+from .util.system import get_project_root, expand_env_vars, expand_env_vars_in_dict
 from .util.paths import get_contrib_paths
 
 __all__ = "ServerConfig", "MaggConfig", "ConfigManager", "AuthConfig", "BearerAuthConfig", "ClientSettings", "KitInfo"
@@ -355,6 +355,19 @@ def load_config(self) -> MaggConfig:
             for name, server_data in data.pop('servers', {}).items():
                 try:
                     server_data['name'] = name
+
+                    # Expand environment variables in env field
+                    if 'env' in server_data and isinstance(server_data['env'], dict):
+                        server_data['env'] = expand_env_vars_in_dict(server_data['env'])
+
+                    # Expand environment variables in transport.headers field
+                    if 'transport' in server_data and isinstance(server_data['transport'], dict):
+                        if 'headers' in server_data['transport'] and isinstance(server_data['transport']['headers'], dict):
+                            server_data['transport']['headers'] = expand_env_vars_in_dict(server_data['transport']['headers'])
+                        # Expand environment variables in transport.auth field
+                        if 'auth' in server_data['transport'] and isinstance(server_data['transport']['auth'], str):
+                            server_data['transport']['auth'] = expand_env_vars(server_data['transport']['auth'])
+
                     servers[name] = ServerConfig.model_validate(server_data)
                 except Exception as e:
                     self.logger.error("Error loading server %r: %s", name, e)

magg/util/system.py

@@ -1,4 +1,6 @@
+"""System utilities for Magg - terminal initialization, paths, and environment handling."""
 import os
+import re
 import sys
 from pathlib import Path
 from typing import Optional
@@ -11,7 +13,14 @@
 except (ImportError, ModuleNotFoundError):
     pass
 
-__all__ = "initterm", "is_subdirectory", "get_project_root", "get_subprocess_environment",
+__all__ = (
+    "initterm",
+    "is_subdirectory",
+    "get_project_root",
+    "get_subprocess_environment",
+    "expand_env_vars",
+    "expand_env_vars_in_dict",
+)
 
 
 def initterm(**kwds) -> Optional["console.Console"]:
@@ -75,3 +84,74 @@ def get_subprocess_environment(*, inherit: bool = False, provided: dict | None =
         env.update(provided)
 
     return env
+
+
+def expand_env_vars(value: str) -> str:
+    """Expand environment variables in a string.
+
+    Supports the following formats:
+    - ${VAR} - expands to the value of VAR (or stays as-is if VAR doesn't exist)
+    - ${VAR:-default} - expands to the value of VAR, or 'default' if VAR is unset
+
+    Args:
+        value: String potentially containing environment variable references
+
+    Returns:
+        String with environment variables expanded
+
+    Examples:
+        >>> os.environ['API_KEY'] = 'secret123'
+        >>> expand_env_vars('Bearer ${API_KEY}')
+        'Bearer secret123'
+        >>> expand_env_vars('${MISSING:-default}')
+        'default'
+    """
+    if not isinstance(value, str):
+        return value
+
+    # Pattern for ${VAR:-default} or ${VAR}
+    def replace_braced(match):
+        var_expr = match.group(1)
+        if ':-' in var_expr:
+            var_name, default = var_expr.split(':-', 1)
+            return os.environ.get(var_name.strip(), default)
+        return os.environ.get(var_expr, match.group(0))
+
+    # Handle ${VAR:-default} and ${VAR}
+    return re.sub(r'\$\{([^}]+)\}', replace_braced, value)
+
+
+def expand_env_vars_in_dict(data: dict) -> dict:
+    """Recursively expand environment variables in dictionary string values.
+
+    Args:
+        data: Dictionary potentially containing string values with env var references
+
+    Returns:
+        Dictionary with all string values expanded
+
+    Examples:
+        >>> os.environ['TOKEN'] = 'abc123'
+        >>> expand_env_vars_in_dict({'auth': 'Bearer ${TOKEN}', 'count': 5})
+        {'auth': 'Bearer abc123', 'count': 5}
+    """
+    if not isinstance(data, dict):
+        return data
+
+    result = {}
+    for key, value in data.items():
+        if isinstance(value, str):
+            result[key] = expand_env_vars(value)
+        elif isinstance(value, dict):
+            result[key] = expand_env_vars_in_dict(value)
+        elif isinstance(value, list):
+            result[key] = [
+                expand_env_vars(item) if isinstance(item, str)
+                else expand_env_vars_in_dict(item) if isinstance(item, dict)
+                else item
+                for item in value
+            ]
+        else:
+            result[key] = value
+
+    return result

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "magg"
-version = "0.10.1"
+version = "0.11.0"
 requires-python = ">=3.12"
 description = "MCP Aggregator"
 authors = [{ name = "Phillip Sitbon", email = "phillip.sitbon@gmail.com"}]

readme.md

@@ -342,6 +342,50 @@ Example configuration:
 }
 ```
 
+#### Environment Variable Expansion in Configuration
+
+Magg automatically expands environment variables in the `env`, `transport.headers`, and `transport.auth` fields when loading configuration. This allows you to keep secrets and sensitive data out of config files.
+
+**Supported Syntax:**
+- `${VAR}` - Expands to the value of VAR (or stays as-is if VAR doesn't exist)
+- `${VAR:-default}` - Expands to the value of VAR, or 'default' if VAR is unset/empty
+
+**Example:**
+```json
+{
+  "servers": {
+    "remote-api": {
+      "source": "https://example.com/api-server",
+      "uri": "https://api.example.com/mcp",
+      "transport": {
+        "auth": "Bearer ${API_TOKEN}",
+        "headers": {
+          "X-Environment": "${ENVIRONMENT:-production}"
+        }
+      }
+    },
+    "local-server": {
+      "source": "https://example.com/local-server",
+      "command": "python",
+      "args": ["server.py"],
+      "env": {
+        "DATABASE_URL": "${DATABASE_URL}",
+        "API_KEY": "${API_KEY}",
+        "LOG_LEVEL": "${LOG_LEVEL:-info}"
+      }
+    }
+  }
+}
+```
+
+Then set your environment variables before running Magg:
+```bash
+export API_TOKEN="secret_token_123"
+export DATABASE_URL="postgresql://localhost/mydb"
+export API_KEY="my_api_key"
+magg serve
+```
+
 ### Adding Servers
 
 Servers can be added in several ways:

test/magg/test_config.py

@@ -244,3 +244,146 @@ def test_load_invalid_config(self):
 
             # Should return empty config on error
             assert config.servers == {}
+
+    def test_env_var_expansion_in_env_field(self):
+        """Test that environment variables are expanded in server env field."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            config_path = Path(tmpdir) / "config.json"
+
+            # Set environment variable for testing
+            with patch.dict(os.environ, {"TEST_API_KEY": "secret123", "TEST_DB": "prod"}):
+                # Create config with env vars
+                config_data = {
+                    "servers": {
+                        "testserver": {
+                            "source": "https://example.com",
+                            "command": "python",
+                            "args": ["server.py"],
+                            "env": {
+                                "API_KEY": "${TEST_API_KEY}",
+                                "DATABASE": "${TEST_DB}",
+                                "FALLBACK": "${MISSING_VAR:-default_value}"
+                            }
+                        }
+                    }
+                }
+
+                with open(config_path, 'w') as f:
+                    json.dump(config_data, f)
+
+                # Load config
+                manager = ConfigManager(str(config_path))
+                config = manager.load_config()
+
+                # Check that env vars were expanded
+                server = config.servers["testserver"]
+                assert server.env["API_KEY"] == "secret123"
+                assert server.env["DATABASE"] == "prod"
+                assert server.env["FALLBACK"] == "default_value"
+
+    def test_env_var_expansion_in_transport_headers(self):
+        """Test that environment variables are expanded in transport headers."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            config_path = Path(tmpdir) / "config.json"
+
+            # Set environment variable for testing
+            with patch.dict(os.environ, {"AUTH_TOKEN": "bearer_token_123"}):
+                # Create config with transport headers
+                config_data = {
+                    "servers": {
+                        "httpserver": {
+                            "source": "https://example.com",
+                            "uri": "https://api.example.com/mcp",
+                            "transport": {
+                                "headers": {
+                                    "Authorization": "Bearer ${AUTH_TOKEN}",
+                                    "X-Custom": "${CUSTOM_HEADER:-default}"
+                                }
+                            }
+                        }
+                    }
+                }
+
+                with open(config_path, 'w') as f:
+                    json.dump(config_data, f)
+
+                # Load config
+                manager = ConfigManager(str(config_path))
+                config = manager.load_config()
+
+                # Check that env vars were expanded
+                server = config.servers["httpserver"]
+                assert server.transport["headers"]["Authorization"] == "Bearer bearer_token_123"
+                assert server.transport["headers"]["X-Custom"] == "default"
+
+    def test_env_var_expansion_in_transport_auth(self):
+        """Test that environment variables are expanded in transport auth."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            config_path = Path(tmpdir) / "config.json"
+
+            # Set environment variable for testing
+            with patch.dict(os.environ, {"API_TOKEN": "secret_token_xyz"}):
+                # Create config with transport auth
+                config_data = {
+                    "servers": {
+                        "apiserver": {
+                            "source": "https://example.com",
+                            "uri": "https://api.example.com/mcp",
+                            "transport": {
+                                "auth": "Bearer ${API_TOKEN}"
+                            }
+                        },
+                        "fallbackserver": {
+                            "source": "https://example.com",
+                            "uri": "https://api2.example.com/mcp",
+                            "transport": {
+                                "auth": "${MISSING_TOKEN:-default_token}"
+                            }
+                        }
+                    }
+                }
+
+                with open(config_path, 'w') as f:
+                    json.dump(config_data, f)
+
+                # Load config
+                manager = ConfigManager(str(config_path))
+                config = manager.load_config()
+
+                # Check that env vars were expanded
+                server1 = config.servers["apiserver"]
+                assert server1.transport["auth"] == "Bearer secret_token_xyz"
+
+                server2 = config.servers["fallbackserver"]
+                assert server2.transport["auth"] == "default_token"
+
+    def test_env_var_expansion_missing_var(self):
+        """Test that missing env vars are left unexpanded without defaults."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            config_path = Path(tmpdir) / "config.json"
+
+            # Make sure the var doesn't exist
+            with patch.dict(os.environ, {}, clear=False):
+                if "NONEXISTENT_VAR" in os.environ:
+                    del os.environ["NONEXISTENT_VAR"]
+
+                config_data = {
+                    "servers": {
+                        "testserver": {
+                            "source": "https://example.com",
+                            "env": {
+                                "SHOULD_STAY": "${NONEXISTENT_VAR}"
+                            }
+                        }
+                    }
+                }
+
+                with open(config_path, 'w') as f:
+                    json.dump(config_data, f)
+
+                manager = ConfigManager(str(config_path))
+                config = manager.load_config()
+
+                # Should keep the original unexpanded value
+                server = config.servers["testserver"]
+                assert server.env["SHOULD_STAY"] == "${NONEXISTENT_VAR}"