diff --git a/docs/dependencies.md b/docs/dependencies.md
index 12d718103..6f676694a 100644
--- a/docs/dependencies.md
+++ b/docs/dependencies.md
@@ -74,9 +74,13 @@ dependencies:
     - microsoft/apm-sample-package  # Design standards, prompts
     - github/awesome-copilot/skills/review-and-refactor  # Code review skill
   mcp:
-    - io.github.github/github-mcp-server
+    - io.github.github/github-mcp-server          # Registry reference
 ```
 
+MCP dependencies resolve via the MCP server registry (e.g. `io.github.github/github-mcp-server`).
+
+MCP dependencies declared by transitive APM packages are collected automatically during `apm install`.
+
 ### 2. Install Dependencies
 
 ```bash
diff --git a/docs/integrations.md b/docs/integrations.md
index 9323e7d56..1cfc840c2 100644
--- a/docs/integrations.md
+++ b/docs/integrations.md
@@ -515,6 +515,7 @@ APM provides first-class support for MCP servers:
 # apm.yml - MCP dependencies
 dependencies:
   mcp:
+    # Registry references
     - ghcr.io/github/github-mcp-server
     - ghcr.io/modelcontextprotocol/filesystem-server
     - ghcr.io/modelcontextprotocol/postgres-server
diff --git a/src/apm_cli/cli.py b/src/apm_cli/cli.py
index 874376b7a..d67fb77f7 100644
--- a/src/apm_cli/cli.py
+++ b/src/apm_cli/cli.py
@@ -701,6 +701,15 @@ def install(ctx, packages, runtime, exclude, only, update, dry_run, verbose):
         elif should_install_apm and not apm_deps:
             _rich_info("No APM dependencies found in apm.yml")
 
+        # Collect transitive MCP dependencies from resolved APM packages
+        apm_modules_path = Path.cwd() / "apm_modules"
+        if should_install_mcp and apm_modules_path.exists():
+            lock_path = Path.cwd() / "apm.lock"
+            transitive_mcp = _collect_transitive_mcp_deps(apm_modules_path, lock_path)
+            if transitive_mcp:
+                _rich_info(f"Collected {len(transitive_mcp)} transitive MCP dependency(ies)")
+                mcp_deps = _deduplicate_mcp_deps(mcp_deps + transitive_mcp)
+
         # Continue with MCP installation (existing logic)
         mcp_count = 0
         if should_install_mcp and mcp_deps:
@@ -2212,24 +2221,107 @@ def matches_filter(dep):
         raise RuntimeError(f"Failed to resolve APM dependencies: {e}")
 
 
