diff --git a/docs/mcp/client.md b/docs/mcp/client.md index 59f96bf865..71e15b77a2 100644 --- a/docs/mcp/client.md +++ b/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"} diff --git a/pydantic_ai_slim/pydantic_ai/mcp.py b/pydantic_ai_slim/pydantic_ai/mcp.py index 778a0f26f2..3d2b607b3f 100644 --- a/pydantic_ai_slim/pydantic_ai/mcp.py +++ b/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(): diff --git a/tests/test_mcp.py b/tests/test_mcp.py index fc7a8e5dc2..9d0654f9df 100644 --- a/tests/test_mcp.py +++ b/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.'