Move input substitution to new server method

It seemed weird doing it in `from_file`, because the caller has to call
`from_file` to find out what the required inputs are and then has to
call `from_file` against to provide the input values.

Now, the caller calls config.get(server, input_values={...})

to provide the input values

Author: msabramo

src/mcp/client/config/mcp_servers_config.py

@@ -138,6 +138,31 @@ class MCPServersConfig(BaseModel):
     servers: dict[str, ServerConfigUnion]
     inputs: list[InputDefinition] | None = None
 
+    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
+
     @model_validator(mode="before")
     @classmethod
     def handle_field_aliases(cls, data: dict[str, Any]) -> dict[str, Any]:
@@ -275,7 +300,9 @@ def _strip_json_comments(cls, content: str) -> str:
 
     @classmethod
     def from_file(
-        cls, config_path: Path | str, use_pyyaml: bool = False, inputs: dict[str, str] | None = None
+        cls,
+        config_path: Path | str,
+        use_pyyaml: bool = False,
     ) -> "MCPServersConfig":
         """Load configuration from a JSON or YAML file.
 
@@ -305,22 +332,4 @@ def from_file(
                 cleaned_content = cls._strip_json_comments(content)
                 data = json.loads(cleaned_content)
 
-            # Create a preliminary config to validate inputs if they're defined
-            preliminary_config = cls.model_validate(data)
-
-            # Validate inputs if provided and input definitions exist
-            if inputs is not None and preliminary_config.inputs:
-                missing_inputs = preliminary_config.validate_inputs(inputs)
-                if missing_inputs:
-                    descriptions: list[str] = []
-                    for input_id in missing_inputs:
-                        desc = preliminary_config.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 inputs:
-                data = cls._substitute_inputs(data, inputs)
-
             return cls.model_validate(data)

tests/client/config/test_mcp_servers_config.py

@@ -188,108 +188,39 @@ def test_servers_field_takes_precedence():
     assert len(config.servers) == 1
 
 
-def test_input_substitution():
-    """Test that ${input:key} placeholders are substituted correctly."""
-    config_data = {
-        "servers": {
-            "azure_server": {
-                "type": "sse",
-                "url": "https://${input:app-name}.azurewebsites.net/mcp/sse",
-                "headers": {"Authorization": "Bearer ${input:api-token}", "X-Custom-Header": "${input:custom-value}"},
-            },
-            "stdio_server": {
-                "type": "stdio",
-                "command": "python -m ${input:module-name}",
-                "args": ["--config", "${input:config-file}"],
-                "env": {"API_KEY": "${input:api-key}", "ENV": "${input:environment}"},
-                "isActive": True,
-            },
-        }
-    }
-
-    inputs = {
-        "app-name": "my-function-app",
-        "api-token": "abc123token",
-        "custom-value": "custom-header-value",
-        "module-name": "my_server",
-        "config-file": "/path/to/config.json",
-        "api-key": "secret-api-key",
-        "environment": "production",
-    }
-
-    # Test substitution
-    substituted = MCPServersConfig._substitute_inputs(config_data, inputs)
-    config = MCPServersConfig.model_validate(substituted)
-
-    # Test SSE server substitution
-    sse_server = config.servers["azure_server"]
-    assert isinstance(sse_server, SSEServerConfig)
-    assert sse_server.url == "https://my-function-app.azurewebsites.net/mcp/sse"
-    assert sse_server.headers == {"Authorization": "Bearer abc123token", "X-Custom-Header": "custom-header-value"}
-
-    # Test stdio server substitution
-    stdio_server = config.servers["stdio_server"]
-    assert isinstance(stdio_server, StdioServerConfig)
-    assert stdio_server.command == "python -m my_server"
-    assert stdio_server.args == ["--config", "/path/to/config.json"]
-    assert stdio_server.env == {"API_KEY": "secret-api-key", "ENV": "production"}
-
-
-def test_input_substitution_missing_key():
-    """Test that missing input keys raise appropriate errors."""
-    config_data = {"servers": {"test_server": {"type": "sse", "url": "https://${input:missing-key}.example.com"}}}
-
-    inputs = {"other-key": "value"}
-
-    with pytest.raises(ValueError, match="Missing input value for key: 'missing-key'"):
-        MCPServersConfig._substitute_inputs(config_data, inputs)
-
-
-def test_input_substitution_partial():
-    """Test that only specified placeholders are substituted."""
-    config_data = {
-        "servers": {
-            "test_server": {
-                "type": "sse",
-                "url": "https://${input:app-name}.example.com/api/${input:version}",
-                "headers": {"Static-Header": "static-value", "Dynamic-Header": "${input:token}"},
-            }
-        }
-    }
-
-    inputs = {
-        "app-name": "myapp",
-        "token": "secret123",
-        # Note: 'version' is intentionally missing
-    }
-
-    with pytest.raises(ValueError, match="Missing input value for key: 'version'"):
-        MCPServersConfig._substitute_inputs(config_data, inputs)
-
-
 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)
 
-    inputs = {"host": "api.example.com", "token": "test-token-123"}
+    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"
 
-    # Load with input substitution
-    config = MCPServersConfig.from_file(config_file, inputs=inputs)
+    input_values = {"host": "api.example.com", "token": "test-token-123"}
+    server = config.server("dynamic_server", input_values=input_values)
 
-    server = config.servers["dynamic_server"]
     assert isinstance(server, StreamableHTTPServerConfig)
     assert server.url == "https://api.example.com/mcp/api"
     assert server.headers == {"Authorization": "Bearer test-token-123"}
@@ -324,6 +255,16 @@ def test_from_file_without_inputs(tmp_path: Path):
 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
@@ -336,13 +277,23 @@ def test_input_substitution_yaml_file(tmp_path: Path):
 """
 
     config_file = tmp_path / "test_config.yaml"
