Make it possible to use environment variables in mcp.json

Author: Walter Gillett

docs/mcp/client.md

@@ -187,6 +187,32 @@ 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_NAME}` syntax. 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}",
+      "args": ["run", "${MCP_MODULE}", "stdio"],
+      "env": {
+        "API_KEY": "${MY_API_KEY}"
+      }
+    },
+    "weather-api": {
+      "url": "https://${SERVER_HOST}:${SERVER_PORT}/sse"
+    }
+  }
+}
+```
+
+When loading this configuration with [`load_mcp_servers()`][pydantic_ai.mcp.load_mcp_servers], the `${VAR_NAME}` references will be replaced with the corresponding environment variable values.
+
+!!! warning
+    If a referenced environment variable is not defined, a `ValueError` will be raised. Make sure all required environment variables are set before loading the configuration.
+
 ### 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
@@ -927,9 +929,44 @@ 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.
+
+    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.
+    """
+    if isinstance(value, str):
+        # Find all environment variable references in the string
+        env_var_pattern = re.compile(r'\$\{([^}]+)\}')
+        matches = env_var_pattern.findall(value)
+
+        for var_name in matches:
+            if var_name not in os.environ:
+                raise ValueError(f'Environment variable ${{{var_name}}} is not defined')
+            value = value.replace(f'${{{var_name}}}', os.environ[var_name])
+
+        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.
+
     Args:
         config_path: The path to the configuration file.
 
@@ -939,13 +976,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.
     """
     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

@@ -1481,6 +1481,81 @@ 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'
+
+    # 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'
+
+    # 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_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'
+
+
 async def test_server_info(mcp_server: MCPServerStdio) -> None:
     with pytest.raises(
         AttributeError, match='The `MCPServerStdio.server_info` is only instantiated after initialization.'