Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion docs/dependencies.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remove # Registry reference qualifier β€” misleading after descoping.

The # Registry reference inline comment implies the existence of a non-registry format. Since inline dict deps were descoped to #132, this qualifier is misleading. Remove the comment or use something neutral like # MCP server.

```

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
Expand Down
1 change: 1 addition & 0 deletions docs/integrations.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
273 changes: 191 additions & 82 deletions src/apm_cli/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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.
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fix return type annotation.

-> list (bare) loses all type information and is inconsistent with the rest of the codebase. Should be -> List[str].

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)
Comment thread
sergio-sisternes-epam marked this conversation as resolved.
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.

Comment thread
sergio-sisternes-epam marked this conversation as resolved.
Args:
deps: Mixed list of str and dict MCP entries.
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Strip dict logic from _deduplicate_mcp_deps β€” descoped to #132.

The seen_names / dict-by-name branch (~lines 2293-2308) handles a type that never enters this function if the model layer filters at parse time (comments #1-#3 above). Simplify to string-only dedup:

def _deduplicate_mcp_deps(deps: List[str]) -> List[str]:
    """Deduplicate MCP dependency strings preserving insertion order."""
    seen: builtins.set = builtins.set()
    result = []
    for dep in deps:
        if dep not in seen:
            seen.add(dep)
            result.append(dep)
    return result

Also: deps: list and -> list should be List[str].


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
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remove registry_deps filter, fix type annotation, clean up terminology.

Three things here:

  1. Type annotation: mcp_deps: list β†’ mcp_deps: List[str]

  2. Dead filter (line ~2327): registry_deps = [d for d in mcp_deps if isinstance(d, str)] is only needed because dicts leak through the pipeline. Once the model layer strips them at parse time (comments Why do we need a GitHub token?Β #1-Will there be MCP coverage?Β #3), all deps are strings. Remove this filter and the # Filter to registry strings only...see #132 comment. Use mcp_deps directly throughout.

  3. "Registry" terminology leak: Four user-facing strings reference "registry servers" / "registry MCP servers" β€” this split terminology only makes sense alongside inline deps, which were descoped. Locations:

    • "Validating {len(registry_deps)} registry servers..." β†’ "Validating {len(mcp_deps)} servers..."
    • "All registry MCP servers already configured" β†’ "All MCP servers already configured"
    • "Already configured registry MCP servers: " β†’ "Already configured MCP servers: "
    • # --- Registry-based deps (strings) --- β†’ # --- Install MCP servers ---


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)]

Comment on lines +2321 to +2324
Copy link

Copilot AI Mar 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_install_mcp_dependencies now accepts mixed MCP entries (str/dict), but immediately filters to registry_deps (strings only) and does not handle dict (inline) entries anywhere else in this module. As a result, projects with only inline MCP dicts (or transitive inline dicts) will report an MCP section but perform no configuration. Either add an inline-dict installation path here, or explicitly reject/skip dict entries with a clear warning so users don’t think they were applied.

This issue also appears on line 2321 of the same file.

Copilot uses AI. Check for mistakes.
console = _get_console()

# Start MCP section with clean header
Expand Down Expand Up @@ -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(
Expand Down
15 changes: 7 additions & 8 deletions src/apm_cli/models/apm_package.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Revert type widening β€” descoped to #132.

This adds dict to the union type but dict MCP deps were explicitly descoped. No consumer in this PR uses dict deps β€” they get parsed, collected, deduplicated, then silently dropped at install time by registry_deps = [d for d in mcp_deps if isinstance(d, str)]. That's worse UX than rejecting at parse time.

Revert to Union[DependencyReference, str].

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)
Expand Down Expand Up @@ -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
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Revert parser change β€” descoped to #132.

isinstance(dep, (str, dict)) preserves dicts that flow through the entire pipeline (collection β†’ dedup β†’ install) only to be filtered out at install time. This creates a silent data loss path: dicts are "collected" and "deduplicated" but never installed, with no warning to the user.

Revert to the original isinstance(dep, str) filter. Comment should remain # Other dependencies (like MCP) remain as strings.

dependencies[dep_type] = [dep for dep in dep_list if isinstance(dep, (str, dict))]

# Parse package content type
pkg_type = None
Expand Down Expand Up @@ -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 []
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Revert return type to List[str]; keep the or [] null guard.

The return type widening to List[Union[str, dict]] and the isinstance(dep, (str, dict)) filter are dict scope creep. The or [] fix for mcp: null is a real bug fix β€” keep that.

Suggested change
return []
def get_mcp_dependencies(self) -> List[str]:
"""Get list of MCP dependencies (as strings for compatibility)."""
if not self.dependencies or 'mcp' not in self.dependencies:
return []
return [dep for dep in (self.dependencies.get('mcp') or [])
if isinstance(dep, str)]

# 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."""
Expand Down
Loading