-    config_file.write_text(yaml_content)
+    assert config_file.write_text(yaml_content)
+
+    config = MCPServersConfig.from_file(config_file)
 
-    inputs = {"module": "test_server", "port": "8080", "debug": "true"}
+    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"
 
-    config = MCPServersConfig.from_file(config_file, inputs=inputs)
+    input_values = {"module": "test_server", "port": "8080", "debug": "true"}
+    server = config.server("yaml_server", input_values=input_values)
 
-    server = config.servers["yaml_server"]
     assert isinstance(server, StdioServerConfig)
     assert server.command == "python -m test_server"
     assert server.args == ["--port", "8080"]
@@ -373,15 +324,14 @@ def test_input_definitions_parsing():
     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"
@@ -401,19 +351,17 @@ def test_get_required_inputs():
     }
 
     config = MCPServersConfig.model_validate(config_data)
-    required_inputs = config.get_required_inputs()
 
-    assert required_inputs == ["input1", "input2", "input3"]
+    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)
-    required_inputs = config.get_required_inputs()
 
-    assert required_inputs == []
+    assert config.get_required_inputs() == []
 
 
 def test_get_required_inputs_empty_inputs_list():
@@ -424,10 +372,9 @@ def test_get_required_inputs_empty_inputs_list():
     }
 
     config = MCPServersConfig.model_validate(config_data)
-    required_inputs = config.get_required_inputs()
-    assert config.validate_inputs({}) == []
 
-    assert required_inputs == []
+    assert config.validate_inputs({}) == []
+    assert config.get_required_inputs() == []
     assert config.inputs == []  # Verify inputs is actually an empty list, not None
 
 
@@ -513,12 +460,19 @@ def test_from_file_with_input_validation_success(tmp_path: Path):
     with open(config_file, "w") as f:
         json.dump(config_content, f)
 
-    inputs = {"app-name": "myapp", "env": "prod"}
+    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)"
 
-    # Should load successfully with all required inputs provided
-    config = MCPServersConfig.from_file(config_file, inputs=inputs)
+    input_values = {"app-name": "myapp", "env": "prod"}
+    server = config.server("app_server", input_values=input_values)
 
-    server = config.servers["app_server"]
     assert isinstance(server, StreamableHTTPServerConfig)
     assert server.url == "https://myapp-prod.example.com/mcp/api"
 
@@ -537,13 +491,15 @@ def test_from_file_with_input_validation_failure(tmp_path: Path):
     with open(config_file, "w") as f:
         json.dump(config_content, f)
 
-    inputs = {
+    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"):
-        MCPServersConfig.from_file(config_file, inputs=inputs)
+        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):
@@ -556,10 +512,11 @@ def test_from_file_without_input_definitions_no_validation(tmp_path: Path):
     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
-    config = MCPServersConfig.from_file(config_file, inputs={})
+    server = config.server("test_server", input_values={})
 
-    server = config.servers["test_server"]
     assert isinstance(server, StdioServerConfig)
     # Placeholder should remain unchanged
     assert server.command == "python -m server --token ${input:token}"
@@ -586,20 +543,20 @@ def test_input_definition_with_yaml_file(tmp_path: Path):
 """
 
     config_file = tmp_path / "test_config.yaml"
-    config_file.write_text(yaml_content)
-
-    inputs = {"module-name": "test_module", "config-path": "/etc/config.json"}
+    assert config_file.write_text(yaml_content)
 
-    config = MCPServersConfig.from_file(config_file, inputs=inputs)
+    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"
 
-    # Verify substitution worked
-    server = config.servers["yaml_server"]
+    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"]
@@ -711,20 +668,18 @@ def test_from_file_with_jsonc_comments(tmp_path: Path):
 """
 
     config_file = tmp_path / "test_config.json"
-    config_file.write_text(jsonc_content)
-
-    inputs = {"api-key": "secret123"}
+    assert config_file.write_text(jsonc_content)
 
     # Should load successfully despite comments
-    config = MCPServersConfig.from_file(config_file, inputs=inputs)
+    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.servers["main_server"]
+    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"}