Support mcp.json environment variable expansion in load_mcp_servers()

docs/mcp/client.md

@@ -187,6 +187,37 @@ The configuration file should be a JSON file with an `mcpServers` object contain
 
     We made this decision given that the SSE transport is deprecated.
 
+### Environment Variables
+
+The configuration file supports environment variable expansion using the `${VAR}` and `${VAR:-default}` syntax,
+[like Claude Code](https://code.claude.com/docs/en/mcp#environment-variable-expansion-in-mcp-json).
+This is useful for keeping sensitive information like API keys or host names out of your configuration files:
+
+```json {title="mcp_config_with_env.json"}
+{
+  "mcpServers": {
+    "python-runner": {
+      "command": "${PYTHON_CMD:-python3}",
+      "args": ["run", "${MCP_MODULE}", "stdio"],
+      "env": {
+        "API_KEY": "${MY_API_KEY}"
+      }
+    },
+    "weather-api": {
+      "url": "https://${SERVER_HOST:-localhost}:${SERVER_PORT:-8080}/sse"
+    }
+  }
+}
+```
+
+When loading this configuration with [`load_mcp_servers()`][pydantic_ai.mcp.load_mcp_servers]:
+
+- `${VAR}` references will be replaced with the corresponding environment variable values.
+- `${VAR:-default}` references will use the environment variable value if set, otherwise the default value.
+
+!!! warning
+    If a referenced environment variable using `${VAR}` syntax is not defined, a `ValueError` will be raised. Use the `${VAR:-default}` syntax to provide a fallback value.
+
 ### Usage
 
 ```python {title="mcp_config_loader.py" test="skip"}

pydantic_ai_slim/pydantic_ai/mcp.py

@@ -2,6 +2,8 @@
 
 import base64
 import functools
+import os
+import re
 import warnings
 from abc import ABC, abstractmethod
 from asyncio import Lock
@@ -51,6 +53,13 @@
     )
 )
 
+# Environment variable expansion pattern
+# Supports both ${VAR_NAME} and ${VAR_NAME:-default} syntax
+# Group 1: variable name
+# Group 2: the ':-' separator (to detect if default syntax is used)
+# Group 3: the default value (can be empty)
+_ENV_VAR_PATTERN = re.compile(r'\$\{([^}:]+)(:-([^}]*))?\}')
+
 
 class MCPServer(AbstractToolset[Any], ABC):
     """Base class for attaching agents to MCP servers.
@@ -927,9 +936,57 @@ class MCPServerConfig(BaseModel):
     ]
 
 
+def _expand_env_vars(value: Any) -> Any:
+    """Recursively expand environment variables in a JSON structure.
+
+    Environment variables can be referenced using ${VAR_NAME} syntax,
+    or ${VAR_NAME:-default} syntax to provide a default value if the variable is not set.
+
+    Args:
+        value: The value to expand (can be str, dict, list, or other JSON types).
+
+    Returns:
+        The value with all environment variables expanded.
+
+    Raises:
+        ValueError: If an environment variable is not defined and no default value is provided.
+    """
+    if isinstance(value, str):
+        # Find all environment variable references in the string
+        # Supports both ${VAR_NAME} and ${VAR_NAME:-default} syntax
+        def replace_match(match: re.Match[str]) -> str:
+            var_name = match.group(1)
+            has_default = match.group(2) is not None
+            default_value = match.group(3) if has_default else None
+
+            # Check if variable exists in environment
+            if var_name in os.environ:
+                return os.environ[var_name]
+            elif has_default:
+                # Use default value if the :- syntax was present (even if empty string)
+                return default_value or ''
+            else:
+                # No default value and variable not set - raise error
+                raise ValueError(f'Environment variable ${{{var_name}}} is not defined')
+
+        value = _ENV_VAR_PATTERN.sub(replace_match, value)
+
+        return value
+    elif isinstance(value, dict):
+        return {k: _expand_env_vars(v) for k, v in value.items()}  # type: ignore[misc]
+    elif isinstance(value, list):
+        return [_expand_env_vars(item) for item in value]  # type: ignore[misc]
+    else:
+        return value
+
+
 def load_mcp_servers(config_path: str | Path) -> list[MCPServerStdio | MCPServerStreamableHTTP | MCPServerSSE]:
     """Load MCP servers from a configuration file.
 
+    Environment variables can be referenced in the configuration file using:
+    - `${VAR_NAME}` syntax - expands to the value of VAR_NAME, raises error if not defined
+    - `${VAR_NAME:-default}` syntax - expands to VAR_NAME if set, otherwise uses the default value
+
     Args:
         config_path: The path to the configuration file.
 
@@ -939,13 +996,16 @@ def load_mcp_servers(config_path: str | Path) -> list[MCPServerStdio | MCPServer
     Raises:
         FileNotFoundError: If the configuration file does not exist.
         ValidationError: If the configuration file does not match the schema.
+        ValueError: If an environment variable referenced in the configuration is not defined and no default value is provided.
     """
     config_path = Path(config_path)
 
     if not config_path.exists():
         raise FileNotFoundError(f'Config file {config_path} not found')
 
-    config = MCPServerConfig.model_validate_json(config_path.read_bytes())
+    config_data = pydantic_core.from_json(config_path.read_bytes())
+    expanded_config_data = _expand_env_vars(config_data)
+    config = MCPServerConfig.model_validate(expanded_config_data)
 
     servers: list[MCPServerStdio | MCPServerStreamableHTTP | MCPServerSSE] = []
     for name, server in config.mcp_servers.items():

tests/test_mcp.py

@@ -1539,6 +1539,214 @@ def test_load_mcp_servers(tmp_path: Path):
         load_mcp_servers(tmp_path / 'does_not_exist.json')
 
 
+def test_load_mcp_servers_with_env_vars(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test environment variable expansion in config files."""
+    config = tmp_path / 'mcp.json'
+
+    # Test with environment variables in command
+    monkeypatch.setenv('PYTHON_CMD', 'python3')
+    monkeypatch.setenv('MCP_MODULE', 'tests.mcp_server')
+    config.write_text('{"mcpServers": {"my_server": {"command": "${PYTHON_CMD}", "args": ["-m", "${MCP_MODULE}"]}}}')
+
+    servers = load_mcp_servers(config)
+
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStdio)
+    assert server.command == 'python3'
+    assert server.args == ['-m', 'tests.mcp_server']
+    assert server.id == 'my_server'
+    assert server.tool_prefix == 'my_server'
+
+
+def test_load_mcp_servers_env_var_in_env_dict(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test environment variable expansion in env dict."""
+    config = tmp_path / 'mcp.json'
+
+    # Test with environment variables in env dict
+    monkeypatch.setenv('API_KEY', 'secret123')
+    config.write_text(
+        '{"mcpServers": {"my_server": {"command": "python", "args": ["-m", "tests.mcp_server"], '
+        '"env": {"API_KEY": "${API_KEY}"}}}}'
+    )
+
+    servers = load_mcp_servers(config)
+
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStdio)
+    assert server.env == {'API_KEY': 'secret123'}
+
+
+def test_load_mcp_servers_env_var_expansion_url(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test environment variable expansion in URL."""
+    config = tmp_path / 'mcp.json'
+
+    # Test with environment variables in URL
+    monkeypatch.setenv('SERVER_HOST', 'example.com')
+    monkeypatch.setenv('SERVER_PORT', '8080')
+    config.write_text('{"mcpServers": {"web_server": {"url": "https://${SERVER_HOST}:${SERVER_PORT}/mcp"}}}')
+
+    servers = load_mcp_servers(config)
+
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStreamableHTTP)
+    assert server.url == 'https://example.com:8080/mcp'
+
+
+def test_load_mcp_servers_undefined_env_var(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test that undefined environment variables raise an error."""
+    config = tmp_path / 'mcp.json'
+
+    # Make sure the environment variable is not set
+    monkeypatch.delenv('UNDEFINED_VAR', raising=False)
+
+    config.write_text('{"mcpServers": {"my_server": {"command": "${UNDEFINED_VAR}", "args": []}}}')
+
+    with pytest.raises(ValueError, match='Environment variable \\$\\{UNDEFINED_VAR\\} is not defined'):
+        load_mcp_servers(config)
+
+
+def test_load_mcp_servers_partial_env_vars(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test environment variables in partial strings."""
+    config = tmp_path / 'mcp.json'
+
+    monkeypatch.setenv('HOST', 'example.com')
+    monkeypatch.setenv('PATH_SUFFIX', 'mcp')
+    config.write_text('{"mcpServers": {"server": {"url": "https://${HOST}/api/${PATH_SUFFIX}"}}}')
+
+    servers = load_mcp_servers(config)
+
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStreamableHTTP)
+    assert server.url == 'https://example.com/api/mcp'
+
+
+def test_load_mcp_servers_with_non_string_values(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test that non-string primitive values (int, bool, null) in nested structures are passed through unchanged."""
+    config = tmp_path / 'mcp.json'
+
+    # Create a config with environment variables and extra fields containing primitives
+    # The extra fields will be ignored during validation but go through _expand_env_vars
+    monkeypatch.setenv('PYTHON_CMD', 'python')
+    config.write_text(
+        '{"mcpServers": {"my_server": {"command": "${PYTHON_CMD}", "args": ["-m", "tests.mcp_server"], '
+        '"metadata": {"count": 42, "enabled": true, "value": null}}}}'
+    )
+
+    # This should successfully expand env vars and ignore the metadata field
+    servers = load_mcp_servers(config)
+
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStdio)
+    assert server.command == 'python'
+
+
+def test_load_mcp_servers_with_default_values(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test ${VAR:-default} syntax for environment variable expansion."""
+    config = tmp_path / 'mcp.json'
+
+    # Test with undefined variable using default
+    monkeypatch.delenv('UNDEFINED_VAR', raising=False)
+    config.write_text('{"mcpServers": {"server": {"command": "${UNDEFINED_VAR:-python3}", "args": []}}}')
+
+    servers = load_mcp_servers(config)
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStdio)
+    assert server.command == 'python3'
+
+    # Test with defined variable (should use actual value, not default)
+    monkeypatch.setenv('DEFINED_VAR', 'actual_value')
+    config.write_text('{"mcpServers": {"server": {"command": "${DEFINED_VAR:-default_value}", "args": []}}}')
+
+    servers = load_mcp_servers(config)
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStdio)
+    assert server.command == 'actual_value'
+
+    # Test with empty string as default
+    monkeypatch.delenv('UNDEFINED_VAR', raising=False)
+    config.write_text('{"mcpServers": {"server": {"command": "${UNDEFINED_VAR:-}", "args": []}}}')
+
+    servers = load_mcp_servers(config)
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStdio)
+    assert server.command == ''
+
+
+def test_load_mcp_servers_with_default_values_in_url(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test ${VAR:-default} syntax in URLs."""
+    config = tmp_path / 'mcp.json'
+
+    # Test with default values in URL
+    monkeypatch.delenv('HOST', raising=False)
+    monkeypatch.setenv('PROTOCOL', 'https')
+    config.write_text('{"mcpServers": {"server": {"url": "${PROTOCOL:-http}://${HOST:-localhost}:${PORT:-8080}/mcp"}}}')
+
+    servers = load_mcp_servers(config)
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStreamableHTTP)
+    assert server.url == 'https://localhost:8080/mcp'
+
+
+def test_load_mcp_servers_with_default_values_in_env_dict(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test ${VAR:-default} syntax in env dictionary."""
+    config = tmp_path / 'mcp.json'
+
+    monkeypatch.delenv('API_KEY', raising=False)
+    monkeypatch.setenv('CUSTOM_VAR', 'custom_value')
+    config.write_text(
+        '{"mcpServers": {"server": {"command": "python", "args": [], '
+        '"env": {"API_KEY": "${API_KEY:-default_key}", "CUSTOM": "${CUSTOM_VAR:-fallback}"}}}}'
+    )
+
+    servers = load_mcp_servers(config)
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStdio)
+    assert server.env == {'API_KEY': 'default_key', 'CUSTOM': 'custom_value'}
+
+
+def test_load_mcp_servers_with_complex_default_values(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test ${VAR:-default} syntax with special characters in default."""
+    config = tmp_path / 'mcp.json'
+
+    monkeypatch.delenv('PATH_VAR', raising=False)
+    # Test default with slashes, dots, and dashes
+    config.write_text('{"mcpServers": {"server": {"command": "${PATH_VAR:-/usr/local/bin/python-3.10}", "args": []}}}')
+
+    servers = load_mcp_servers(config)
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStdio)
+    assert server.command == '/usr/local/bin/python-3.10'
+
+
+def test_load_mcp_servers_with_mixed_syntax(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Test mixing ${VAR} and ${VAR:-default} syntax in the same config."""
+    config = tmp_path / 'mcp.json'
+
+    monkeypatch.setenv('REQUIRED_VAR', 'required_value')
+    monkeypatch.delenv('OPTIONAL_VAR', raising=False)
+    config.write_text(
+        '{"mcpServers": {"server": {"command": "${REQUIRED_VAR}", "args": ["${OPTIONAL_VAR:-default_arg}"]}}}'
+    )
+
+    servers = load_mcp_servers(config)
+    assert len(servers) == 1
+    server = servers[0]
+    assert isinstance(server, MCPServerStdio)
+    assert server.command == 'required_value'
+    assert server.args == ['default_arg']
+
+
 async def test_server_info(mcp_server: MCPServerStdio) -> None:
     with pytest.raises(
         AttributeError, match='The `MCPServerStdio.server_info` is only instantiated after initialization.'