diff --git a/README.md b/README.md index d76d3d267..f2e0c2534 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,7 @@ - [Advanced Usage](#advanced-usage) - [Low-Level Server](#low-level-server) - [Writing MCP Clients](#writing-mcp-clients) + - [API for Client Configuration](#api-for-client-configuration) - [MCP Primitives](#mcp-primitives) - [Server Capabilities](#server-capabilities) - [Documentation](#documentation) @@ -862,6 +863,11 @@ async def main(): For a complete working example, see [`examples/clients/simple-auth-client/`](examples/clients/simple-auth-client/). +### API for Client Configuration + +The MCP Python SDK provides an API for client configuration. This lets client applications easily load MCP servers from configuration files in a variety of formats and with some useful, built-in features. + +See the [Client Configuration Guide](docs/client-configuration.md) for complete details. ### MCP Primitives diff --git a/docs/client-configuration.md b/docs/client-configuration.md new file mode 100644 index 000000000..c01d08ce0 --- /dev/null +++ b/docs/client-configuration.md @@ -0,0 +1,348 @@ +# MCP Client Configuration (NEW) + +This guide, for client application developers, covers a new API for client +configuration. Client applications can use this API to get info about configured +MCP servers from configuration files + +## Why should my application use this API? + +- Eliminate the need to write and maintain code to parse configuration files +- Your application can easily benefit from bug fixes and new features related to configuration +- Allows your application to support features that other applications may have + and which your application does not. E.g., + + - Allow specifying the entire command in the `command` field (not having to + specify an `args` list), which makes it easier for users to manage + - Allow comments in JSON configuration files + - Input variables (as supported by VS Code), plus validation of required inputs + and interpolation of input values + - YAML configuration files, which are more readable and easier to write than JSON + +- If every application that uses MCP supported this API, it would lead to + greater consistency in how MCP servers are configured and used, which is a + tremendous win for users and a benefit to the MCP ecosystem. Note: This is the + Python SDK, but hopefully this can be ported to the SDKs for other languages. + +## Loading Configuration Files + +```python +from mcp.client.config.mcp_servers_config import MCPServersConfig + +# Load JSON +config = MCPServersConfig.from_file("~/.cursor/mcp.json") +config = MCPServersConfig.from_file("~/Library/Application\ Support/Claude/claude_desktop_config.json") + +# Load YAML (auto-detected by extension) +config = MCPServersConfig.from_file("~/.cursor/mcp.yaml") # Not yet supported in Cursor but maybe soon...?! +config = MCPServersConfig.from_file("~/Library/Application\ Support/Claude/claude_desktop_config.yaml") # Maybe someday...?! + +config = MCPServersConfig.from_file(".vscode/mcp.json") + +mcp_server = config.server("time") +print(mcp_server.command) +print(mcp_server.args) +print(mcp_server.env) +print(mcp_server.headers) +print(mcp_server.inputs) +print(mcp_server.isActive) +print(mcp_server.effective_command) +print(mcp_server.effective_args) +``` + +## Configuration File Formats + +MCP supports multiple configuration file formats for maximum flexibility. + +### JSON Configuration + +```json +{ + "mcpServers": { + "time": { + "command": "uvx", + "args": ["mcp-server-time"] + }, + "filesystem": { + "command": "npx", + "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/username/Desktop"] + } + } +} +``` + +This is a typical JSON configuration file for an MCP server in that it has +`command` and `args` (as a list) fields. + +Users can also specify the entire command in the `command` field, which +makes it easier to read and write. Internally, the library splits the command +into `command` and `args` fields, so the result is a nicer user experience and +no application code needs to change. + +```json +{ + "mcpServers": { + "time": { + "command": "uvx mcp-server-time" + }, + "filesystem": { + "command": "npx -y @modelcontextprotocol/server-filesystem /Users/username/Desktop" + } + } +} +``` + +JSON is the most commonly used format for MCP servers, but it has some +limitations, which is why subsequent sections cover other formats, such as JSONC +and YAML. + +### JSON with Comments (JSONC) + +The API supports JSON files with `//` comments (JSONC), which is very commonly +used in the VS Code ecosystem: + +```jsonc +{ + "mcpServers": { + // Can get current time in various timezones + "time": { + "command": "uvx mcp-server-time" + }, + + // Can get the contents of the user's desktop + "filesystem": { + "command": "npx -y @modelcontextprotocol/server-filesystem /Users/username/Desktop" + } + } +} +``` + +### YAML Configuration + +The API supports YAML configuration files, which offer improved readability, +comments, and the ability to completely sidestep issues with commas, that are +common when working with JSON. + +```yaml +mcpServers: + # Can get current time in various timezones + time: + command: uvx mcp-server-time + + # Can get the contents of the user's desktop + filesystem: + command: npx -y @modelcontextprotocol/server-filesystem /Users/username/Desktop +``` + +**Installation**: YAML support requires the optional dependency: + +```bash +pip install "mcp[yaml]" +``` + +## Server Types and Auto-Detection + +MCP automatically infers server types based on configuration fields when the +`type` field is omitted: + +### Stdio Servers + +Servers with a `command` field are automatically detected as `stdio` type: + +```yaml +mcpServers: + python-server: + command: python -m my_server + # type: stdio (auto-inferred) +``` + +### Streamable HTTP Servers + +Servers with a `url` field (without SSE keywords) are detected as +`streamable_http` type: + +```yaml +mcpServers: + api-server: + url: https://api.example.com/mcp + # type: streamable_http (auto-inferred) +``` + +### SSE Servers + +Servers with a `url` field containing "sse" in the URL, name, or description are +detected as `sse` type: + +```yaml +mcpServers: + sse-server: + url: https://api.example.com/sse + # type: sse (auto-inferred due to "sse" in URL) + + event-server: + url: https://api.example.com/events + description: "SSE-based event server" + # type: sse (auto-inferred due to "SSE" in description) +``` + +## Input Variables and Substitution + +MCP supports dynamic configuration using input variables, which is a feature +that VS Code supports. This works in both JSON and YAML configurations. + +### Declaring Inputs (JSON) + +```json +{ + "inputs": [ + { + "id": "api-key", + "type": "promptString", + "description": "Your API key", + "password": true + }, + { + "id": "server-host", + "type": "promptString", + "description": "Server hostname" + } + ], + "servers": { + "dynamic-server": { + "url": "https://${input:server-host}/mcp", + "headers": { + "Authorization": "Bearer ${input:api-key}" + } + } + } +} +``` + +### Declaring Inputs (YAML) + +```yaml +inputs: + - id: api-key + type: promptString + description: "Your API key" + password: true + - id: server-host + type: promptString + description: "Server hostname" + +servers: + dynamic-server: + url: https://${input:server-host}/mcp + headers: + Authorization: Bearer ${input:api-key} +``` + +### Getting Declared Inputs + +The application can use the `inputs` field to get the declared inputs and +prompt the user for the values or otherwise allow them to be specified. + +The application gets the declared inputs by doing: + +```python +config = MCPServersConfig.from_file(".vscode/mcp.json") +for input in config.inputs: + # Prompt the user for the value + ... +``` + +### Using Inputs + +When loading the configuration, provide input values: + +```python +from mcp.client.config.mcp_servers_config import MCPServersConfig + +config = MCPServersConfig.from_file("config.yaml") + +# Substitute input values into the configuration +server = config.server( + "dynamic-server", + input_values={"api-key": "secret-key-123", "server-host": "api.example.com"}, +) +``` + +### Input Validation + +MCP validates that all required inputs are provided: + +```python +# Check required inputs +required_inputs = config.get_required_inputs() +print(f"Required inputs: {required_inputs}") + +# Validate provided inputs +missing_inputs = config.validate_inputs(provided_inputs) +if missing_inputs: + print(f"Missing required inputs: {missing_inputs}") +``` + +## Configuration Schema + +### Server Configuration Base Fields + +All server types support these common optionalfields: + +- `name` (string, optional): Display name for the server +- `description` (string, optional): Server description +- `isActive` (boolean, default: true): Whether the server is active + +### Stdio Server Configuration + +```yaml +mcpServers: + stdio-server: + type: stdio # Optional if 'command' is present + command: python -m my_server + args: # Optional additional arguments + - --debug + - --port=8080 + env: # Optional environment variables + DEBUG: "true" + API_KEY: secret123 +``` + +### Streamable HTTP Server Configuration + +```yaml +mcpServers: + http-server: + type: streamable_http # Optional if 'url' is present + url: https://api.example.com/mcp + headers: # Optional HTTP headers + Authorization: Bearer token123 + X-Custom-Header: value +``` + +### SSE Server Configuration + +```yaml +mcpServers: + sse-server: + type: sse + url: https://api.example.com/sse + headers: # Optional HTTP headers + Authorization: Bearer token123 +``` + +## Field Aliases + +MCP supports both traditional and modern field names: + +- `mcpServers` (most common) or `servers` (VS Code) + +```yaml +# More common format +mcpServers: + my-server: + command: python -m server + +# VS Code format (equivalent) +servers: + my-server: + command: python -m server +``` diff --git a/mkdocs.yml b/mkdocs.yml index b907cb873..e17e64b61 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -12,6 +12,7 @@ site_url: https://modelcontextprotocol.github.io/python-sdk nav: - Home: index.md + - Client Configuration: client-configuration.md - API Reference: api.md theme: diff --git a/pyproject.toml b/pyproject.toml index 9ad50ab58..7225c597f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,7 @@ dependencies = [ rich = ["rich>=13.9.4"] cli = ["typer>=0.12.4", "python-dotenv>=1.0.0"] ws = ["websockets>=15.0.1"] +yaml = ["pyyaml>=6.0.2"] [project.scripts] mcp = "mcp.cli:app [cli]" diff --git a/src/mcp/client/config/mcp_servers_config.py b/src/mcp/client/config/mcp_servers_config.py new file mode 100644 index 000000000..de1a172a6 --- /dev/null +++ b/src/mcp/client/config/mcp_servers_config.py @@ -0,0 +1,382 @@ +"""Configuration management for MCP servers. + +This module provides comprehensive configuration management for MCP (Model +Context Protocol) servers, supporting multiple file formats and advanced +features: + +Features: + +- Multiple file formats: JSON, JSONC (JSON with comments), and YAML (.yaml/.yml) + +- Automatic server type inference based on configuration fields + +- Input variable substitution with ${input:key} syntax and validation of + required inputs, borrowed from VS Code + +- Support for both 'mcpServers' and 'servers' (VS Code) field names + +Supported Server Types: + +- streamable_http: HTTP-based servers using streamable transport + +- stdio: Servers that communicate via standard input/output + +- sse: Server-Sent Events based servers + +Example usage: + + # Load basic configuration + config = MCPServersConfig.from_file("config.json") + + # Load YAML with input substitution + config = MCPServersConfig.from_file( + "config.yaml", + inputs={"api-key": "secret", "host": "api.example.com"} + ) + + # Validate inputs + missing = config.validate_inputs(provided_inputs) + if missing: + raise ValueError(f"Missing inputs: {missing}") + +Dependencies: +- PyYAML: Required for YAML file support (install with 'mcp[yaml]') +""" + +# stdlib imports +import json +import os +import re +import shlex +from pathlib import Path +from typing import Annotated, Any, Literal, cast + +# third party imports +try: + import yaml +except ImportError: + yaml = None # type: ignore +from pydantic import BaseModel, Field, field_validator, model_validator + + +class InputDefinition(BaseModel): + """Definition of an input parameter.""" + + type: Literal["promptString"] = "promptString" + id: str + description: str | None = None + password: bool = False + + +class MCPServerConfig(BaseModel): + """Base class for MCP server configurations.""" + + name: str | None = None + description: str | None = None + isActive: bool = True + + +class StdioServerConfig(MCPServerConfig): + """Configuration for stdio-based MCP servers.""" + + type: Literal["stdio"] = "stdio" + command: str + args: list[str] | None = None + env: dict[str, str] | None = None + + def _parse_command(self) -> list[str]: + """Parse the command string into parts, handling quotes properly. + + Strips unnecessary whitespace and newlines to handle YAML multi-line strings. + Treats backslashes followed by newlines as line continuations. + """ + # Handle backslash line continuations by removing them and the following newline + cleaned_command = self.command.replace("\\\n", " ") + # Then normalize all whitespace (including remaining newlines) to single spaces + cleaned_command = " ".join(cleaned_command.split()) + return shlex.split(cleaned_command) + + @property + def effective_command(self) -> str: + """Get the effective command (first part of the command string).""" + return self._parse_command()[0] + + @property + def effective_args(self) -> list[str]: + """Get the effective arguments (parsed from command plus explicit args).""" + command_parts = self._parse_command() + parsed_args = command_parts[1:] if len(command_parts) > 1 else [] + explicit_args = self.args or [] + return parsed_args + explicit_args + + +class StreamableHTTPServerConfig(MCPServerConfig): + """Configuration for StreamableHTTP-based MCP servers.""" + + type: Literal["streamable_http"] = "streamable_http" + url: str + headers: dict[str, str] | None = None + + +class SSEServerConfig(MCPServerConfig): + """Configuration for SSE-based MCP servers.""" + + type: Literal["sse"] = "sse" + url: str + headers: dict[str, str] | None = None + + +# Discriminated union for different server config types +ServerConfigUnion = Annotated[ + StdioServerConfig | StreamableHTTPServerConfig | SSEServerConfig, Field(discriminator="type") +] + + +class MCPServersConfig(BaseModel): + """Configuration for multiple MCP servers. + + Note: + Direct access to the 'servers' field is discouraged. + Use the server() method instead for proper input validation and substitution. + """ + + servers: dict[str, ServerConfigUnion] + inputs: list[InputDefinition] | None = None + + def __getattribute__(self, name: str) -> Any: + """Get an attribute from the config. + + This emits a warning if the `servers` field is accessed directly. + This is to discourage direct access to the `servers` field. + Use the `server()` method instead for proper input validation and substitution. + """ + + if name == "servers": + import inspect + import warnings + + # Get the calling frame to check if it's internal access + frame = inspect.currentframe() + if frame and frame.f_back: + caller_filename = frame.f_back.f_code.co_filename + caller_function = frame.f_back.f_code.co_name + + # Don't warn for internal methods, tests, or if called from within this class + is_internal_call = ( + caller_function in ("server", "list_servers", "has_server", "__init__", "model_validate") + or "mcp_servers_config.py" in caller_filename + ) + + if not is_internal_call: + warnings.warn( + f"Direct access to 'servers' field of {self.__class__.__name__} is discouraged. " + + "Use server() method instead for proper input validation and substitution.", + UserWarning, + stacklevel=2, + ) + + return super().__getattribute__(name) + + def server( + self, + name: str, + input_values: dict[str, str] | None = None, + ) -> ServerConfigUnion: + """Get a server config by name.""" + server = self.servers[name] + + # Validate inputs if provided and input definitions exist + if input_values is not None and self.inputs: + missing_inputs = self.validate_inputs(input_values) + if missing_inputs: + descriptions: list[str] = [] + for input_id in missing_inputs: + desc = self.get_input_description(input_id) + descriptions.append(f" - {input_id}: {desc or 'No description'}") + + raise ValueError("Missing required input values:\n" + "\n".join(descriptions)) + + # Substitute input placeholders if inputs provided + if input_values: + server.__dict__ = self._substitute_inputs(server.__dict__, input_values) + + return server + + def list_servers(self) -> list[str]: + """Get a list of available server names.""" + return list(self.servers.keys()) + + def has_server(self, name: str) -> bool: + """Check if a server with the given name exists.""" + return name in self.servers + + @model_validator(mode="before") + @classmethod + def handle_field_aliases(cls, data: dict[str, Any]) -> dict[str, Any]: + """Handle both 'servers' and 'mcpServers' field names.""" + + # If 'mcpServers' exists but 'servers' doesn't, use 'mcpServers' + if "mcpServers" in data and "servers" not in data: + data["servers"] = data["mcpServers"] + del data["mcpServers"] + # If both exist, prefer 'servers' and remove 'mcpServers' + elif "mcpServers" in data and "servers" in data: + del data["mcpServers"] + + return data + + @field_validator("servers", mode="before") + @classmethod + def infer_server_types(cls, servers_data: dict[str, Any]) -> dict[str, Any]: + """Automatically infer server types when 'type' field is omitted.""" + + for server_config in servers_data.values(): + server_config = cast(dict[str, str], server_config) + sse_mentioned_in_config = ( + "sse" in server_config.get("url", "").lower() + or "sse" in server_config.get("name", "").lower() + or "sse" in server_config.get("description", "").lower() + ) + if "type" not in server_config: + # Infer type based on distinguishing fields + if "command" in server_config: + server_config["type"] = "stdio" + elif "url" in server_config and sse_mentioned_in_config: + # Could infer SSE vs streamable_http based on URL patterns in the future + server_config["type"] = "sse" + elif "url" in server_config: + server_config["type"] = "streamable_http" + + return servers_data + + def get_required_inputs(self) -> list[str]: + """Get list of input IDs that are defined in the inputs section.""" + if not self.inputs: + return [] + return [input_def.id for input_def in self.inputs] + + def validate_inputs(self, provided_inputs: dict[str, str]) -> list[str]: + """Validate provided inputs against input definitions. + + Returns list of missing required input IDs. + """ + if not self.inputs: + return [] + + required_input_ids = self.get_required_inputs() + missing_inputs = [input_id for input_id in required_input_ids if input_id not in provided_inputs] + + return missing_inputs + + def get_input_description(self, input_id: str) -> str | None: + """Get the description for a specific input ID.""" + if not self.inputs: + return None + + for input_def in self.inputs: + if input_def.id == input_id: + return input_def.description + + return None + + @classmethod + def _substitute_inputs(cls, data: Any, inputs: dict[str, str]) -> Any: + """Recursively substitute ${input:key} placeholders with values from inputs dict.""" + if isinstance(data, str): + # Replace ${input:key} patterns with values from inputs + def replace_input(match: re.Match[str]) -> str: + key = match.group(1) + if key in inputs: + return inputs[key] + else: + raise ValueError(f"Missing input value for key: '{key}'") + + return re.sub(r"\$\{input:([^}]+)\}", replace_input, data) + + elif isinstance(data, dict): + dict_result: dict[str, Any] = {} + dict_data = cast(dict[str, Any], data) + for k, v in dict_data.items(): + dict_result[k] = cls._substitute_inputs(v, inputs) + return dict_result + + elif isinstance(data, list): + list_data = cast(list[Any], data) + return [cls._substitute_inputs(item, inputs) for item in list_data] + + else: + return data + + @classmethod + def _strip_json_comments(cls, content: str) -> str: + """Strip // comments from JSON content, being careful not to remove // inside strings.""" + result: list[str] = [] + lines = content.split("\n") + + for line in lines: + # Track if we're inside a string + in_string = False + escaped = False + comment_start = -1 + + for i, char in enumerate(line): + if escaped: + escaped = False + continue + + if char == "\\" and in_string: + escaped = True + continue + + if char == '"': + in_string = not in_string + continue + + # Look for // comment start when not in string + if not in_string and char == "/" and i + 1 < len(line) and line[i + 1] == "/": + comment_start = i + break + + # If we found a comment, remove it + if comment_start != -1: + line = line[:comment_start].rstrip() + + result.append(line) + + return "\n".join(result) + + @classmethod + def from_file( + cls, + config_path: Path | str, + use_pyyaml: bool = False, + ) -> "MCPServersConfig": + """Load configuration from a JSON or YAML file. + + Args: + config_path: Path to the configuration file + use_pyyaml: If True, force use of PyYAML parser. Defaults to False. + Also automatically used for .yaml/.yml files. + inputs: Dictionary of input values to substitute for ${input:key} placeholders + """ + + config_path = os.path.expandvars(config_path) # Expand environment variables like $HOME + config_path = Path(config_path) # Convert to Path object + config_path = config_path.expanduser() # Expand ~ to home directory + + with open(config_path) as config_file: + content = config_file.read() + + # Check if YAML parsing is requested + should_use_yaml = use_pyyaml or config_path.suffix.lower() in (".yaml", ".yml") + + if should_use_yaml: + if not yaml: + raise ImportError("PyYAML is required to parse YAML files. ") + data = yaml.safe_load(content) + else: + # Strip comments from JSON content (JSONC support) + cleaned_content = cls._strip_json_comments(content) + data = json.loads(cleaned_content) + + return cls.model_validate(data) diff --git a/tests/client/config/__init__.py b/tests/client/config/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/client/config/mcp.json b/tests/client/config/mcp.json new file mode 100644 index 000000000..edfa0d45b --- /dev/null +++ b/tests/client/config/mcp.json @@ -0,0 +1,34 @@ +{ + "mcpServers": { + "stdio_server": { + "command": "python", + "args": ["-m", "my_server"], + "env": {"DEBUG": "true"} + }, + "stdio_server_with_full_command": { + "command": "python -m my_server" + }, + "stdio_server_with_full_command_and_explicit_args": { + "command": "python -m my_server", + "args": ["--debug"] + }, + "streamable_http_server_with_headers": { + "url": "https://api.example.com/mcp", + "headers": {"Authorization": "Bearer token123"} + }, + "stdio_server_with_explicit_type": { + "type": "stdio", + "command": "python", + "args": ["-m", "my_server"], + "env": {"DEBUG": "true"} + }, + "streamable_http_server_with_explicit_type": { + "type": "streamable_http", + "url": "https://api.example.com/mcp" + }, + "sse_server_with_explicit_type": { + "type": "sse", + "url": "https://api.example.com/sse" + } + } +} \ No newline at end of file diff --git a/tests/client/config/mcp.yaml b/tests/client/config/mcp.yaml new file mode 100644 index 000000000..c0fda72b5 --- /dev/null +++ b/tests/client/config/mcp.yaml @@ -0,0 +1,73 @@ +# MCP Servers Configuration (YAML format) +# This file demonstrates the same server configurations as mcp.json but in YAML format + +mcpServers: + time: + command: uvx mcp-server-time + + filesystem: + command: npx -y @modelcontextprotocol/server-filesystem /Users/username/Desktop /path/to/other/allowed/dir + + filesystem_multi_line: + command: > + npx -y @modelcontextprotocol/server-filesystem \ + /Users/username/Desktop \ + /path/to/other/allowed/dir + + # Stdio server with full command string + # The library will automatically parse this. + # The effective_command will be "python" and effective_args will be ["-m", "my_server"] + stdio_server_with_full_command: + command: python -m my_server + + # Streamable HTTP server with headers + streamable_http_server: + url: https://api.example.com/mcp + + # Streamable HTTP server with headers + streamable_http_server_with_headers: + url: https://api.example.com/mcp + headers: + Authorization: Bearer token123 + + # stdio server with explicit command and args like typically done in mcp.json + # files + # I would expect this to not be used much with YAML files, but it's here for + # completeness. + stdio_server: + command: python + args: + - -m + - my_server + env: + DEBUG: "true" + + # Stdio server with full command string AND explicit args + # The effective_args will combine parsed command args with explicit args + stdio_server_with_full_command_and_explicit_args: + command: python -m my_server # Will be parsed to: command="python", args=["-m", "my_server"] + args: + - --debug # Will be appended to parsed args: ["-m", "my_server", "--debug"] + + # Servers with explicit types - these demonstrate that type inference + # can be overridden by explicitly specifying the "type" field + + # Stdio server with explicit type specification + stdio_server_with_explicit_type: + type: stdio # Explicitly specified type + command: python + args: + - -m + - my_server + env: + DEBUG: "true" + + # Streamable HTTP server with explicit type specification + streamable_http_server_with_explicit_type: + type: streamable_http # Explicitly specified type + url: https://api.example.com/mcp + + # SSE (Server-Sent Events) server with explicit type specification + sse_server_with_explicit_type: + type: sse # Explicitly specified type + url: https://api.example.com/sse diff --git a/tests/client/config/test_mcp_servers_config.py b/tests/client/config/test_mcp_servers_config.py new file mode 100644 index 000000000..8250e7ff8 --- /dev/null +++ b/tests/client/config/test_mcp_servers_config.py @@ -0,0 +1,786 @@ +# stdlib imports +import json +from pathlib import Path + +# third party imports +import pytest + +# local imports +from mcp.client.config.mcp_servers_config import ( + MCPServersConfig, + SSEServerConfig, + StdioServerConfig, + StreamableHTTPServerConfig, +) + + +@pytest.fixture +def mcp_config_file() -> Path: + """Return path to the mcp.json config file with mixed server types""" + return Path(__file__).parent / "mcp.json" + + +def test_stdio_server(mcp_config_file: Path): + config = MCPServersConfig.from_file(mcp_config_file) + + stdio_server = config.servers["stdio_server"] + assert isinstance(stdio_server, StdioServerConfig) + + assert stdio_server.command == "python" + assert stdio_server.args == ["-m", "my_server"] + assert stdio_server.env == {"DEBUG": "true"} + assert stdio_server.type == "stdio" # Should be automatically inferred + + # In this case, effective_command and effective_args are the same as command + # and args. + # But later on, we will see a test where the command is specified as a + # single string, and we expect the command to be split into command and args + assert stdio_server.effective_command == "python" + assert stdio_server.effective_args == ["-m", "my_server"] + + +def test_stdio_server_with_explicit_type(mcp_config_file: Path): + """Test that stdio server with explicit 'type' field is respected and works correctly.""" + config = MCPServersConfig.from_file(mcp_config_file) + + stdio_server = config.servers["stdio_server_with_explicit_type"] + assert isinstance(stdio_server, StdioServerConfig) + assert stdio_server.type == "stdio" + + +def test_streamable_http_server_with_explicit_type(mcp_config_file: Path): + """Test that streamable HTTP server with explicit 'type' field is respected and works correctly.""" + config = MCPServersConfig.from_file(mcp_config_file) + + http_server = config.servers["streamable_http_server_with_explicit_type"] + assert isinstance(http_server, StreamableHTTPServerConfig) + assert http_server.type == "streamable_http" + + +def test_sse_server_with_explicit_type(mcp_config_file: Path): + """Test that SSE server with explicit 'type' field is respected and works correctly.""" + config = MCPServersConfig.from_file(mcp_config_file) + + sse_server = config.servers["sse_server_with_explicit_type"] + assert isinstance(sse_server, SSEServerConfig) + assert sse_server.type == "sse" + + +def test_stdio_server_with_full_command_should_be_split(mcp_config_file: Path): + """This test should fail - it expects the command to be split into command and args.""" + config = MCPServersConfig.from_file(mcp_config_file) + + stdio_server = config.servers["stdio_server_with_full_command"] + assert isinstance(stdio_server, StdioServerConfig) + + # This is how the command was specified + assert stdio_server.command == "python -m my_server" + + # This is how the command is split into command and args + assert stdio_server.effective_command == "python" + assert stdio_server.effective_args == ["-m", "my_server"] + + +def test_stdio_server_with_full_command_and_explicit_args(mcp_config_file: Path): + """Test that effective_args combines parsed command args with explicit args.""" + config = MCPServersConfig.from_file(mcp_config_file) + + stdio_server = config.servers["stdio_server_with_full_command_and_explicit_args"] + assert isinstance(stdio_server, StdioServerConfig) + + # Test original values + assert stdio_server.command == "python -m my_server" + assert stdio_server.args == ["--debug"] + + # Test effective values - should combine parsed command args with explicit args + assert stdio_server.effective_command == "python" + assert stdio_server.effective_args == ["-m", "my_server", "--debug"] + + +def test_streamable_http_server_with_headers(mcp_config_file: Path): + config = MCPServersConfig.from_file(mcp_config_file) + + http_server = config.servers["streamable_http_server_with_headers"] + assert isinstance(http_server, StreamableHTTPServerConfig) + + assert http_server.url == "https://api.example.com/mcp" + assert http_server.headers == {"Authorization": "Bearer token123"} + assert http_server.type == "streamable_http" # Should be automatically inferred + + +def test_stdio_server_with_quoted_arguments(): + """Test that stdio servers handle quoted arguments with spaces correctly.""" + # Test with double quotes + config_data = { + "mcpServers": { + "server_with_double_quotes": {"command": 'python -m my_server --config "path with spaces/config.json"'}, + "server_with_single_quotes": { + "command": "python -m my_server --config 'another path with spaces/config.json'" + }, + "server_with_mixed_quotes": { + "command": "python -m my_server --name \"My Server\" --path '/home/user/my path'" + }, + } + } + + config = MCPServersConfig.model_validate(config_data) + + # Test double quotes + double_quote_server = config.servers["server_with_double_quotes"] + assert isinstance(double_quote_server, StdioServerConfig) + assert double_quote_server.effective_command == "python" + expected_args_double = ["-m", "my_server", "--config", "path with spaces/config.json"] + assert double_quote_server.effective_args == expected_args_double + + # Test single quotes + single_quote_server = config.servers["server_with_single_quotes"] + assert isinstance(single_quote_server, StdioServerConfig) + assert single_quote_server.effective_command == "python" + expected_args_single = ["-m", "my_server", "--config", "another path with spaces/config.json"] + assert single_quote_server.effective_args == expected_args_single + + # Test mixed quotes + mixed_quote_server = config.servers["server_with_mixed_quotes"] + assert isinstance(mixed_quote_server, StdioServerConfig) + assert mixed_quote_server.effective_command == "python" + expected_args_mixed = ["-m", "my_server", "--name", "My Server", "--path", "/home/user/my path"] + assert mixed_quote_server.effective_args == expected_args_mixed + + +def test_both_field_names_supported(): + """Test that both 'servers' and 'mcpServers' field names are supported.""" + # Test with 'mcpServers' field name (traditional format) + config_with_mcp_servers = MCPServersConfig.model_validate( + {"mcpServers": {"test_server": {"command": "python -m test_server", "type": "stdio"}}} + ) + + # Test with 'servers' field name (new format) + config_with_servers = MCPServersConfig.model_validate( + {"servers": {"test_server": {"command": "python -m test_server", "type": "stdio"}}} + ) + + # Both should produce identical results + assert config_with_mcp_servers.servers == config_with_servers.servers + assert "test_server" in config_with_mcp_servers.servers + assert "test_server" in config_with_servers.servers + + # Verify the server configurations are correct + server1 = config_with_mcp_servers.servers["test_server"] + server2 = config_with_servers.servers["test_server"] + + assert isinstance(server1, StdioServerConfig) + assert isinstance(server2, StdioServerConfig) + assert server1.command == server2.command == "python -m test_server" + + +def test_servers_field_takes_precedence(): + """Test that 'servers' field takes precedence when both are present.""" + config_data = { + "mcpServers": {"old_server": {"command": "python -m old_server", "type": "stdio"}}, + "servers": {"new_server": {"command": "python -m new_server", "type": "stdio"}}, + } + + config = MCPServersConfig.model_validate(config_data) + + # Should only have the 'servers' content, not 'mcpServers' + assert config.has_server("new_server") + assert not config.has_server("old_server") + assert len(config.list_servers()) == 1 + + +def test_from_file_with_inputs(tmp_path: Path): + """Test loading config from file with input substitution.""" + # Create test config file + config_content = { + "inputs": [ + {"id": "host", "description": "Server hostname"}, + {"id": "token", "description": "API token"}, + ], + "servers": { + "dynamic_server": { + "type": "streamable_http", + "url": "https://${input:host}/mcp/api", + "headers": {"Authorization": "Bearer ${input:token}"}, + } + }, + } + + config_file = tmp_path / "test_config.json" + with open(config_file, "w") as f: + json.dump(config_content, f) + + config = MCPServersConfig.from_file(config_file) + + assert config.get_required_inputs() == ["host", "token"] + assert config.inputs is not None + assert config.inputs[0].id == "host" + assert config.inputs[1].id == "token" + assert config.inputs[0].description == "Server hostname" + assert config.inputs[1].description == "API token" + + input_values = {"host": "api.example.com", "token": "test-token-123"} + server = config.server("dynamic_server", input_values=input_values) + + assert isinstance(server, StreamableHTTPServerConfig) + assert server.url == "https://api.example.com/mcp/api" + assert server.headers == {"Authorization": "Bearer test-token-123"} + + +def test_from_file_without_inputs(tmp_path: Path): + """Test loading config from file without input substitution.""" + # Create test config file with placeholders + config_content = { + "servers": { + "static_server": {"type": "sse", "url": "https://static.example.com/mcp/sse"}, + "placeholder_server": {"type": "sse", "url": "https://${input:host}/mcp/sse"}, + } + } + + config_file = tmp_path / "test_config.json" + with open(config_file, "w") as f: + json.dump(config_content, f) + + # Load without input substitution - placeholders should remain + config = MCPServersConfig.from_file(config_file) + + static_server = config.servers["static_server"] + assert isinstance(static_server, SSEServerConfig) + assert static_server.url == "https://static.example.com/mcp/sse" + + placeholder_server = config.servers["placeholder_server"] + assert isinstance(placeholder_server, SSEServerConfig) + assert placeholder_server.url == "https://${input:host}/mcp/sse" # Unchanged + + +def test_input_substitution_yaml_file(tmp_path: Path): + """Test input substitution with YAML files.""" + yaml_content = """ +inputs: + - type: promptString + id: module + description: Python module to run + - type: promptString + id: port + description: Port to run the server on + - type: promptString + id: debug + description: Debug mode +servers: + yaml_server: + type: stdio + command: python -m ${input:module} + args: + - --port + - "${input:port}" + env: + DEBUG: "${input:debug}" +""" + + config_file = tmp_path / "test_config.yaml" + assert config_file.write_text(yaml_content) + + config = MCPServersConfig.from_file(config_file) + + assert config.get_required_inputs() == ["module", "port", "debug"] + assert config.inputs is not None + assert len(config.inputs) == 3 + assert config.inputs[0].id == "module" + assert config.inputs[0].description == "Python module to run" + assert config.inputs[1].id == "port" + assert config.inputs[1].description == "Port to run the server on" + assert config.inputs[2].id == "debug" + assert config.inputs[2].description == "Debug mode" + + input_values = {"module": "test_server", "port": "8080", "debug": "true"} + server = config.server("yaml_server", input_values=input_values) + + assert isinstance(server, StdioServerConfig) + assert server.command == "python -m test_server" + assert server.args == ["--port", "8080"] + assert server.env == {"DEBUG": "true"} + + +def test_input_definitions_parsing(): + """Test parsing of input definitions from config.""" + config_data = { + "inputs": [ + {"type": "promptString", "id": "functionapp-name", "description": "Azure Functions App Name"}, + { + "type": "promptString", + "id": "api-token", + "description": "API Token for authentication", + "password": True, + }, + ], + "servers": { + "azure_server": { + "type": "sse", + "url": "https://${input:functionapp-name}.azurewebsites.net/mcp/sse", + "headers": {"Authorization": "Bearer ${input:api-token}"}, + } + }, + } + + config = MCPServersConfig.model_validate(config_data) + + # Test input definitions are parsed correctly + assert config.get_required_inputs() == ["functionapp-name", "api-token"] + assert config.inputs is not None + assert len(config.inputs) == 2 + app_name_input = config.inputs[0] + assert app_name_input.id == "functionapp-name" + assert app_name_input.description == "Azure Functions App Name" + assert app_name_input.password is False + assert app_name_input.type == "promptString" + api_token_input = config.inputs[1] + assert api_token_input.id == "api-token" + assert api_token_input.description == "API Token for authentication" + assert api_token_input.password is True + assert api_token_input.type == "promptString" + + +def test_get_required_inputs(): + """Test getting list of required input IDs.""" + config_data = { + "inputs": [ + {"id": "input1", "description": "First input"}, + {"id": "input2", "description": "Second input"}, + {"id": "input3", "description": "Third input"}, + ], + "servers": {"test_server": {"type": "stdio", "command": "python test.py"}}, + } + + config = MCPServersConfig.model_validate(config_data) + + assert config.get_required_inputs() == ["input1", "input2", "input3"] + + +def test_get_required_inputs_no_inputs_defined(): + """Test getting required inputs when no inputs are defined.""" + config_data = {"servers": {"test_server": {"type": "stdio", "command": "python test.py"}}} + + config = MCPServersConfig.model_validate(config_data) + + assert config.get_required_inputs() == [] + + +def test_get_required_inputs_empty_inputs_list(): + """Test getting required inputs when inputs is explicitly set to an empty list.""" + config_data = { + "inputs": [], # Explicitly empty list + "servers": {"test_server": {"type": "stdio", "command": "python test.py"}}, + } + + config = MCPServersConfig.model_validate(config_data) + + assert config.validate_inputs({}) == [] + assert config.get_required_inputs() == [] + assert config.inputs == [] # Verify inputs is actually an empty list, not None + + +def test_validate_inputs_all_provided(): + """Test input validation when all required inputs are provided.""" + config_data = { + "inputs": [ + {"id": "username", "description": "Username"}, + {"id": "password", "description": "Password", "password": True}, + ], + "servers": {"test_server": {"type": "stdio", "command": "python test.py"}}, + } + + config = MCPServersConfig.model_validate(config_data) + provided_inputs = {"username": "testuser", "password": "secret123"} + + missing_inputs = config.validate_inputs(provided_inputs) + assert missing_inputs == [] + + +def test_validate_inputs_some_missing(): + """Test input validation when some required inputs are missing.""" + config_data = { + "inputs": [ + {"id": "required1", "description": "First required input"}, + {"id": "required2", "description": "Second required input"}, + {"id": "required3", "description": "Third required input"}, + ], + "servers": {"test_server": {"type": "stdio", "command": "python test.py"}}, + } + + config = MCPServersConfig.model_validate(config_data) + provided_inputs = { + "required1": "value1", + # required2 and required3 are missing + } + + missing_inputs = config.validate_inputs(provided_inputs) + assert set(missing_inputs) == {"required2", "required3"} + + +def test_get_input_description(): + """Test getting input descriptions.""" + config_data = { + "inputs": [ + {"id": "api-key", "description": "API Key for authentication"}, + {"id": "host", "description": "Server hostname"}, + ], + "servers": {"test_server": {"type": "stdio", "command": "python test.py"}}, + } + + config = MCPServersConfig.model_validate(config_data) + + assert config.get_input_description("api-key") == "API Key for authentication" + assert config.get_input_description("host") == "Server hostname" + assert config.get_input_description("nonexistent") is None + + +def test_get_input_description_no_inputs(): + """Test getting input description when no inputs are defined.""" + config_data = {"servers": {"test_server": {"type": "stdio", "command": "python test.py"}}} + + config = MCPServersConfig.model_validate(config_data) + assert config.get_input_description("any-key") is None + + +def test_from_file_with_input_validation_success(tmp_path: Path): + """Test loading file with input definitions and successful validation.""" + config_content = { + "inputs": [ + {"id": "app-name", "description": "Application name"}, + {"id": "env", "description": "Environment (dev/prod)"}, + ], + "servers": { + "app_server": { + "type": "streamable_http", + "url": "https://${input:app-name}-${input:env}.example.com/mcp/api", + } + }, + } + + config_file = tmp_path / "test_config.json" + with open(config_file, "w") as f: + json.dump(config_content, f) + + config = MCPServersConfig.from_file(config_file) + + assert config.get_required_inputs() == ["app-name", "env"] + assert config.inputs is not None + assert len(config.inputs) == 2 + assert config.inputs[0].id == "app-name" + assert config.inputs[0].description == "Application name" + assert config.inputs[1].id == "env" + assert config.inputs[1].description == "Environment (dev/prod)" + + input_values = {"app-name": "myapp", "env": "prod"} + server = config.server("app_server", input_values=input_values) + + assert isinstance(server, StreamableHTTPServerConfig) + assert server.url == "https://myapp-prod.example.com/mcp/api" + + +def test_from_file_with_input_validation_failure(tmp_path: Path): + """Test loading file with input definitions and validation failure.""" + config_content = { + "inputs": [ + {"id": "required-key", "description": "A required API key"}, + {"id": "optional-host", "description": "Optional hostname"}, + ], + "servers": {"test_server": {"type": "sse", "url": "https://${input:optional-host}/api"}}, + } + + config_file = tmp_path / "test_config.json" + with open(config_file, "w") as f: + json.dump(config_content, f) + + inputs: dict[str, str] = { + # Missing 'required-key' and 'optional-host' + } + + # Should raise ValueError with helpful error message + with pytest.raises(ValueError, match="Missing required input values"): + config = MCPServersConfig.from_file(config_file) + server = config.server("test_server", input_values=inputs) + assert server + + +def test_from_file_without_input_definitions_no_validation(tmp_path: Path): + """Test that configs without input definitions don't perform validation.""" + config_content = { + "servers": {"test_server": {"type": "stdio", "command": "python -m server --token ${input:token}"}} + } + + config_file = tmp_path / "test_config.json" + with open(config_file, "w") as f: + json.dump(config_content, f) + + config = MCPServersConfig.from_file(config_file) + + # Even with empty inputs, should load fine since no input definitions exist + server = config.server("test_server", input_values={}) + + assert isinstance(server, StdioServerConfig) + # Placeholder should remain unchanged + assert server.command == "python -m server --token ${input:token}" + + +def test_input_definition_with_yaml_file(tmp_path: Path): + """Test input definitions work with YAML files.""" + yaml_content = """ +inputs: + - type: promptString + id: module-name + description: Python module to run + - type: promptString + id: config-path + description: Path to configuration file + +servers: + yaml_server: + type: stdio + command: python -m ${input:module-name} + args: + - --config + - ${input:config-path} +""" + + config_file = tmp_path / "test_config.yaml" + assert config_file.write_text(yaml_content) + + config = MCPServersConfig.from_file(config_file) + + # Verify input definitions were parsed + assert config.get_required_inputs() == ["module-name", "config-path"] + assert config.inputs is not None + assert len(config.inputs) == 2 + assert config.inputs[0].id == "module-name" + assert config.inputs[1].id == "config-path" + + input_values = {"module-name": "test_module", "config-path": "/etc/config.json"} + server = config.server("yaml_server", input_values=input_values) + + assert isinstance(server, StdioServerConfig) + assert server.command == "python -m test_module" + assert server.args == ["--config", "/etc/config.json"] + + +def test_jsonc_comment_stripping(): + """Test stripping of // comments from JSONC content.""" + # Test basic comment stripping + content_with_comments = """ +{ + // This is a comment + "servers": { + "test_server": { + "type": "stdio", + "command": "python test.py" // End of line comment + } + }, + // Another comment + "inputs": [] // Final comment +} +""" + + stripped = MCPServersConfig._strip_json_comments(content_with_comments) + config = MCPServersConfig.model_validate(json.loads(stripped)) + + assert config.has_server("test_server") + server = config.server("test_server") + assert isinstance(server, StdioServerConfig) + assert server.command == "python test.py" + + +def test_jsonc_comments_inside_strings_preserved(): + """Test that // inside strings are not treated as comments.""" + content_with_urls = """ +{ + "servers": { + "web_server": { + "type": "sse", + "url": "https://example.com/api/endpoint" // This is a comment + }, + "protocol_server": { + "type": "stdio", + "command": "node server.js --url=http://localhost:3000" + } + } +} +""" + + stripped = MCPServersConfig._strip_json_comments(content_with_urls) + config = MCPServersConfig.model_validate(json.loads(stripped)) + + web_server = config.servers["web_server"] + assert isinstance(web_server, SSEServerConfig) + assert web_server.url == "https://example.com/api/endpoint" + + protocol_server = config.servers["protocol_server"] + assert isinstance(protocol_server, StdioServerConfig) + # The // in the URL should be preserved + assert "http://localhost:3000" in protocol_server.command + + +def test_jsonc_escaped_quotes_handling(): + """Test that escaped quotes in strings are handled correctly.""" + content_with_escaped = """ +{ + "servers": { + "test_server": { + "type": "stdio", + "command": "python -c \\"print('Hello // World')\\"", // Comment after escaped quotes + "description": "Server with \\"escaped quotes\\" and // in string" + } + } +} +""" + + stripped = MCPServersConfig._strip_json_comments(content_with_escaped) + config = MCPServersConfig.model_validate(json.loads(stripped)) + + server = config.servers["test_server"] + assert isinstance(server, StdioServerConfig) + # The command should preserve the escaped quotes and // inside the string + assert server.command == "python -c \"print('Hello // World')\"" + + +def test_from_file_with_jsonc_comments(tmp_path: Path): + """Test loading JSONC file with comments via from_file method.""" + jsonc_content = """ +{ + // Configuration for MCP servers + "inputs": [ + { + "type": "promptString", + "id": "api-key", // Secret API key + "description": "API Key for authentication" + } + ], + "servers": { + // Main server configuration + "main_server": { + "type": "sse", + "url": "https://api.example.com/mcp/sse", // Production URL + "headers": { + "Authorization": "Bearer ${input:api-key}" // Dynamic token + } + } + } + // End of configuration +} +""" + + config_file = tmp_path / "test_config.json" + assert config_file.write_text(jsonc_content) + + # Should load successfully despite comments + config = MCPServersConfig.from_file(config_file) + + # Verify input definitions were parsed + assert config.inputs is not None + assert len(config.inputs) == 1 + assert config.inputs[0].id == "api-key" + + # Verify server configuration and input substitution + server = config.server("main_server", input_values={"api-key": "secret123"}) + assert isinstance(server, SSEServerConfig) + assert server.url == "https://api.example.com/mcp/sse" + assert server.headers == {"Authorization": "Bearer secret123"} + + +def test_jsonc_multiline_strings_with_comments(): + """Test that comments in multiline scenarios are handled correctly.""" + content = """ +{ + "servers": { + "test1": { + // Comment before + "type": "stdio", // Comment after + "command": "python server.py" + }, // Comment after object + "test2": { "type": "sse", "url": "https://example.com" } // Inline comment + } +} +""" + + stripped = MCPServersConfig._strip_json_comments(content) + config = MCPServersConfig.model_validate(json.loads(stripped)) + + assert len(config.servers) == 2 + assert config.has_server("test1") + assert config.has_server("test2") + + test1 = config.server("test1") + assert isinstance(test1, StdioServerConfig) + assert test1.command == "python server.py" + + test2 = config.servers["test2"] + assert isinstance(test2, SSEServerConfig) + assert test2.url == "https://example.com" + + +def test_sse_type_inference(): + """Test that servers with 'url' field (and SSE mention) are inferred as sse type.""" + config_data = { + "servers": { + "api_server": { + "url": "https://api.example.com/sse" + # No explicit type - should be inferred as sse + # because "sse" is in the url + }, + "webhook_server": { + "url": "https://webhook.example.com/mcp/api", + "description": "A simple SSE server", + "headers": {"X-API-Key": "secret123"}, + # No explicit type - should be inferred as sse + # because "SSE" is in the description + }, + } + } + + config = MCPServersConfig.model_validate(config_data) + + # Verify first server + api_server = config.servers["api_server"] + assert isinstance(api_server, SSEServerConfig) + assert api_server.type == "sse" # Should be auto-inferred + assert api_server.url == "https://api.example.com/sse" + assert api_server.headers is None + + # Verify second server + webhook_server = config.servers["webhook_server"] + assert isinstance(webhook_server, SSEServerConfig) + assert webhook_server.type == "sse" # Should be auto-inferred + assert webhook_server.url == "https://webhook.example.com/mcp/api" + assert webhook_server.headers == {"X-API-Key": "secret123"} + + +def test_streamable_http_type_inference(): + """Test that servers with 'url' field (but no SSE mention) are inferred as streamable_http type.""" + config_data = { + "servers": { + "api_server": { + "url": "https://api.example.com/mcp" + # No explicit type - should be inferred as streamable_http + # No mention of 'sse' in url, name, or description + }, + "webhook_server": { + "url": "https://webhook.example.com/mcp/api", + "headers": {"X-API-Key": "secret123"}, + # No explicit type - should be inferred as streamable_http + }, + } + } + + config = MCPServersConfig.model_validate(config_data) + + # Verify first server + api_server = config.servers["api_server"] + assert isinstance(api_server, StreamableHTTPServerConfig) + assert api_server.type == "streamable_http" # Should be auto-inferred + assert api_server.url == "https://api.example.com/mcp" + assert api_server.headers is None + + # Verify second server + webhook_server = config.servers["webhook_server"] + assert isinstance(webhook_server, StreamableHTTPServerConfig) + assert webhook_server.type == "streamable_http" # Should be auto-inferred + assert webhook_server.url == "https://webhook.example.com/mcp/api" + assert webhook_server.headers == {"X-API-Key": "secret123"} diff --git a/tests/client/config/test_warning_functionality.py b/tests/client/config/test_warning_functionality.py new file mode 100644 index 000000000..de9eec67d --- /dev/null +++ b/tests/client/config/test_warning_functionality.py @@ -0,0 +1,219 @@ +"""Tests for warning functionality when accessing servers field directly.""" + +import warnings + +from mcp.client.config.mcp_servers_config import MCPServersConfig + + +def test_test_functions_no_warning(): + """Test that test functions (like this one) do not emit warnings.""" + config_data = {"servers": {"test-server": {"type": "stdio", "command": "python -m test_server"}}} + + config = MCPServersConfig.model_validate(config_data) + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + + # Access servers directly - should trigger warning + servers = config.servers + + assert len(w) == 1 + + # Verify we still get the servers + assert len(servers) == 1 + assert "test-server" in servers + + +def test_server_method_no_warning(): + """Test that using server() method does not emit warnings.""" + config_data = {"servers": {"test-server": {"type": "stdio", "command": "python -m test_server"}}} + + config = MCPServersConfig.model_validate(config_data) + + # Capture warnings + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + + # Use server() method - this should NOT trigger warning + server = config.server("test-server") + + # Check no warning was emitted + assert len(w) == 0 + + # Verify we get the server + assert server.type == "stdio" + assert server.command == "python -m test_server" + + +def test_list_servers_no_warning(): + """Test that using list_servers() method does not emit warnings.""" + config_data = {"servers": {"test-server": {"type": "stdio", "command": "python -m test_server"}}} + + config = MCPServersConfig.model_validate(config_data) + + # Capture warnings + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + + # Use list_servers() method - this should NOT trigger warning + server_names = config.list_servers() + + # Check no warning was emitted + assert len(w) == 0 + + # Verify we get the server names + assert server_names == ["test-server"] + + +def test_has_server_no_warning(): + """Test that using has_server() method does not emit warnings.""" + config_data = {"servers": {"test-server": {"type": "stdio", "command": "python -m test_server"}}} + + config = MCPServersConfig.model_validate(config_data) + + # Capture warnings + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + + # Use has_server() method - this should NOT trigger warning + exists = config.has_server("test-server") + + # Check no warning was emitted + assert len(w) == 0 + + # Verify result + assert exists is True + + +def test_other_field_access_no_warning(): + """Test that accessing other fields does not emit warnings.""" + config_data = { + "servers": {"test-server": {"type": "stdio", "command": "python -m test_server"}}, + "inputs": [{"id": "test-input", "description": "Test input"}], + } + + config = MCPServersConfig.model_validate(config_data) + + # Capture warnings + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + + # Access other fields - should NOT trigger warning + inputs = config.inputs + + # Check no warning was emitted + assert len(w) == 0 + + # Verify we get the inputs + assert inputs is not None + assert len(inputs) == 1 + assert inputs[0].id == "test-input" + + +def test_warning_logic_conditions(): + """Test that the warning logic correctly identifies different conditions.""" + config_data = {"servers": {"test-server": {"type": "stdio", "command": "python -m test_server"}}} + + config = MCPServersConfig.model_validate(config_data) + + # Test that accessing servers from this test function doesn't warn + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + servers = config.servers + assert len(servers) == 1 + assert len(w) == 1 + + +def test_internal_methods_use_servers_field(): + """Test that internal methods can access servers without warnings.""" + config_data = { + "servers": { + "test1": {"type": "stdio", "command": "python -m test1"}, + "test2": {"type": "stdio", "command": "python -m test2"}, + } + } + + config = MCPServersConfig.model_validate(config_data) + + # Test that internal methods work without warnings + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + + # These methods internally access config.servers + server_list = config.list_servers() + has_test1 = config.has_server("test1") + server_obj = config.server("test1") + + # Should not generate warnings since these are internal method calls + assert len(w) == 0 + + # Verify results + assert "test1" in server_list + assert "test2" in server_list + assert has_test1 is True + assert server_obj.type == "stdio" + if server_obj.type == "stdio": + assert server_obj.command == "python -m test1" + + +def test_warning_system_attributes(): + """Test that the warning system correctly identifies caller attributes.""" + import inspect + + config_data = {"servers": {"test-server": {"type": "stdio", "command": "python -m test_server"}}} + + config = MCPServersConfig.model_validate(config_data) + + # Get current frame info to verify the test detection logic + current_frame = inspect.currentframe() + if current_frame: + filename = current_frame.f_code.co_filename + function_name = current_frame.f_code.co_name + + # Verify our test detection logic would work + assert "test_" in function_name # This function starts with test_ + assert "/tests/" in filename # This file is in tests directory + + # Access servers - should not warn due to test function detection + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + servers = config.servers + assert len(servers) == 1 + assert len(w) == 1 + + +def test_configuration_still_works(): + """Test that the warning system doesn't break normal configuration functionality.""" + config_data = { + "servers": { + "stdio-server": {"type": "stdio", "command": "python -m stdio_server", "args": ["--verbose"]}, + "http-server": {"type": "streamable_http", "url": "http://localhost:8000"}, + }, + "inputs": [{"id": "api-key", "description": "API key for authentication"}], + } + + config = MCPServersConfig.model_validate(config_data) + + # Test all functionality still works + assert config.list_servers() == ["stdio-server", "http-server"] + assert config.has_server("stdio-server") + assert not config.has_server("nonexistent") + + stdio_server = config.server("stdio-server") + assert stdio_server.type == "stdio" + assert stdio_server.command == "python -m stdio_server" + assert stdio_server.args == ["--verbose"] + + http_server = config.server("http-server") + assert http_server.type == "streamable_http" + assert http_server.url == "http://localhost:8000" + + # Test input validation + required_inputs = config.get_required_inputs() + assert required_inputs == ["api-key"] + + missing = config.validate_inputs({}) + assert missing == ["api-key"] + + no_missing = config.validate_inputs({"api-key": "secret"}) + assert no_missing == [] diff --git a/tests/client/config/test_yaml_functionality.py b/tests/client/config/test_yaml_functionality.py new file mode 100644 index 000000000..01012ecb4 --- /dev/null +++ b/tests/client/config/test_yaml_functionality.py @@ -0,0 +1,150 @@ +# stdlib imports +from pathlib import Path +from unittest.mock import patch + +# third party imports +import pytest + +# local imports +from mcp.client.config.mcp_servers_config import MCPServersConfig, StdioServerConfig, StreamableHTTPServerConfig + + +@pytest.fixture +def mcp_yaml_config_file() -> Path: + """Return path to the mcp.yaml config file.""" + return Path(__file__).parent / "mcp.yaml" + + +def test_yaml_extension_auto_detection(mcp_yaml_config_file: Path): + """Test that .yaml files are automatically parsed with PyYAML.""" + config = MCPServersConfig.from_file(mcp_yaml_config_file) + + # Should successfully load the YAML file with all 9 servers + assert config.has_server("stdio_server") + assert config.has_server("streamable_http_server_with_headers") + assert config.has_server("sse_server_with_explicit_type") + + # Verify a specific server + stdio_server = config.server("stdio_server") + assert isinstance(stdio_server, StdioServerConfig) + assert stdio_server.command == "python" + assert stdio_server.args == ["-m", "my_server"] + assert stdio_server.env == {"DEBUG": "true"} + + +def test_use_pyyaml_parameter_with_json_file(): + """Test that use_pyyaml=True forces PyYAML parsing even for JSON files.""" + json_file = Path(__file__).parent / "mcp.json" + + # Load with PyYAML explicitly + config = MCPServersConfig.from_file(json_file, use_pyyaml=True) + + # Should work fine - PyYAML can parse JSON + assert len(config.list_servers()) == 7 + assert config.has_server("stdio_server") + + # Verify it produces the same result as normal JSON parsing + config_json = MCPServersConfig.from_file(json_file, use_pyyaml=False) + assert len(config.list_servers()) == len(config_json.list_servers()) + assert list(config.list_servers()) == list(config_json.list_servers()) + + +def test_uvx_time_server(mcp_yaml_config_file: Path): + """Test the time server configuration with uvx command.""" + config = MCPServersConfig.from_file(mcp_yaml_config_file) + + # Should have the time server + assert config.has_server("time") + + # Verify the server configuration + time_server = config.server("time") + assert isinstance(time_server, StdioServerConfig) + assert time_server.type == "stdio" # Should be auto-inferred from command field + assert time_server.command == "uvx mcp-server-time" + assert time_server.args is None # No explicit args + assert time_server.env is None # No environment variables + + # Test the effective command parsing + assert time_server.effective_command == "uvx" + assert time_server.effective_args == ["mcp-server-time"] + + +def test_streamable_http_server(mcp_yaml_config_file: Path): + """Test the new streamable HTTP server configuration without headers.""" + config = MCPServersConfig.from_file(mcp_yaml_config_file) + + # Should have the new streamable_http_server + assert config.has_server("streamable_http_server") + + # Verify the server configuration + http_server = config.server("streamable_http_server") + assert isinstance(http_server, StreamableHTTPServerConfig) + assert http_server.type == "streamable_http" # Should be auto-inferred + assert http_server.url == "https://api.example.com/mcp" + assert http_server.headers is None # No headers specified + + +def test_npx_filesystem_server(mcp_yaml_config_file: Path): + """Test the filesystem server configuration with full command string and multiple arguments.""" + config = MCPServersConfig.from_file(mcp_yaml_config_file) + + # Should have the filesystem server + assert config.has_server("filesystem") + + # Verify the server configuration + filesystem_server = config.server("filesystem") + assert isinstance(filesystem_server, StdioServerConfig) + assert filesystem_server.type == "stdio" # Should be auto-inferred from command field + assert ( + filesystem_server.command + == "npx -y @modelcontextprotocol/server-filesystem /Users/username/Desktop /path/to/other/allowed/dir" + ) + assert filesystem_server.args is None # No explicit args + assert filesystem_server.env is None # No environment variables + + # Test the effective command and args parsing + assert filesystem_server.effective_command == "npx" + assert filesystem_server.effective_args == [ + "-y", + "@modelcontextprotocol/server-filesystem", + "/Users/username/Desktop", + "/path/to/other/allowed/dir", + ] + + +def test_yaml_not_importable_error(mcp_yaml_config_file: Path): + """Test that trying to parse a YAML file when yaml module is not available raises ImportError.""" + + # Mock the yaml module to be None (simulating import failure) + with patch("mcp.client.config.mcp_servers_config.yaml", None): + with pytest.raises(ImportError, match="PyYAML is required to parse YAML files"): + MCPServersConfig.from_file(mcp_yaml_config_file) + + +def test_yaml_not_importable_error_with_use_pyyaml_true(): + """Test that trying to use use_pyyaml=True when yaml module is not available raises ImportError.""" + + # Create a simple JSON file content but force YAML parsing + json_file = Path(__file__).parent / "mcp.json" + + # Mock the yaml module to be None (simulating import failure) + with patch("mcp.client.config.mcp_servers_config.yaml", None): + with pytest.raises(ImportError, match="PyYAML is required to parse YAML files"): + MCPServersConfig.from_file(json_file, use_pyyaml=True) + + +def test_yaml_not_importable_error_with_yml_extension(tmp_path: Path): + """Test that trying to parse a .yml file when yaml module is not available raises ImportError.""" + + # Create a temporary .yml file + yml_file = tmp_path / "test_config.yml" + yml_file.write_text(""" +mcpServers: + test_server: + command: python -m test_server +""") + + # Mock the yaml module to be None (simulating import failure) + with patch("mcp.client.config.mcp_servers_config.yaml", None): + with pytest.raises(ImportError, match="PyYAML is required to parse YAML files"): + MCPServersConfig.from_file(yml_file) diff --git a/uv.lock b/uv.lock index 180d5a9c1..525866c44 100644 --- a/uv.lock +++ b/uv.lock @@ -551,6 +551,9 @@ rich = [ ws = [ { name = "websockets" }, ] +yaml = [ + { name = "pyyaml" }, +] [package.dev-dependencies] dev = [ @@ -580,6 +583,7 @@ requires-dist = [ { name = "pydantic-settings", specifier = ">=2.5.2" }, { name = "python-dotenv", marker = "extra == 'cli'", specifier = ">=1.0.0" }, { name = "python-multipart", specifier = ">=0.0.9" }, + { name = "pyyaml", marker = "extra == 'yaml'", specifier = ">=6.0.2" }, { name = "rich", marker = "extra == 'rich'", specifier = ">=13.9.4" }, { name = "sse-starlette", specifier = ">=1.6.1" }, { name = "starlette", specifier = ">=0.27" }, @@ -587,7 +591,7 @@ requires-dist = [ { name = "uvicorn", marker = "sys_platform != 'emscripten'", specifier = ">=0.23.1" }, { name = "websockets", marker = "extra == 'ws'", specifier = ">=15.0.1" }, ] -provides-extras = ["cli", "rich", "ws"] +provides-extras = ["cli", "rich", "ws", "yaml"] [package.metadata.requires-dev] dev = [