+def _collect_transitive_mcp_deps(apm_modules_dir: Path, lock_path: Path = None) -> list:
+    """Collect MCP dependencies from resolved APM packages listed in apm.lock.
+
+    Only scans apm.yml files for packages present in apm.lock to avoid
+    picking up stale/orphaned packages from previous installs.
+    Falls back to scanning all apm.yml files if no lock file is available.
+
+    Args:
+        apm_modules_dir: Path to the apm_modules directory.
+        lock_path: Path to the apm.lock file (optional).
+
+    Returns:
+        List of MCP dependency entries (str or dict).
+    """
+    if not apm_modules_dir.exists():
+        return []
+
+    from apm_cli.models.apm_package import APMPackage
+
+    # Build set of expected apm.yml paths from apm.lock
+    locked_paths = None
+    if lock_path and lock_path.exists():
+        lockfile = LockFile.read(lock_path)
+        if lockfile is not None:
+            locked_paths = builtins.set()
+            for dep in lockfile.get_all_dependencies():
+                if dep.repo_url:
+                    yml = apm_modules_dir / dep.repo_url / dep.virtual_path / "apm.yml" if dep.virtual_path else apm_modules_dir / dep.repo_url / "apm.yml"
+                    locked_paths.add(yml.resolve())
+
+    # Prefer iterating lock-derived paths directly (existing files only).
+    # Fall back to full scan only when lock parsing is unavailable.
+    if locked_paths is not None:
+        apm_yml_paths = [path for path in sorted(locked_paths) if path.exists()]
+    else:
+        apm_yml_paths = apm_modules_dir.rglob("apm.yml")
+
+    collected = []
+    for apm_yml_path in apm_yml_paths:
+        try:
+            pkg = APMPackage.from_apm_yml(apm_yml_path)
+            mcp = pkg.get_mcp_dependencies()
+            if mcp:
+                collected.extend(mcp)
+        except Exception:
+            # Skip packages that fail to parse
+            continue
+    return collected
+
+
+def _deduplicate_mcp_deps(deps: list) -> list:
+    """Deduplicate a mixed list of MCP deps (strings and dicts).
+
+    Strings are deduped by value; dicts are deduped by their 'name' key.
+
+    Args:
+        deps: Mixed list of str and dict MCP entries.
+
+    Returns:
+        Deduplicated list preserving insertion order.
+    """
+    seen_strings: builtins.set = builtins.set()
+    seen_names: builtins.set = builtins.set()
+    result = []
+    for dep in deps:
+        if isinstance(dep, str):
+            if dep not in seen_strings:
+                seen_strings.add(dep)
+                result.append(dep)
+        elif isinstance(dep, dict):
+            name = dep.get("name", "")
+            if name and name not in seen_names:
+                seen_names.add(name)
+                result.append(dep)
+            elif not name and dep not in result:
+                result.append(dep)
+    return result
+
+
 def _install_mcp_dependencies(
-    mcp_deps: List[str], runtime: str = None, exclude: str = None, verbose: bool = False
+    mcp_deps: list, runtime: str = None, exclude: str = None, verbose: bool = False
 ):
     """Install MCP dependencies using existing logic.
 
     Args:
-        mcp_deps: List of MCP dependency names
+        mcp_deps: List of MCP dependency entries (registry strings)
         runtime: Target specific runtime only
         exclude: Exclude specific runtime from installation
         verbose: Show detailed installation information
 
     Returns:
-        int: Number of MCP servers configured
+        int: Number of MCP servers newly configured
     """
     if not mcp_deps:
         _rich_warning("No MCP dependencies found in apm.yml")
         return 0
 
+    # Filter to registry strings only (inline dicts are preserved during
+    # parsing for data fidelity but not installed — see #132).
+    registry_deps = [d for d in mcp_deps if isinstance(d, str)]
+
     console = _get_console()
 
     # Start MCP section with clean header
