fix(mcp): drop previous-transport keys on Claude server redeclaration

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Claude MCP redeclarations drop stale transport fields and repair mixed entries when rewritten, while preserving partial updates and unmanaged configuration. -- by @edenfunf (#3041)
 - Cursor rules now use comma-joined `globs` and readable descriptions, while retaining safe escaping for control characters. (by @YGuyomar, #3011)
 
 ## [0.32.0] - 2026-09-25

docs/src/content/docs/consumer/install-mcp-servers.md

@@ -131,6 +131,17 @@ never enter `mcp.json`. A required variable without a collected value or
 default declines that target configuration; VS Code treats `workspaceFolder`
 as its built-in `${workspaceFolder}` token.
 
+Claude Code is the one target whose entries are merged key by key rather
+than replaced, so keys APM does not manage (a hand-authored OAuth block, for
+example) survive a reinstall. The keys describing a transport are not among
+them: an entry is rewritten to carry only the transport its declaration names,
+so redeclaring a server from `http` to `stdio` drops the previous `url` and
+`headers` instead of leaving both transports on one entry. This cleanup also
+repairs older mixed entries, but only when APM writes that server. An unchanged
+self-defined declaration with matching lock state can skip the write; repeating
+that install does not automatically repair a mixed entry. Rewriting an unchanged,
+valid entry preserves its field order.
+
 For VS Code and Copilot-family adapters, non-container `npm`, `pypi`,
 and generic packages preserve typed v0.1 `runtimeArguments` and
 `packageArguments` in authored order, with exactly one semantic package

packages/apm-guide/.apm/skills/apm-usage/commands.md

@@ -96,6 +96,11 @@ A normal project install creates or updates `apm.lock.yaml` when the manifest de
 
 `apm install --frozen` validates package and MCP lock state before lockfile, target config, deployment, or cache mutation. A missing or stale MCP-only lock exits nonzero without writing; run normal `apm install` to repair it. `--only=mcp` follows the same guard. Add-style `--mcp NAME` is incompatible with `--frozen` because it mutates `apm.yml`.
 
+Claude MCP rewrites remove the previous transport's fields while preserving
+unmanaged keys such as OAuth blocks. This also repairs older mixed entries
+when they are written. An unchanged self-defined declaration with matching
+lock state may skip the write, so repeating that install is not a migration.
+
 ### Registry MCP runtime variables
 
 For registry MCP runtime variables, `apm install` prompts once for a required

src/apm_cli/adapters/client/claude.py

@@ -63,6 +63,14 @@ class ClaudeClientAdapter(CopilotClientAdapter):
     _CONVERGED_SKILL_PREFIX = ".agents/skills/"
     _CLAUDE_SKILL_PREFIX = ".claude/skills/"
 
+    # Entry ``type`` values Claude Code uses for URL-addressed servers.
+    _REMOTE_TYPES = ("http", "sse", "streamable-http")
+
+    # Transport fields are replaced across families; a partial update can
+    # retain an omitted type only when it still belongs to the selected family.
+    _REMOTE_TRANSPORT_KEYS = frozenset({"type", "url", "headers"})
+    _STDIO_TRANSPORT_KEYS = frozenset({"type", "command", "args", "env", "cwd"})
+
     @classmethod
     def _rewrite_self_defined_skill_command(cls, command: str) -> str:
         if isinstance(command, str) and command.startswith(cls._CONVERGED_SKILL_PREFIX):
@@ -75,8 +83,13 @@ def _format_server_config(self, server_info, env_overrides=None, runtime_vars=No
             config["command"] = self._rewrite_self_defined_skill_command(config["command"])
         return config
 
-    @staticmethod
-    def _normalize_mcp_entry_for_claude_code(entry: dict) -> dict:
+    @classmethod
+    def _is_remote_mcp_entry(cls, entry: dict) -> bool:
+        """Return whether *entry* describes a remote (URL-addressed) server."""
+        return bool(entry.get("url")) or entry.get("type") in cls._REMOTE_TYPES
+
+    @classmethod
+    def _normalize_mcp_entry_for_claude_code(cls, entry: dict) -> dict:
         """Normalize a server entry to Claude Code's on-disk shape.
 
         For remote servers, keep ``type``/``url``/``headers`` per Claude
@@ -91,11 +104,8 @@ def _normalize_mcp_entry_for_claude_code(entry: dict) -> dict:
         if not isinstance(entry, dict):
             return entry
         out = dict(entry)
-        url = out.get("url")
-        t = out.get("type")
-        is_remote = bool(url) or t in ("http", "sse", "streamable-http")
 
-        if is_remote:
+        if cls._is_remote_mcp_entry(out):
             if out.get("id") in ("", None):
                 out.pop("id", None)
             if out.get("tools") == ["*"]:
@@ -110,22 +120,51 @@ def _normalize_mcp_entry_for_claude_code(entry: dict) -> dict:
             out.pop("id", None)
         return out
 
-    @staticmethod
-    def _merge_mcp_server_dicts(existing_servers: dict, config_updates: dict) -> None:
+    @classmethod
+    def _retained_previous_entry(cls, prev: dict, new_cfg: dict) -> dict:
+        """Return *prev* without the keys of the transport *new_cfg* is not.
+
+        A redeclaration that switches transport must not leave the previous
+        transport's keys behind.  A surviving ``url`` keeps the entry
+        classified as remote, so the stdio shape never wins and a stale
+        ``Authorization`` header stays in the Claude Code config.
+
+        The keys are selected from the update rather than from a comparison
+        of the two entries, so an entry already carrying both transports --
+        written by a release that merged them unconditionally -- is repaired
+        by the next install that writes it instead of matching its own mixed shape and
+        surviving.  Keys APM does not manage (hand-authored OAuth blocks and
+        the like) describe no transport and are retained either way.
+        """
+        declares_transport = any(new_cfg.get(key) for key in ("type", "url", "command"))
+        is_remote = cls._is_remote_mcp_entry(new_cfg if declares_transport else prev)
+        stale = cls._STDIO_TRANSPORT_KEYS if is_remote else cls._REMOTE_TRANSPORT_KEYS
+        compatible_types = cls._REMOTE_TYPES if is_remote else ("local", "stdio")
+        # Keep a compatible type in place so repeated writes retain field order.
+        return {
+            key: value
+            for key, value in prev.items()
+            if key not in stale or (key == "type" and value in compatible_types)
+        }
+
+    @classmethod
+    def _merge_mcp_server_dicts(cls, existing_servers: dict, config_updates: dict) -> None:
         """Merge *config_updates* into *existing_servers* in place.
 
         Per-server entries are shallow-merged: ``{**old, **new}`` so keys present
-        only on plugin- or hand-authored configs (e.g. ``type``, OAuth blocks)
+        only on plugin- or hand-authored configs (e.g. OAuth blocks)
         survive when an update omits them.  Keys in *new* overwrite *old* on
         conflict so APM/registry installs still refresh ``command``/``args``/etc.
+        The previous entry first drops the keys of the transport the update
+        does not declare, so the merged entry describes exactly one.
         """
         for name, new_cfg in config_updates.items():
             if not isinstance(new_cfg, dict):
                 existing_servers[name] = new_cfg
                 continue
             prev = existing_servers.get(name)
             if isinstance(prev, dict):
-                merged = {**prev, **new_cfg}
+                merged = {**cls._retained_previous_entry(prev, new_cfg), **new_cfg}
                 existing_servers[name] = merged
             else:
                 existing_servers[name] = dict(new_cfg)

tests/integration/test_claude_mcp_transport_lifecycle.py

@@ -0,0 +1,230 @@
+"""Hermetic Python CLI lifecycles for Claude transport redeclarations."""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from typing import Any
+from urllib.parse import urlparse
+
+import pytest
+
+from apm_cli.utils.yaml_io import dump_yaml, load_yaml
+from tests.utils.apm_lifecycle_runner import ApmLifecycleRunner, CommandResult
+from tests.utils.isolated_apm_environment import IsolatedApmEnvironment
+from tests.utils.lifecycle_state import LifecycleStateSnapshot
+from tests.utils.local_package import LocalPackageFactory
+
+pytestmark = [pytest.mark.integration, pytest.mark.e2e, pytest.mark.lifecycle_smoke]
+
+_SERVER = "transport-fixture"
+_OAUTH = {"accountUuid": "fixture-account"}
+_HEADERS = {"Authorization": "Bearer fixture-only"}
+_TRANSPORT_KEYS = {
+    "http": {"command", "args", "env", "cwd"},
+    "stdio": {"url", "headers"},
+}
+
+
+def _declaration(transport: str, revision: int = 1) -> dict[str, Any]:
+    declaration: dict[str, Any] = {
+        "name": _SERVER,
+        "registry": False,
+        "transport": transport,
+    }
+    if transport == "http":
+        declaration.update(url=f"https://example.invalid/mcp/{revision}", headers=_HEADERS)
+    else:
+        declaration.update(
+            command="python",
+            args=["-m", "fixture_server", str(revision)],
+            env={"FIXTURE_TOKEN": "stdio-value"},
+        )
+    return declaration
+
+
+def _write_config(path: Path, document: dict[str, Any]) -> None:
+    path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8")
+
+
+def _assert_success(result: CommandResult) -> None:
+    assert result.returncode == 0, (
+        f"command={result.command!r}\nstdout={result.stdout}\nstderr={result.stderr}"
+    )
+
+
+@dataclass
+class _Scenario:
+    isolated: IsolatedApmEnvironment
+    project: Path
+    user_scope: bool
+    runner: ApmLifecycleRunner
+
+    def root(self, user_scope: bool) -> Path:
+        return self.isolated.config_root if user_scope else self.project
+
+    def config_path(self, user_scope: bool) -> Path:
+        return self.isolated.home / ".claude.json" if user_scope else self.project / ".mcp.json"
+
+    def document(self) -> dict[str, Any]:
+        return json.loads(self.config_path(self.user_scope).read_text(encoding="utf-8"))
+
+    def declare(self, transport: str, revision: int = 1) -> None:
+        path = self.root(self.user_scope) / "apm.yml"
+        manifest = load_yaml(path)
+        manifest["dependencies"]["mcp"] = [_declaration(transport, revision)]
+        dump_yaml(manifest, path)
+
+    def install(self, *flags: str, user_scope: bool | None = None) -> CommandResult:
+        scope = self.user_scope if user_scope is None else user_scope
+        return self.runner.run(
+            (
+                "install",
+                "--target",
+                "claude",
+                "--no-policy",
+                *(("--global",) if scope else ()),
+                *flags,
+            ),
+            scenario_id=f"claude-transport-{'user' if scope else 'project'}",
+            cwd=self.project,
+            env=self.isolated.subprocess_env(overrides={"CLAUDE_CONFIG_DIR": ""}),
+        )
+
+    def snapshot(self, user_scope: bool) -> tuple[LifecycleStateSnapshot, bytes]:
+        return (
+            LifecycleStateSnapshot.capture(
+                self.root(user_scope),
+                config_paths=() if user_scope else (PurePosixPath(".mcp.json"),),
+            ),
+            self.config_path(user_scope).read_bytes(),
+        )
+
+
+@pytest.fixture(params=[False, True], ids=["project", "user"])
+def scenario(
+    tmp_path: Path, apm_engine_command: tuple[str, ...], request: pytest.FixtureRequest
+) -> _Scenario:
+    isolated = IsolatedApmEnvironment.create(tmp_path / "scenario", base_env=dict(os.environ))
+    package = LocalPackageFactory(isolated.work_root).create(
+        "claude-consumer", mcp_dependencies=(_declaration("stdio"),), targets=("claude",)
+    )
+    (package.root / ".claude").mkdir()
+    dump_yaml(load_yaml(package.manifest_path), isolated.config_root / "apm.yml")
+    result = _Scenario(
+        isolated, package.root, request.param, ApmLifecycleRunner(apm_engine_command)
+    )
+    for scope in (False, True):
+        _write_config(
+            result.config_path(scope),
+            {
+                "mcpServers": {
+                    "unmanaged": {
+                        "type": "http",
+                        "url": "https://untouched.invalid/mcp",
+                        "command": "do-not-normalize",
+                    }
+                },
+                "oauthAccount": {"accountUuid": "top-level-fixture"},
+                "preferences": {"keep": True},
+                "projects": {"private-project": {"mcpServers": {"keep": {"command": "keep"}}}},
+            },
+        )
+    return result
+
+
+def _prepare(scenario: _Scenario, transport: str) -> dict[str, Any]:
+    scenario.declare(transport)
+    _assert_success(scenario.install(user_scope=not scenario.user_scope))
+    _assert_success(scenario.install())
+    document = scenario.document()
+    document["mcpServers"][_SERVER]["oauthAccount"] = _OAUTH
+    if transport == "stdio":
+        document["mcpServers"][_SERVER]["cwd"] = "/fixture-cwd"
+    _write_config(scenario.config_path(scenario.user_scope), document)
+    return _unrelated(document)
+
+
+def _unrelated(document: dict[str, Any]) -> dict[str, Any]:
+    return {
+        **document,
+        "mcpServers": {
+            key: value for key, value in document["mcpServers"].items() if key != _SERVER
+        },
+    }
+
+
+def _assert_entry(scenario: _Scenario, transport: str, revision: int) -> None:
+    entry = scenario.document()["mcpServers"][_SERVER]
+    assert entry["type"] == transport
+    assert not (_TRANSPORT_KEYS[transport] & entry.keys())
+    assert entry["oauthAccount"] == _OAUTH
+    if transport == "http":
+        parsed = urlparse(entry["url"])
+        assert (parsed.scheme, parsed.hostname, parsed.path) == (
+            "https",
+            "example.invalid",
+            f"/mcp/{revision}",
+        )
+        assert entry["headers"] == _HEADERS
+    else:
+        assert entry["command"] == "python"
+        assert entry["args"] == ["-m", "fixture_server", str(revision)]
+        assert entry["env"] == {"FIXTURE_TOKEN": "stdio-value"}
+
+
+def _assert_repeat(scenario: _Scenario) -> None:
+    selected = scenario.snapshot(scenario.user_scope)
+    opposite = scenario.snapshot(not scenario.user_scope)
+    _assert_success(scenario.install())
+    assert scenario.snapshot(scenario.user_scope) == selected
+    assert scenario.snapshot(not scenario.user_scope) == opposite
+
+
+@pytest.mark.parametrize("initial", ["http", "stdio"])
+def test_transport_redeclaration_preserves_unowned_state(scenario: _Scenario, initial: str) -> None:
+    """Real install, denied frozen rewrite, rewrite and repeat converge safely."""
+    unrelated = _prepare(scenario, initial)
+    opposite = scenario.snapshot(not scenario.user_scope)
+    target = "stdio" if initial == "http" else "http"
+    scenario.declare(target)
+    before_frozen = scenario.snapshot(scenario.user_scope)
+    frozen = scenario.install("--frozen")
+    assert frozen.returncode != 0
+    assert f"MCP server '{_SERVER}' config differs" in frozen.stdout
+    assert scenario.snapshot(scenario.user_scope) == before_frozen
+    assert scenario.snapshot(not scenario.user_scope) == opposite
+
+    _assert_success(scenario.install())
+    _assert_entry(scenario, target, 1)
+    assert _unrelated(scenario.document()) == unrelated
+    assert scenario.snapshot(not scenario.user_scope) == opposite
+    _assert_repeat(scenario)
+
+
+@pytest.mark.parametrize("transport", ["http", "stdio"])
+def test_legacy_mixed_state_is_repaired_only_on_redeclaration(
+    scenario: _Scenario, transport: str
+) -> None:
+    """An unchanged install skips legacy state; actual same-family drift repairs it."""
+    unrelated = _prepare(scenario, transport)
+    document = scenario.document()
+    entry = document["mcpServers"][_SERVER]
+    if transport == "http":
+        entry.update(command="old", args=["old"], env={"OLD_TOKEN": "fixture-only"}, cwd="/old")
+    else:
+        entry.update(url="https://stale.invalid/mcp", headers={"Authorization": "old-fixture"})
+    _write_config(scenario.config_path(scenario.user_scope), document)
+    _assert_repeat(scenario)
+    opposite = scenario.snapshot(not scenario.user_scope)
+
+    scenario.declare(transport, revision=2)
+    _assert_success(scenario.install())
+    _assert_entry(scenario, transport, 2)
+    if transport == "stdio":
+        assert scenario.document()["mcpServers"][_SERVER]["cwd"] == "/fixture-cwd"
+    assert _unrelated(scenario.document()) == unrelated
+    assert scenario.snapshot(not scenario.user_scope) == opposite
+    _assert_repeat(scenario)