refactor: eliminate duplicate code blocks (phases 1-3)

.github/workflows/ci.yml

@@ -75,7 +75,7 @@ jobs:
       run: |
         uv run --extra dev python -m pylint \
           --disable=all --enable=R0801 \
-          --min-similarity-lines=50 \
+          --min-similarity-lines=10 \
           --fail-on=R0801 \
           src/apm_cli/
 

CHANGELOG.md

@@ -25,6 +25,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Refactored duplicate code blocks across MCP client adapters and marketplace modules: extracted shared helpers `_apply_pypi_homebrew_generic_config`, `_apply_auth_and_headers_impl`, and `_resolve_env_vars_with_prompting` into `MCPClientAdapter` base class; extracted `iter_semver_tags` into `marketplace._shared`; fixed homebrew formula slash-stripping, `Authorization` header skip-only-when-injected logic, and CI-aware prompting guard (`CI`/`APM_E2E_TESTS` env vars now suppress interactive prompts). (#1360)
+
 - **Breaking:** MCP registry client now speaks the official MCP Registry v0.1 spec. **Self-hosted registries must serve `/v0.1/` paths -- registries serving only the legacy `/v0/` paths will return 404.** The public registry at `api.mcp.github.com` and the official `registry.modelcontextprotocol.io` are unaffected. Python API: `SimpleRegistryClient.get_server_info(server_id)` is renamed to `get_server(server_name, version="latest")` (the parameter is now a server *name* per the spec, not a UUID); the old name remains for one minor as a `DeprecationWarning` shim. The legacy UUID strategy in `find_server_by_reference` is removed -- the spec keys per-server lookup on serverName. v0.1 package fields (`identifier`, `registryType`, `runtimeHint`, `packageArguments`, `runtimeArguments`, `environmentVariables`) are normalized at the registry boundary to the snake_case shape adapters consume. Thanks @fassmus for the report. (#1337, closes #1210)
 - `--marketplace-output PATH` is now hidden from `--help` and emits a stderr deprecation warning; it auto-translates to `--marketplace-path claude=PATH`. Removal tracked in #1318. (#1317)
 - `extends: org` now correctly layers `dependencies.require` and `dependencies.deny` from the parent policy when the child omits the `dependencies:` block entirely; `None` signals "no opinion" (transparent) while `[]` signals explicit override. (#1290)

CONTRIBUTING.md

@@ -281,6 +281,20 @@ If your changes affect how users interact with the project, update the documenta
 
 ## Extending APM
 
+### Adding or modifying an MCP client adapter
+
+The MCP client adapters (`src/apm_cli/adapters/client/`) inherit shared
+utilities from `MCPClientAdapter` in `base.py`.  When adding a new adapter
+or modifying an existing one:
+
+- Inherit from `MCPClientAdapter` and reuse the shared helpers
+  (`_apply_pypi_homebrew_generic_config`, `_apply_auth_and_headers_impl`,
+  `_resolve_env_vars_with_prompting`).
+- Do **not** copy-paste logic from sibling adapters -- the pylint R0801
+  similarity threshold is 10 lines and CI will fail on duplicated blocks.
+- For marketplace tag-parsing, use `marketplace._shared.iter_semver_tags`
+  rather than re-implementing the refs-iteration loop.
+
 ### How to add an experimental feature flag
 
 Use an experimental flag to de-risk rollout of a user-visible behavioural change that may need early adopter feedback. Do not add a flag for a bug fix, internal refactor, or any change that should simply ship as the default behaviour.

src/apm_cli/adapters/client/base.py

@@ -5,6 +5,8 @@
 from abc import ABC, abstractmethod
 from pathlib import Path
 
+from ...utils.console import _rich_error, _rich_warning
+
 _INPUT_VAR_RE = re.compile(r"\$\{input:([^}]+)\}")
 
 # Matches ${VAR} and ${env:VAR}, capturing VAR. Intentionally does NOT match
@@ -181,8 +183,8 @@ def _warn_input_variables(mapping, server_name, runtime_label):
                 if var_id in seen:
                     continue
                 seen.add(var_id)
-                print(
-                    f"[!]  Warning: ${{input:{var_id}}} in server "
+                _rich_warning(
+                    f"${{input:{var_id}}} in server "
                     f"'{server_name}' will not be resolved -- "
                     f"{runtime_label} does not support input variable prompts"
                 )
@@ -196,3 +198,254 @@ def normalize_project_arg(self, value):
         ):
             return "."
         return value
+
+    # ------------------------------------------------------------------
+    # Shared server-info helpers (used by all adapter subclasses)
+    # ------------------------------------------------------------------
+
+    def _fetch_server_info(self, server_url: str, server_info_cache: dict | None) -> dict | None:
+        """Look up *server_url* in *server_info_cache* or fetch from registry.
+
+        Prints a user-visible error and returns ``None`` when the server is
+        not found, so callers can do a simple ``if server_info is None: return False``
+        guard and the error message stays consistent across adapters.
+
+        Args:
+            server_url: Registry reference (``owner/repo`` or full URL).
+            server_info_cache: Optional pre-fetched cache; ``None`` skips
+                the cache lookup.
+
+        Returns:
+            Server-info dict on success; ``None`` when not found.
+        """
+        if server_info_cache and server_url in server_info_cache:
+            return server_info_cache[server_url]
+        server_info = self.registry_client.find_server_by_reference(server_url)
+        if not server_info:
+            _rich_error(f"Error: MCP server '{server_url}' not found in registry")
+            return None
+        return server_info
+
+    @staticmethod
+    def _determine_config_key(server_url: str, server_name: str) -> str:
+        """Return the configuration key to use for *server_url*/*server_name*.
+
+        The caller-supplied *server_name* takes precedence; if empty the last
+        path segment of *server_url* is used as a fallback, which mirrors the
+        convention ``owner/repo -> repo``.
+
+        Args:
+            server_url: Registry reference used as fallback source.
+            server_name: Explicit caller-supplied name (may be empty string).
+
+        Returns:
+            Non-empty configuration key string.
+        """
+        if server_name:
+            return server_name
+        if "/" in server_url:
+            return server_url.split("/")[-1]
+        return server_url
+
+    @staticmethod
+    def _apply_pypi_homebrew_generic_config(
+        config: dict,
+        registry_name: str,
+        package_name: str,
+        runtime_hint: str,
+        processed_runtime_args: list,
+        processed_package_args: list,
+        resolved_env: dict,
+    ) -> None:
+        """Apply pypi / homebrew / generic (uvx / brew / npx) run config to *config*.
+
+        Mutates *config* in-place with ``command``, ``args``, and optionally
+        ``env`` keys appropriate for the detected registry type.
+
+        Args:
+            config: Mutable server-config dict to populate.
+            registry_name: Registry identifier (``"pypi"``, ``"homebrew"``,
+                ``"npm"``, or any other string treated as generic).
+            package_name: Base package / formula / module name.
+            runtime_hint: Caller-specified runtime hint (e.g. ``"uvx"``).
+            processed_runtime_args: Fully resolved positional args for the
+                runtime launcher.
+            processed_package_args: Fully resolved positional args appended
+                after the package name.
+            resolved_env: Pre-resolved environment variables dict; an empty
+                dict is omitted.
+        """
+        if registry_name == "pypi":
+            launcher = runtime_hint or "uvx"
+            config["command"] = launcher
+            config["args"] = [package_name] + processed_runtime_args + processed_package_args  # noqa: RUF005
+        elif registry_name == "homebrew":
+            formula_name = package_name.split("/")[-1] if "/" in package_name else package_name
+            config["command"] = formula_name
+            config["args"] = processed_runtime_args + processed_package_args
+        else:
+            # Generic / npm-compatible fallback
+            config["command"] = "npx"
+            config["args"] = processed_runtime_args + ["-y", package_name] + processed_package_args  # noqa: RUF005
+        if resolved_env:
+            config["env"] = resolved_env
+
+    def _apply_auth_and_headers_impl(
+        self,
+        config: dict,
+        remote: dict,
+        server_info: dict,
+        env_overrides: dict,
+        runtime_label: str,
+        token_manager_class,
+    ) -> None:
+        """Core implementation of GitHub-token injection and header merging.
+
+        Factored out so that each concrete adapter subclass can supply its own
+        *token_manager_class* (looked up from the subclass module's namespace),
+        allowing :func:`unittest.mock.patch` to intercept the class at the
+        right module scope in tests.
+
+        Args:
+            config: Mutable config dict updated in place.
+            remote: Registry remote entry (may contain a ``"headers"`` list).
+            server_info: Registry server metadata used for name / URL lookup.
+            env_overrides: Caller-supplied env-var override mapping.
+            runtime_label: Label for diagnostic messages.
+            token_manager_class: The ``GitHubTokenManager`` class (or mock) to
+                instantiate.  Passed by the caller so tests can patch the right
+                module-level name.
+        """
+        server_name = server_info.get("name", "")
+        is_github_server = self._is_github_server(server_name, remote.get("url", ""))
+        local_token_injected = False
+        if is_github_server:
+            _tm = token_manager_class()
+            github_token = _tm.get_token_for_purpose("copilot") or os.getenv(
+                "GITHUB_PERSONAL_ACCESS_TOKEN"
+            )
+            if github_token:
+                config["headers"] = {"Authorization": f"Bearer {github_token}"}
+                local_token_injected = True
+        headers = remote.get("headers", [])
+        if headers:
+            if "headers" not in config:
+                config["headers"] = {}
+            for header in headers:
+                header_name = header.get("name", "")
+                header_value = header.get("value", "")
+                if header_name and header_value:
+                    if header_name == "Authorization" and local_token_injected:
+                        continue
+                    resolved_value = self._resolve_env_variable(
+                        header_name, header_value, env_overrides
+                    )
+                    config["headers"][header_name] = resolved_value
+        if config.get("headers"):
+            self._warn_input_variables(
+                config["headers"], server_info.get("name", ""), runtime_label
+            )
+
+    @staticmethod
+    def _resolve_env_vars_with_prompting(
+        env_vars: list,
+        env_overrides: dict,
+        default_github_env: dict,
+    ) -> dict:
+        """Resolve *env_vars* from overrides, environment, or interactive prompts.
+
+        Identical logic shared between
+        :meth:`CopilotClientAdapter._process_environment_variables` and
+        :meth:`CodexClientAdapter._process_environment_variables`.
+
+        All imports are deferred so that ``rich.prompt`` (an optional
+        dependency) is never imported at module load time.
+
+        Args:
+            env_vars: List of env-var descriptor dicts from the registry.
+            env_overrides: Pre-collected ``{name: value}`` overrides (empty
+                dict when none).
+            default_github_env: Mapping of well-known GitHub variable names
+                to their preferred environment-variable lookup names.
+
+        Returns:
+            ``resolved`` dict mapping each env-var name to its resolved value
+            (empty string when unresolvable).
+        """
+        import sys
+
+        env_overrides = env_overrides or {}
+        resolved: dict = {}
+
+        # Determine whether interactive prompting is available.
+        # If env_overrides is provided the CLI has already collected variables -- never prompt again.
+        skip_prompting = (
+            bool(env_overrides)
+            or bool(os.getenv("CI"))
+            or bool(os.getenv("APM_E2E_TESTS"))
+            or not sys.stdout.isatty()
+            or not sys.stdin.isatty()
+        )
+
+        # First pass: identify variables with empty values to warn the user.
+        empty_value_vars = [ev for ev in env_vars if ev.get("required") and not ev.get("value")]
+        if empty_value_vars and skip_prompting:
+            var_names = [ev.get("name") for ev in empty_value_vars]
+            _rich_warning(
+                f"Warning: The following required environment variables have no default "
+                f"value and cannot be prompted in non-interactive mode: {var_names}"
+            )
+
+        for env_var in env_vars:
+            name = env_var.get("name", "")
+            if not name:
+                continue
+
+            # Priority 1: caller-supplied override
+            if name in env_overrides:
+                resolved[name] = env_overrides[name]
+                continue
+
+            # Priority 2: check GitHub-specific defaults (values are literal defaults, not env-var names)
+            if name in default_github_env:
+                resolved[name] = os.getenv(name) or default_github_env[name]
+                continue
+
+            # Priority 3: environment variable with the same name
+            env_val = os.getenv(name, "")
+            if env_val:
+                resolved[name] = env_val
+                continue
+
+            # Priority 4: interactive prompt
+            default_value = env_var.get("value", "")
+            required = env_var.get("required", False)
+
+            if not skip_prompting:
+                from rich.prompt import Prompt
+
+                description = env_var.get("description", "")
+                prompt_text = f"Enter value for {name}"
+                if description:
+                    prompt_text += f" ({description})"
+                is_secret = "token" in name.lower() or "key" in name.lower()
+                user_input = Prompt.ask(
+                    prompt_text,
+                    default=default_value,
+                    password=True  # noqa: SIM210
+                    if is_secret
+                    else False,
+                )
+                resolved[name] = user_input
+            elif default_value:
+                resolved[name] = default_value
+            elif required:
+                _rich_warning(
+                    f"Warning: Required environment variable '{name}' could not be resolved. "
+                    f"The MCP server may not function correctly."
+                )
+                resolved[name] = ""
+            else:
+                resolved[name] = default_value
+
+        return resolved