diff --git a/src/apm_cli/adapters/client/base.py b/src/apm_cli/adapters/client/base.py index 35e51e00..3741bafb 100644 --- a/src/apm_cli/adapters/client/base.py +++ b/src/apm_cli/adapters/client/base.py @@ -1,7 +1,10 @@ """Base adapter interface for MCP clients.""" +import re from abc import ABC, abstractmethod +_INPUT_VAR_RE = re.compile(r"\$\{input:([^}]+)\}") + class MCPClientAdapter(ABC): """Base adapter for MCP clients.""" @@ -84,3 +87,33 @@ def _infer_registry_name(package): return "nuget" return "" + + @staticmethod + def _warn_input_variables(mapping, server_name, runtime_label): + """Emit a warning for each ``${input:...}`` reference found in *mapping*. + + Runtimes that do not support VS Code-style input prompts (Copilot CLI, + Codex CLI, etc.) should call this so users know their placeholders + will not be resolved at runtime. + + Args: + mapping (dict): Header or env dict to scan. + server_name (str): Server name for the warning message. + runtime_label (str): Human-readable runtime name (e.g. "Copilot CLI"). + """ + if not mapping: + return + seen: set = set() + for value in mapping.values(): + if not isinstance(value, str): + continue + for match in _INPUT_VAR_RE.finditer(value): + var_id = match.group(1) + if var_id in seen: + continue + seen.add(var_id) + print( + f"[!] Warning: ${{input:{var_id}}} in server " + f"'{server_name}' will not be resolved \u2014 " + f"{runtime_label} does not support input variable prompts" + ) diff --git a/src/apm_cli/adapters/client/codex.py b/src/apm_cli/adapters/client/codex.py index fe7da478..eda11493 100644 --- a/src/apm_cli/adapters/client/codex.py +++ b/src/apm_cli/adapters/client/codex.py @@ -181,6 +181,7 @@ def _format_server_config(self, server_info, env_overrides=None, runtime_vars=No config["args"] = raw["args"] if raw.get("env"): config["env"] = raw["env"] + self._warn_input_variables(raw["env"], server_info.get("name", ""), "Codex CLI") return config # Note: Remote servers (SSE type) are handled in configure_mcp_server and rejected early diff --git a/src/apm_cli/adapters/client/copilot.py b/src/apm_cli/adapters/client/copilot.py index 77dfcae9..64147f39 100644 --- a/src/apm_cli/adapters/client/copilot.py +++ b/src/apm_cli/adapters/client/copilot.py @@ -173,6 +173,7 @@ def _format_server_config(self, server_info, env_overrides=None, runtime_vars=No config["args"] = raw["args"] if raw.get("env"): config["env"] = raw["env"] + self._warn_input_variables(raw["env"], server_info.get("name", ""), "Copilot CLI") # Apply tools override if present tools_override = server_info.get("_apm_tools_override") if tools_override: @@ -218,6 +219,10 @@ def _format_server_config(self, server_info, env_overrides=None, runtime_vars=No resolved_value = self._resolve_env_variable(header_name, header_value, env_overrides) config["headers"][header_name] = resolved_value + # Warn about unresolvable ${input:...} references in headers + if config.get("headers"): + self._warn_input_variables(config["headers"], server_info.get("name", ""), "Copilot CLI") + # Apply tools override from MCP dependency overlay if present tools_override = server_info.get("_apm_tools_override") if tools_override: diff --git a/src/apm_cli/adapters/client/vscode.py b/src/apm_cli/adapters/client/vscode.py index 91372cf1..6932f86d 100644 --- a/src/apm_cli/adapters/client/vscode.py +++ b/src/apm_cli/adapters/client/vscode.py @@ -8,7 +8,7 @@ import json import os from pathlib import Path -from .base import MCPClientAdapter +from .base import MCPClientAdapter, _INPUT_VAR_RE from ...registry.client import SimpleRegistryClient from ...registry.integration import RegistryIntegration @@ -197,6 +197,9 @@ def _format_server_config(self, server_info): } if raw.get("env"): server_config["env"] = raw["env"] + input_vars.extend( + self._extract_input_variables(raw["env"], server_info.get("name", "")) + ) return server_config, input_vars # Check for packages information @@ -308,6 +311,9 @@ def _format_server_config(self, server_info): "url": remote.get("url", ""), "headers": headers, } + input_vars.extend( + self._extract_input_variables(headers, server_info.get("name", "")) + ) # If no packages AND no endpoints/remotes, fail with clear error else: packages = server_info.get("packages", []) @@ -323,6 +329,36 @@ def _format_server_config(self, server_info): return server_config, input_vars + def _extract_input_variables(self, mapping, server_name): + """Scan dict values for ${input:...} references and return input variable definitions. + + Args: + mapping (dict): Header or env dict whose values may contain + ``${input:}`` placeholders. + server_name (str): Server name used in the description field. + + Returns: + list[dict]: Input variable definitions (``promptString``, ``password: true``). + Duplicates within *mapping* are already deduplicated. + """ + seen: set = set() + result: list = [] + for value in (mapping or {}).values(): + if not isinstance(value, str): + continue + for match in _INPUT_VAR_RE.finditer(value): + var_id = match.group(1) + if var_id in seen: + continue + seen.add(var_id) + result.append({ + "type": "promptString", + "id": var_id, + "description": f"{var_id} for MCP server {server_name}", + "password": True, + }) + return result + @staticmethod def _extract_package_args(package): """Extract positional arguments from a package entry. diff --git a/tests/unit/test_vscode_adapter.py b/tests/unit/test_vscode_adapter.py index 5cb277fe..3acbde02 100644 --- a/tests/unit/test_vscode_adapter.py +++ b/tests/unit/test_vscode_adapter.py @@ -8,6 +8,7 @@ import pytest from unittest.mock import patch, MagicMock from apm_cli.adapters.client.vscode import VSCodeClientAdapter +from apm_cli.adapters.client.base import MCPClientAdapter class TestVSCodeClientAdapter(unittest.TestCase): @@ -724,5 +725,209 @@ def test_pypi_inferred_from_name_pattern(self, mock_get_path): self.assertEqual(server["args"], ["my-mcp-server"]) +class TestExtractInputVariables(unittest.TestCase): + """Tests for ${input:...} variable extraction in self-defined MCP servers.""" + + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.vscode_dir = os.path.join(self.temp_dir.name, ".vscode") + os.makedirs(self.vscode_dir, exist_ok=True) + self.temp_path = os.path.join(self.vscode_dir, "mcp.json") + with open(self.temp_path, "w") as f: + json.dump({"servers": {}, "inputs": []}, f) + + self.mock_registry_patcher = patch('apm_cli.adapters.client.vscode.SimpleRegistryClient') + self.mock_registry_class = self.mock_registry_patcher.start() + self.mock_registry = MagicMock() + self.mock_registry_class.return_value = self.mock_registry + + self.mock_integration_patcher = patch('apm_cli.adapters.client.vscode.RegistryIntegration') + self.mock_integration_class = self.mock_integration_patcher.start() + self.mock_integration = MagicMock() + self.mock_integration_class.return_value = self.mock_integration + + def tearDown(self): + self.mock_registry_patcher.stop() + self.mock_integration_patcher.stop() + self.temp_dir.cleanup() + + def test_extract_single_input_variable(self): + adapter = VSCodeClientAdapter() + result = adapter._extract_input_variables( + {"Authorization": "Bearer ${input:my-token}"}, "my-server" + ) + assert len(result) == 1 + assert result[0]["id"] == "my-token" + assert result[0]["type"] == "promptString" + assert result[0]["password"] is True + assert "my-server" in result[0]["description"] + + def test_extract_multiple_input_variables(self): + adapter = VSCodeClientAdapter() + result = adapter._extract_input_variables( + { + "Authorization": "Bearer ${input:my-token}", + "X-Project": "${input:my-project}", + }, + "my-server", + ) + ids = {v["id"] for v in result} + assert ids == {"my-token", "my-project"} + + def test_dedup_same_variable(self): + adapter = VSCodeClientAdapter() + result = adapter._extract_input_variables( + { + "Authorization": "Bearer ${input:shared-token}", + "X-Alt-Auth": "Token ${input:shared-token}", + }, + "my-server", + ) + assert len(result) == 1 + assert result[0]["id"] == "shared-token" + + def test_no_input_variables(self): + adapter = VSCodeClientAdapter() + result = adapter._extract_input_variables( + {"Content-Type": "application/json"}, "my-server" + ) + assert result == [] + + def test_empty_mapping(self): + adapter = VSCodeClientAdapter() + assert adapter._extract_input_variables({}, "s") == [] + assert adapter._extract_input_variables(None, "s") == [] + + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") + def test_self_defined_http_headers_generate_inputs(self, mock_get_path): + """End-to-end: self-defined HTTP server with ${input:} in headers.""" + mock_get_path.return_value = self.temp_path + + server_info = { + "name": "my-server", + "remotes": [ + { + "transport_type": "http", + "url": "https://my-server.example.com/mcp/", + "headers": [ + {"name": "Authorization", "value": "Bearer ${input:my-server-token}"}, + {"name": "X-Project", "value": "${input:my-server-project}"}, + ], + } + ], + } + self.mock_registry.find_server_by_reference.return_value = server_info + + adapter = VSCodeClientAdapter() + result = adapter.configure_mcp_server( + server_url="my-server", server_name="my-server" + ) + + assert result is True + with open(self.temp_path, "r") as f: + config = json.load(f) + + inputs = config["inputs"] + input_ids = {v["id"] for v in inputs} + assert "my-server-token" in input_ids + assert "my-server-project" in input_ids + for inp in inputs: + assert inp["type"] == "promptString" + assert inp["password"] is True + + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") + def test_self_defined_stdio_env_generates_inputs(self, mock_get_path): + """End-to-end: self-defined stdio server with ${input:} in env.""" + mock_get_path.return_value = self.temp_path + + server_info = { + "name": "my-cli", + "_raw_stdio": { + "command": "my-cli", + "args": ["serve"], + "env": {"API_KEY": "${input:my-cli-api-key}"}, + }, + } + self.mock_registry.find_server_by_reference.return_value = server_info + + adapter = VSCodeClientAdapter() + result = adapter.configure_mcp_server( + server_url="my-cli", server_name="my-cli" + ) + + assert result is True + with open(self.temp_path, "r") as f: + config = json.load(f) + + inputs = config["inputs"] + assert len(inputs) == 1 + assert inputs[0]["id"] == "my-cli-api-key" + + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") + def test_input_variables_dedup_across_servers(self, mock_get_path): + """Input variables already present in config are not duplicated.""" + mock_get_path.return_value = self.temp_path + + # Pre-populate with an existing input + with open(self.temp_path, "w") as f: + json.dump( + { + "servers": {}, + "inputs": [ + {"type": "promptString", "id": "my-server-token", "description": "existing", "password": True} + ], + }, + f, + ) + + server_info = { + "name": "my-server", + "remotes": [ + { + "transport_type": "http", + "url": "https://example.com/mcp/", + "headers": [ + {"name": "Authorization", "value": "Bearer ${input:my-server-token}"}, + ], + } + ], + } + self.mock_registry.find_server_by_reference.return_value = server_info + + adapter = VSCodeClientAdapter() + adapter.configure_mcp_server(server_url="my-server", server_name="my-server") + + with open(self.temp_path, "r") as f: + config = json.load(f) + + token_entries = [i for i in config["inputs"] if i["id"] == "my-server-token"] + assert len(token_entries) == 1 + + +class TestWarnInputVariables(unittest.TestCase): + """Tests for _warn_input_variables on adapters that don't support input prompts.""" + + def test_warning_emitted_for_input_reference(self, ): + mapping = {"Authorization": "Bearer ${input:my-token}"} + with patch("builtins.print") as mock_print: + MCPClientAdapter._warn_input_variables(mapping, "my-server", "Copilot CLI") + mock_print.assert_called_once() + msg = mock_print.call_args[0][0] + assert "my-token" in msg + assert "Copilot CLI" in msg + + def test_no_warning_for_plain_values(self): + mapping = {"Content-Type": "application/json"} + with patch("builtins.print") as mock_print: + MCPClientAdapter._warn_input_variables(mapping, "s", "Codex CLI") + mock_print.assert_not_called() + + def test_no_warning_for_empty_mapping(self): + with patch("builtins.print") as mock_print: + MCPClientAdapter._warn_input_variables({}, "s", "Codex CLI") + MCPClientAdapter._warn_input_variables(None, "s", "Codex CLI") + mock_print.assert_not_called() + + if __name__ == "__main__": unittest.main()