@@ -2356,99 +2448,116 @@ def _install_mcp_dependencies(
 
     # Use the new registry operations module for better server detection
     configured_count = 0
-    try:
-        from apm_cli.registry.operations import MCPServerOperations
 
-        operations = MCPServerOperations()
+    # --- Registry-based deps (strings) ---
+    if registry_deps:
+        try:
+            from apm_cli.registry.operations import MCPServerOperations
 
-        # Early validation: check if all servers exist in registry (fail-fast like npm)
-        if verbose:
-            _rich_info(f"Validating {len(mcp_deps)} servers...")
-        valid_servers, invalid_servers = operations.validate_servers_exist(mcp_deps)
+            operations = MCPServerOperations()
 
-        if invalid_servers:
-            _rich_error(
-                f"Server(s) not found in registry: {', '.join(invalid_servers)}"
-            )
-            _rich_info("Run 'apm mcp search <query>' to find available servers")
-            raise RuntimeError(
-                f"Cannot install {len(invalid_servers)} missing server(s)"
-            )
+            # Early validation: check if all servers exist in registry (fail-fast like npm)
+            if verbose:
+                _rich_info(f"Validating {len(registry_deps)} registry servers...")
+            valid_servers, invalid_servers = operations.validate_servers_exist(registry_deps)
 
-        if not valid_servers:
-            if console:
-                console.print("└─ [green]No servers to install[/green]")
-            else:
-                _rich_success("No servers to install")
-            return 0
+            if invalid_servers:
+                _rich_error(
+                    f"Server(s) not found in registry: {', '.join(invalid_servers)}"
+                )
+                _rich_info("Run 'apm mcp search <query>' to find available servers")
+                raise RuntimeError(
+                    f"Cannot install {len(invalid_servers)} missing server(s)"
+                )
 
-        # Check which valid servers actually need installation
-        servers_to_install = operations.check_servers_needing_installation(
-            target_runtimes, valid_servers
-        )
+            if valid_servers:
+                # Check which valid servers actually need installation
+                servers_to_install = operations.check_servers_needing_installation(
+                    target_runtimes, valid_servers
+                )
+                already_configured_servers = [
+                    dep for dep in valid_servers if dep not in servers_to_install
+                ]
 
-        if not servers_to_install:
-            # All already configured
-            if console:
-                for dep in mcp_deps:
-                    console.print(
-                        f"│  [green]✓[/green] {dep} [dim](already configured)[/dim]"
-                    )
-                console.print("└─ [green]All servers up to date[/green]")
-            else:
-                _rich_success("All MCP servers already configured")
-            return len(mcp_deps)
-        else:
-            # Batch fetch server info once to avoid duplicate registry calls
-            if verbose:
-                _rich_info(f"Installing {len(servers_to_install)} servers...")
-            server_info_cache = operations.batch_fetch_server_info(servers_to_install)
+                if not servers_to_install:
+                    # All already configured
+                    if console:
+                        for dep in already_configured_servers:
+                            console.print(
+                                f"│  [green]✓[/green] {dep} [dim](already configured)[/dim]"
+                            )
+                    else:
+                        _rich_success("All registry MCP servers already configured")
+                else:
+                    # Surface already-configured servers distinctly from newly configured ones
+                    if already_configured_servers:
+                        if console:
+                            for dep in already_configured_servers:
+                                console.print(
+                                    f"│  [green]✓[/green] {dep} [dim](already configured)[/dim]"
+                                )
+                        elif verbose:
+                            _rich_info(
+                                "Already configured registry MCP servers: "
+                                f"{', '.join(already_configured_servers)}"
+                            )
 
-            # Collect both environment and runtime variables using cached server info
-            shared_env_vars = operations.collect_environment_variables(
-                servers_to_install, server_info_cache
-            )
-            shared_runtime_vars = operations.collect_runtime_variables(
-                servers_to_install, server_info_cache
-            )
+                    # Batch fetch server info once to avoid duplicate registry calls
+                    if verbose:
+                        _rich_info(f"Installing {len(servers_to_install)} servers...")
+                    server_info_cache = operations.batch_fetch_server_info(servers_to_install)
 
-            # Install for each target runtime using cached server info and shared variables
-            for dep in servers_to_install:
-                if console:
-                    console.print(f"│  [cyan]⬇️[/cyan]  {dep}")
-                    console.print(
-                        f"│     └─ Configuring for {', '.join([rt.title() for rt in target_runtimes])}..."
+                    # Collect both environment and runtime variables using cached server info
+                    shared_env_vars = operations.collect_environment_variables(
+                        servers_to_install, server_info_cache
                     )
-
-                for rt in target_runtimes:
-                    if verbose:
-                        _rich_info(f"Configuring {rt}...")
-                    _install_for_runtime(
-                        rt,
-                        [dep],  # Install one at a time for better output
-                        shared_env_vars,
-                        server_info_cache,
-                        shared_runtime_vars,
+                    shared_runtime_vars = operations.collect_runtime_variables(
+                        servers_to_install, server_info_cache
                     )
 
-                if console:
-                    console.print(
-                        f"│  [green]✓[/green]  {dep} → {', '.join([rt.title() for rt in target_runtimes])}"
-                    )
-                configured_count += 1
+                    # Install for each target runtime using cached server info and shared variables
+                    for dep in servers_to_install:
+                        if console:
+                            console.print(f"│  [cyan]⬇️[/cyan]  {dep}")
+                            console.print(
+                                f"│     └─ Configuring for {', '.join([rt.title() for rt in target_runtimes])}..."
+                            )
+
+                        for rt in target_runtimes:
+                            if verbose:
+                                _rich_info(f"Configuring {rt}...")
+                            _install_for_runtime(
+                                rt,
+                                [dep],  # Install one at a time for better output
+                                shared_env_vars,
+                                server_info_cache,
+                                shared_runtime_vars,
+                            )
+
+                        if console:
+                            console.print(
+                                f"│  [green]✓[/green]  {dep} → {', '.join([rt.title() for rt in target_runtimes])}"
+                            )
+                        configured_count += 1
+
+        except ImportError:
+            _rich_warning("Registry operations not available")
+            _rich_error("Cannot validate MCP servers without registry operations")
+            raise RuntimeError("Registry operations module required for MCP installation")
+
+    # Close the panel
+    if console:
+        if configured_count > 0:
+            console.print(
+                f"└─ [green]Configured {configured_count} server{'s' if configured_count != 1 else ''}[/green]"
+            )
+        else:
+            console.print("└─ [green]All servers up to date[/green]")
+
+    return configured_count
 
-            # Close the panel
-            if console:
-                console.print(
-                    f"└─ [green]Configured {configured_count} server{'s' if configured_count != 1 else ''}[/green]"
-                )
 
-        return configured_count
 
-    except ImportError:
-        _rich_warning("Registry operations not available")
-        _rich_error("Cannot validate MCP servers without registry operations")
-        raise RuntimeError("Registry operations module required for MCP installation")
 
 
 def _show_install_summary(
diff --git a/src/apm_cli/models/apm_package.py b/src/apm_cli/models/apm_package.py
index a5484f12d..0bd7704aa 100644
--- a/src/apm_cli/models/apm_package.py
+++ b/src/apm_cli/models/apm_package.py
@@ -709,7 +709,7 @@ class APMPackage:
     license: Optional[str] = None
     source: Optional[str] = None  # Source location (for dependencies)
     resolved_commit: Optional[str] = None  # Resolved commit SHA (for dependencies)
-    dependencies: Optional[Dict[str, List[Union[DependencyReference, str]]]] = None  # Mixed types for APM/MCP
+    dependencies: Optional[Dict[str, List[Union[DependencyReference, str, dict]]]] = None  # Mixed types for APM/MCP/inline
     scripts: Optional[Dict[str, str]] = None
     package_path: Optional[Path] = None  # Local path to package
     target: Optional[str] = None  # Target agent: vscode, claude, or all (applies to compile and install)
@@ -764,8 +764,8 @@ def from_apm_yml(cls, apm_yml_path: Path) -> "APMPackage":
                                     raise ValueError(f"Invalid APM dependency '{dep_str}': {e}")
                         dependencies[dep_type] = parsed_deps
                     else:
-                        # Other dependencies (like MCP) remain as strings
-                        dependencies[dep_type] = [str(dep) for dep in dep_list if isinstance(dep, str)]
+                        # Other dependencies (like MCP): keep strings and dicts
+                        dependencies[dep_type] = [dep for dep in dep_list if isinstance(dep, (str, dict))]
         
         # Parse package content type
         pkg_type = None
@@ -798,13 +798,12 @@ def get_apm_dependencies(self) -> List[DependencyReference]:
         # Filter to only return DependencyReference objects
         return [dep for dep in self.dependencies['apm'] if isinstance(dep, DependencyReference)]
     
-    def get_mcp_dependencies(self) -> List[str]:
-        """Get list of MCP dependencies (as strings for compatibility)."""
+    def get_mcp_dependencies(self) -> List[Union[str, dict]]:
+        """Get list of MCP dependencies (strings for registry, dicts for inline configs)."""
         if not self.dependencies or 'mcp' not in self.dependencies:
             return []
-        # MCP deps are stored as strings, not DependencyReference objects
-        return [str(dep) if isinstance(dep, DependencyReference) else dep 
-                for dep in self.dependencies.get('mcp', [])]
+        return [dep for dep in (self.dependencies.get('mcp') or [])
+                if isinstance(dep, (str, dict))]
     
     def has_apm_dependencies(self) -> bool:
         """Check if this package has APM dependencies."""
diff --git a/tests/unit/test_transitive_mcp.py b/tests/unit/test_transitive_mcp.py
new file mode 100644
index 000000000..e802a0012
--- /dev/null
+++ b/tests/unit/test_transitive_mcp.py
@@ -0,0 +1,370 @@
+"""Tests for transitive MCP dependency collection and deduplication."""
+
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+import yaml
+
+from apm_cli.models.apm_package import APMPackage
+from apm_cli.cli import (
+    _collect_transitive_mcp_deps,
+    _deduplicate_mcp_deps,
+    _install_mcp_dependencies,
+)
+
+
+# ---------------------------------------------------------------------------
+# APMPackage – MCP dict parsing
+# ---------------------------------------------------------------------------
+class TestAPMPackageMCPParsing:
+    """Ensure apm_package preserves both string and dict MCP entries."""
+
+    def test_parse_string_mcp_deps(self, tmp_path):
+        """String-only MCP deps parse correctly."""
+        yml = tmp_path / "apm.yml"
+        yml.write_text(yaml.dump({
+            "name": "pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": ["ghcr.io/some/server"]},
+        }))
+        pkg = APMPackage.from_apm_yml(yml)
+        deps = pkg.get_mcp_dependencies()
+
+        assert deps == ["ghcr.io/some/server"]
+
+    def test_parse_dict_mcp_deps(self, tmp_path):
+        """Inline dict MCP deps are preserved."""
+        inline = {"name": "my-srv", "type": "sse", "url": "https://example.com"}
+        yml = tmp_path / "apm.yml"
+        yml.write_text(yaml.dump({
+            "name": "pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": [inline]},
+        }))
+        pkg = APMPackage.from_apm_yml(yml)
+        deps = pkg.get_mcp_dependencies()
+
+        assert len(deps) == 1
+        assert isinstance(deps[0], dict)
+        assert deps[0]["name"] == "my-srv"
+
+    def test_parse_mixed_mcp_deps(self, tmp_path):
+        """A mix of string and dict entries is preserved in order."""
+        inline = {"name": "inline-srv", "type": "http", "url": "https://x"}
+        yml = tmp_path / "apm.yml"
+        yml.write_text(yaml.dump({
+            "name": "pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": ["registry-srv", inline]},
+        }))
+        pkg = APMPackage.from_apm_yml(yml)
+        deps = pkg.get_mcp_dependencies()
+
+        assert len(deps) == 2
+        assert isinstance(deps[0], str)
+        assert isinstance(deps[1], dict)
+
+    def test_no_mcp_section(self, tmp_path):
+        """Missing MCP section returns empty list."""
+        yml = tmp_path / "apm.yml"
+        yml.write_text(yaml.dump({
+            "name": "pkg",
+            "version": "1.0.0",
+        }))
+        pkg = APMPackage.from_apm_yml(yml)
+        assert pkg.get_mcp_dependencies() == []
+
+    def test_mcp_null_returns_empty(self, tmp_path):
+        """mcp: null should return empty list, not raise TypeError."""
+        yml = tmp_path / "apm.yml"
+        yml.write_text(yaml.dump({
+            "name": "pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": None},
+        }))
+        pkg = APMPackage.from_apm_yml(yml)
+        assert pkg.get_mcp_dependencies() == []
+
+    def test_mcp_empty_list_returns_empty(self, tmp_path):
+        """mcp: [] should return empty list."""
+        yml = tmp_path / "apm.yml"
+        yml.write_text(yaml.dump({
+            "name": "pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": []},
+        }))
+        pkg = APMPackage.from_apm_yml(yml)
+        assert pkg.get_mcp_dependencies() == []
+
+
+# ---------------------------------------------------------------------------
+# _collect_transitive_mcp_deps
+# ---------------------------------------------------------------------------
+class TestCollectTransitiveMCPDeps:
+    """Tests for scanning apm_modules/ for MCP deps."""
+
+    def test_empty_when_dir_missing(self, tmp_path):
+        result = _collect_transitive_mcp_deps(tmp_path / "nonexistent")
+        assert result == []
+
+    def test_collects_string_deps(self, tmp_path):
+        pkg_dir = tmp_path / "org" / "pkg-a"
+        pkg_dir.mkdir(parents=True)
+        (pkg_dir / "apm.yml").write_text(yaml.dump({
+            "name": "pkg-a",
+            "version": "1.0.0",
+            "dependencies": {"mcp": ["ghcr.io/a/server"]},
+        }))
+        result = _collect_transitive_mcp_deps(tmp_path)
+        assert result == ["ghcr.io/a/server"]
+
+    def test_collects_dict_deps(self, tmp_path):
+        inline = {"name": "kb", "type": "sse", "url": "https://kb.example.com"}
+        pkg_dir = tmp_path / "org" / "pkg-b"
+        pkg_dir.mkdir(parents=True)
+        (pkg_dir / "apm.yml").write_text(yaml.dump({
+            "name": "pkg-b",
+            "version": "1.0.0",
+            "dependencies": {"mcp": [inline]},
+        }))
+        result = _collect_transitive_mcp_deps(tmp_path)
+        assert len(result) == 1
+        assert result[0]["name"] == "kb"
+
+    def test_collects_from_multiple_packages(self, tmp_path):
+        for i, dep in enumerate(["ghcr.io/a/s1", "ghcr.io/b/s2"]):
+            d = tmp_path / "org" / f"pkg-{i}"
+            d.mkdir(parents=True)
+            (d / "apm.yml").write_text(yaml.dump({
+                "name": f"pkg-{i}",
+                "version": "1.0.0",
+                "dependencies": {"mcp": [dep]},
+            }))
+        result = _collect_transitive_mcp_deps(tmp_path)
+        assert len(result) == 2
+
+    def test_skips_unparseable_apm_yml(self, tmp_path):
+        pkg_dir = tmp_path / "org" / "bad-pkg"
+        pkg_dir.mkdir(parents=True)
+        (pkg_dir / "apm.yml").write_text("invalid: yaml: [")
+        # Should not raise
+        result = _collect_transitive_mcp_deps(tmp_path)
+        assert result == []
+
+    def test_lockfile_scopes_collection_to_locked_packages(self, tmp_path):
+        """Lock-file filtering should only collect MCP deps from locked packages."""
+        apm_modules = tmp_path / "apm_modules"
+        # Package that IS in the lock file
+        locked_dir = apm_modules / "org" / "locked-pkg"
+        locked_dir.mkdir(parents=True)
+        (locked_dir / "apm.yml").write_text(yaml.dump({
+            "name": "locked-pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": ["ghcr.io/locked/server"]},
+        }))
+        # Package that is NOT in the lock file (orphan)
+        orphan_dir = apm_modules / "org" / "orphan-pkg"
+        orphan_dir.mkdir(parents=True)
+        (orphan_dir / "apm.yml").write_text(yaml.dump({
+            "name": "orphan-pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": ["ghcr.io/orphan/server"]},
+        }))
+        # Write lock file referencing only the locked package
+        lock_path = tmp_path / "apm.lock"
+        lock_path.write_text(yaml.dump({
+            "lockfile_version": "1",
+            "dependencies": [
+                {"repo_url": "org/locked-pkg", "host": "github.com"},
+            ],
+        }))
+        result = _collect_transitive_mcp_deps(apm_modules, lock_path)
+        assert result == ["ghcr.io/locked/server"]
+
+    def test_lockfile_with_virtual_path(self, tmp_path):
+        """Lock-file filtering works for subdirectory (virtual_path) packages."""
+        apm_modules = tmp_path / "apm_modules"
+        # Subdirectory package matching lock entry
+        sub_dir = apm_modules / "org" / "monorepo" / "skills" / "azure"
+        sub_dir.mkdir(parents=True)
+        (sub_dir / "apm.yml").write_text(yaml.dump({
+            "name": "azure-skill",
+            "version": "1.0.0",
+            "dependencies": {"mcp": [{"name": "learn", "type": "http", "url": "https://learn.example.com"}]},
+        }))
+        # Another subdirectory NOT in the lock
+        other_dir = apm_modules / "org" / "monorepo" / "skills" / "other"
+        other_dir.mkdir(parents=True)
+        (other_dir / "apm.yml").write_text(yaml.dump({
+            "name": "other-skill",
+            "version": "1.0.0",
+            "dependencies": {"mcp": ["ghcr.io/other/server"]},
+        }))
+        lock_path = tmp_path / "apm.lock"
+        lock_path.write_text(yaml.dump({
+            "lockfile_version": "1",
+            "dependencies": [
+                {"repo_url": "org/monorepo", "host": "github.com", "virtual_path": "skills/azure"},
+            ],
+        }))
+        result = _collect_transitive_mcp_deps(apm_modules, lock_path)
+        assert len(result) == 1
+        assert result[0]["name"] == "learn"
+
+    def test_lockfile_paths_do_not_use_full_rglob_scan(self, tmp_path):
+        """When lock-derived paths are available, avoid full recursive scanning."""
+        apm_modules = tmp_path / "apm_modules"
+        locked_dir = apm_modules / "org" / "locked-pkg"
+        locked_dir.mkdir(parents=True)
+        (locked_dir / "apm.yml").write_text(yaml.dump({
+            "name": "locked-pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": ["ghcr.io/locked/server"]},
+        }))
+
+        lock_path = tmp_path / "apm.lock"
+        lock_path.write_text(yaml.dump({
+            "lockfile_version": "1",
+            "dependencies": [
+                {"repo_url": "org/locked-pkg", "host": "github.com"},
+            ],
+        }))
+
+        with patch("pathlib.Path.rglob", side_effect=AssertionError("rglob should not be called")):
+            result = _collect_transitive_mcp_deps(apm_modules, lock_path)
+
+        assert result == ["ghcr.io/locked/server"]
+
+    def test_invalid_lockfile_falls_back_to_rglob_scan(self, tmp_path):
+        """If lock parsing fails, function falls back to scanning all apm.yml files."""
+        apm_modules = tmp_path / "apm_modules"
+        pkg_dir = apm_modules / "org" / "pkg-a"
+        pkg_dir.mkdir(parents=True)
+        (pkg_dir / "apm.yml").write_text(yaml.dump({
+            "name": "pkg-a",
+            "version": "1.0.0",
+            "dependencies": {"mcp": ["ghcr.io/a/server"]},
+        }))
+
+        lock_path = tmp_path / "apm.lock"
+        lock_path.write_text("dependencies: [")
+
+        result = _collect_transitive_mcp_deps(apm_modules, lock_path)
+        assert result == ["ghcr.io/a/server"]
+
+
+# ---------------------------------------------------------------------------
+# _deduplicate_mcp_deps
+# ---------------------------------------------------------------------------
+class TestDeduplicateMCPDeps:
+
+    def test_deduplicates_strings(self):
+        deps = ["a", "b", "a", "c", "b"]
+        assert _deduplicate_mcp_deps(deps) == ["a", "b", "c"]
+
+    def test_deduplicates_dicts_by_name(self):
+        d1 = {"name": "srv", "type": "sse", "url": "https://one"}
+        d2 = {"name": "srv", "type": "sse", "url": "https://two"}  # same name
+        d3 = {"name": "other", "type": "sse", "url": "https://three"}
+        result = _deduplicate_mcp_deps([d1, d2, d3])
+        assert len(result) == 2
+        assert result[0]["url"] == "https://one"  # first wins
+
+    def test_mixed_dedup(self):
+        inline = {"name": "kb", "type": "sse", "url": "https://kb"}
+        deps = ["a", inline, "a", {"name": "kb", "type": "sse", "url": "https://kb2"}]
+        result = _deduplicate_mcp_deps(deps)
+        assert len(result) == 2
+        assert isinstance(result[0], str)
+        assert isinstance(result[1], dict)
+
+    def test_empty_list(self):
+        assert _deduplicate_mcp_deps([]) == []
+
+    def test_dict_without_name_kept(self):
+        """Dicts without 'name' are kept if not already in result."""
+        d = {"type": "sse", "url": "https://x"}
+        result = _deduplicate_mcp_deps([d, d])
+        assert len(result) == 1
+
+    def test_root_deps_take_precedence_over_transitive(self):
+        """When root and transitive share a key, the first (root) wins."""
+        root = [{"name": "shared", "type": "sse", "url": "https://root-url"}]
+        transitive = [{"name": "shared", "type": "sse", "url": "https://transitive-url"}]
+        # Root deps come first in the combined list
+        combined = root + transitive
+        result = _deduplicate_mcp_deps(combined)
+        assert len(result) == 1
+        assert result[0]["url"] == "https://root-url"
+
+
+
+# ---------------------------------------------------------------------------
+# _install_mcp_dependencies
+# ---------------------------------------------------------------------------
+class TestInstallMCPDependencies:
+
+    @patch("apm_cli.cli._get_console", return_value=None)
+    @patch("apm_cli.registry.operations.MCPServerOperations")
+    def test_already_configured_registry_servers_not_counted_as_new(
+        self, mock_ops_cls, _console
+    ):
+        mock_ops = mock_ops_cls.return_value
+        mock_ops.validate_servers_exist.return_value = (["ghcr.io/org/server"], [])
+        mock_ops.check_servers_needing_installation.return_value = []
+
+        count = _install_mcp_dependencies(["ghcr.io/org/server"], runtime="vscode")
+
+        assert count == 0
+
+    @patch("apm_cli.cli._install_for_runtime")
+    @patch("apm_cli.cli._get_console", return_value=None)
+    @patch("apm_cli.registry.operations.MCPServerOperations")
+    def test_counts_only_newly_configured_registry_servers(
+        self, mock_ops_cls, _console, mock_install_runtime
+    ):
+        mock_ops = mock_ops_cls.return_value
+        mock_ops.validate_servers_exist.return_value = (
+            ["ghcr.io/org/already", "ghcr.io/org/new"],
+            [],
+        )
+        mock_ops.check_servers_needing_installation.return_value = ["ghcr.io/org/new"]
+        mock_ops.batch_fetch_server_info.return_value = {"ghcr.io/org/new": {}}
+        mock_ops.collect_environment_variables.return_value = {}
+        mock_ops.collect_runtime_variables.return_value = {}
+
+        count = _install_mcp_dependencies(
+            ["ghcr.io/org/already", "ghcr.io/org/new"], runtime="vscode"
+        )
+
+        assert count == 1
+        mock_install_runtime.assert_called_once()
+
+    @patch("apm_cli.cli._install_for_runtime")
+    @patch("apm_cli.registry.operations.MCPServerOperations")
+    def test_mixed_registry_servers_show_already_configured_and_count_only_new(
+        self, mock_ops_cls, mock_install_runtime
+    ):
+        mock_console = MagicMock()
+        mock_ops = mock_ops_cls.return_value
+        mock_ops.validate_servers_exist.return_value = (
+            ["ghcr.io/org/already", "ghcr.io/org/new"],
+            [],
+        )
+        mock_ops.check_servers_needing_installation.return_value = ["ghcr.io/org/new"]
+        mock_ops.batch_fetch_server_info.return_value = {"ghcr.io/org/new": {}}
+        mock_ops.collect_environment_variables.return_value = {}
+        mock_ops.collect_runtime_variables.return_value = {}
+
+        with patch("apm_cli.cli._get_console", return_value=mock_console):
+            count = _install_mcp_dependencies(
+                ["ghcr.io/org/already", "ghcr.io/org/new"], runtime="vscode"
+            )
+
+        assert count == 1
+        mock_install_runtime.assert_called_once()
+        printed_lines = "\n".join(
+            str(call.args[0]) for call in mock_console.print.call_args_list if call.args
+        )
+        assert "ghcr.io/org/already" in printed_lines
+        assert "already configured" in printed_lines