diff --git a/docs/dependencies.md b/docs/dependencies.md
index f59532a2..88794324 100644
--- a/docs/dependencies.md
+++ b/docs/dependencies.md
@@ -290,7 +290,7 @@ mcp:
 - `url` — required for `http`, `sse`, `streamable-http` transports
 - `command` — required for `stdio` transport
 
-⚠️ **Transitive trust rule:** Self-defined servers from transitive APM packages are skipped with a warning by default. You can either re-declare them in your own `apm.yml`, or use `--trust-transitive-mcp` to trust all self-defined servers from upstream packages:
+⚠️ **Transitive trust rule:** Self-defined servers from direct dependencies (depth=1 in the lockfile) are auto-trusted. Self-defined servers from transitive dependencies (depth > 1) are skipped with a warning by default. You can either re-declare them in your own `apm.yml`, or use `--trust-transitive-mcp` to trust all self-defined servers from upstream packages:
 
 ```bash
 apm install --trust-transitive-mcp
diff --git a/docs/plugins.md b/docs/plugins.md
index 02c0f3ef..2e332e49 100644
--- a/docs/plugins.md
+++ b/docs/plugins.md
@@ -160,6 +160,36 @@ By default APM looks for `agents/`, `skills/`, `commands/`, and `hooks/` directo
 - For **agents**, directory contents are flattened into `.apm/agents/` (agents are flat files, not named directories)
 - `hooks` also accepts an inline object: `"hooks": {"hooks": {"PreToolUse": [...]}}`
 
+#### MCP Server Definitions
+
+Plugins can ship MCP servers that are automatically deployed through APM's MCP pipeline. Define servers using `mcpServers` in `plugin.json`:
+
+```json
+{
+  "name": "my-plugin",
+  "mcpServers": {
+    "my-server": {
+      "command": "npx",
+      "args": ["-y", "my-mcp-server"]
+    },
+    "my-api": {
+      "url": "https://api.example.com/mcp"
+    }
+  }
+}
+```
+
+`mcpServers` supports three forms:
+- **Object** — inline server definitions (as above)
+- **String** — path to a JSON file containing `mcpServers`
+- **Array** — list of JSON file paths (merged, last-wins on name conflicts)
+
+When `mcpServers` is absent, APM auto-discovers `.mcp.json` at the plugin root (then `.github/.mcp.json` as fallback), matching Claude Code's auto-discovery behavior.
+
+Servers with `command` are configured as `stdio` transport; servers with `url` use `http` (or the `type` field if it specifies `sse` or `streamable-http`). All plugin-defined MCP servers are treated as self-defined (`registry: false`).
+
+**Trust model**: Self-defined MCP servers from direct dependencies (depth=1) are auto-trusted. Transitive dependencies require `--trust-transitive-mcp`. See [dependencies.md](./dependencies.md#self-defined-servers) for details.
+
 ## Examples
 
 ### Installing Plugins from GitHub
diff --git a/src/apm_cli/adapters/client/codex.py b/src/apm_cli/adapters/client/codex.py
index 8aa1d264..88b5e97e 100644
--- a/src/apm_cli/adapters/client/codex.py
+++ b/src/apm_cli/adapters/client/codex.py
@@ -173,6 +173,15 @@ def _format_server_config(self, server_info, env_overrides=None, runtime_vars=No
             "env": {},
             "id": server_info.get("id", "")  # Add registry UUID for conflict detection
         }
+
+        # Self-defined stdio deps carry raw command/args — use directly
+        raw = server_info.get("_raw_stdio")
+        if raw:
+            config["command"] = raw["command"]
+            config["args"] = raw["args"]
+            if raw.get("env"):
+                config["env"] = raw["env"]
+            return config
         
         # Note: Remote servers (SSE type) are handled in configure_mcp_server and rejected early
         # This method only handles local servers with packages
diff --git a/src/apm_cli/adapters/client/copilot.py b/src/apm_cli/adapters/client/copilot.py
index d1e8266c..8cbcc73d 100644
--- a/src/apm_cli/adapters/client/copilot.py
+++ b/src/apm_cli/adapters/client/copilot.py
@@ -165,6 +165,19 @@ def _format_server_config(self, server_info, env_overrides=None, runtime_vars=No
             "tools": ["*"],  # Required by Copilot CLI specification - default to all tools
             "id": server_info.get("id", "")  # Add registry UUID for conflict detection
         }
+
+        # Self-defined stdio deps carry raw command/args — use directly
+        raw = server_info.get("_raw_stdio")
+        if raw:
+            config["command"] = raw["command"]
+            config["args"] = raw["args"]
+            if raw.get("env"):
+                config["env"] = raw["env"]
+            # Apply tools override if present
+            tools_override = server_info.get("_apm_tools_override")
+            if tools_override:
+                config["tools"] = tools_override
+            return config
         
         # Check for remote endpoints first (registry-defined priority)
         remotes = server_info.get("remotes", [])
diff --git a/src/apm_cli/adapters/client/vscode.py b/src/apm_cli/adapters/client/vscode.py
index 1a493181..ad722ac6 100644
--- a/src/apm_cli/adapters/client/vscode.py
+++ b/src/apm_cli/adapters/client/vscode.py
@@ -186,6 +186,18 @@ def _format_server_config(self, server_info):
         # Initialize the base config structure
         server_config = {}
         input_vars = []
+
+        # Self-defined stdio deps carry raw command/args — use directly
+        raw = server_info.get("_raw_stdio")
+        if raw:
+            server_config = {
+                "type": "stdio",
+                "command": raw["command"],
+                "args": raw["args"],
+            }
+            if raw.get("env"):
+                server_config["env"] = raw["env"]
+            return server_config, input_vars
         
         # Check for packages information
         if "packages" in server_info and server_info["packages"]:
diff --git a/src/apm_cli/deps/plugin_parser.py b/src/apm_cli/deps/plugin_parser.py
index c411060a..ce47b14c 100644
--- a/src/apm_cli/deps/plugin_parser.py
+++ b/src/apm_cli/deps/plugin_parser.py
@@ -15,7 +15,7 @@
 import logging
 import shutil
 from pathlib import Path
-from typing import Dict, Any, Optional
+from typing import Dict, Any, List, Optional
 import yaml
 
 
@@ -108,6 +108,13 @@ def synthesize_apm_yml_from_plugin(plugin_path: Path, manifest: Dict[str, Any])
     # Map plugin structure into .apm/ subdirectories
     _map_plugin_artifacts(plugin_path, apm_dir, manifest)
 
+    # Extract MCP servers from plugin and convert to dependency format
+    mcp_servers = _extract_mcp_servers(plugin_path, manifest)
+    if mcp_servers:
+        mcp_deps = _mcp_servers_to_apm_deps(mcp_servers, plugin_path)
+        if mcp_deps:
+            manifest['_mcp_deps'] = mcp_deps
+
     # Generate apm.yml from plugin metadata
     apm_yml_content = _generate_apm_yml(manifest)
     apm_yml_path = plugin_path / "apm.yml"
@@ -123,6 +130,182 @@ def _ignore_symlinks(directory, contents):
     return [name for name in contents if (Path(directory) / name).is_symlink()]
 
 
+def _extract_mcp_servers(plugin_path: Path, manifest: Dict[str, Any]) -> Dict[str, Any]:
+    """Extract MCP server definitions from a plugin manifest.
+
+    Resolves ``mcpServers`` by type (per Claude Code spec):
+    - ``str``  → read that file path relative to plugin root, parse JSON,
+      extract ``mcpServers`` key.
+    - ``list`` → read each file path, merge (last-wins on name conflict).
+    - ``dict`` → use directly as inline server definitions.
+
+    When ``mcpServers`` is absent and ``.mcp.json`` (or ``.github/.mcp.json``)
+    exists at plugin root, read it as the default (matches Claude Code
+    auto-discovery).
+
+    Security: symlinks are skipped, JSON parse errors are logged as warnings.
+
+    ``${CLAUDE_PLUGIN_ROOT}`` in string values is replaced with the absolute
+    plugin path.
+
+    Args:
+        plugin_path: Root of the plugin directory.
+        manifest: Parsed plugin.json dict.
+
+    Returns:
+        dict mapping server name → server config.  Empty on failure.
+    """
+    logger = logging.getLogger("apm")
+    mcp_value = manifest.get("mcpServers")
+
+    if mcp_value is not None:
+        # Manifest explicitly defines mcpServers
+        if isinstance(mcp_value, dict):
+            servers = dict(mcp_value)
+        elif isinstance(mcp_value, str):
+            servers = _read_mcp_file(plugin_path, mcp_value, logger)
+        elif isinstance(mcp_value, list):
+            servers = {}
+            for entry in mcp_value:
+                if isinstance(entry, str):
+                    servers.update(_read_mcp_file(plugin_path, entry, logger))
+                else:
+                    logger.warning("Ignoring non-string entry in mcpServers array: %s", entry)
+        else:
+            logger.warning("Unsupported mcpServers type %s; ignoring", type(mcp_value).__name__)
+            return {}
+    else:
+        # Fall back to auto-discovery: .mcp.json then .github/.mcp.json
+        servers = {}
+        for fallback in (".mcp.json", ".github/.mcp.json"):
+            candidate = plugin_path / fallback
+            if candidate.exists() and candidate.is_file() and not candidate.is_symlink():
+                servers = _read_mcp_json(candidate, logger)
+                if servers:
+                    break
+
+    # Substitute ${CLAUDE_PLUGIN_ROOT} in all string values
+    if servers:
+        abs_root = str(plugin_path.resolve())
+        servers = _substitute_plugin_root(servers, abs_root, logger)
+
+    return servers
+
+
+def _read_mcp_file(plugin_path: Path, rel_path: str, logger: logging.Logger) -> Dict[str, Any]:
+    """Read a JSON file relative to *plugin_path* and return its ``mcpServers`` dict."""
+    target = (plugin_path / rel_path).resolve()
+    # Security: must stay inside plugin_path and not be a symlink
+    try:
+        target.relative_to(plugin_path.resolve())
+    except ValueError:
+        logger.warning("MCP file path escapes plugin root: %s", rel_path)
+        return {}
+    candidate = plugin_path / rel_path
+    if not candidate.exists() or not candidate.is_file():
+        logger.warning("MCP file not found: %s", candidate)
+        return {}
+    if candidate.is_symlink():
+        logger.warning("Skipping symlinked MCP file: %s", candidate)
+        return {}
+    return _read_mcp_json(candidate, logger)
+
+
+def _read_mcp_json(path: Path, logger: logging.Logger) -> Dict[str, Any]:
+    """Parse a JSON file and return the ``mcpServers`` mapping."""
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, OSError) as exc:
+        logger.warning("Failed to read MCP config %s: %s", path, exc)
+        return {}
+    if not isinstance(data, dict):
+        return {}
+    servers = data.get("mcpServers", {})
+    return dict(servers) if isinstance(servers, dict) else {}
+
+
+def _substitute_plugin_root(
+    servers: Dict[str, Any], abs_root: str, logger: logging.Logger
+) -> Dict[str, Any]:
+    """Replace ``${CLAUDE_PLUGIN_ROOT}`` in server config string values."""
+    token = "${CLAUDE_PLUGIN_ROOT}"
+    substituted = False
+
+    def _walk(obj: Any) -> Any:
+        nonlocal substituted
+        if isinstance(obj, str) and token in obj:
+            substituted = True
+            return obj.replace(token, abs_root)
+        if isinstance(obj, dict):
+            return {k: _walk(v) for k, v in obj.items()}
+        if isinstance(obj, list):
+            return [_walk(item) for item in obj]
+        return obj
+
+    result = {name: _walk(cfg) for name, cfg in servers.items()}
+    if substituted:
+        logger.info("Substituted ${CLAUDE_PLUGIN_ROOT} with %s", abs_root)
+    return result
+
+
+def _mcp_servers_to_apm_deps(
+    servers: Dict[str, Any], plugin_path: Path
+) -> List[Dict[str, Any]]:
+    """Convert raw MCP server configs to ``dependencies.mcp`` dicts.
+
+    Transport inference:
+    - ``command`` present → stdio
+    - ``url`` present → http (or ``type`` if it's a valid transport)
+    - Neither → skipped with warning
+
+    Every entry gets ``registry: false`` (self-defined, not registry lookups).
+
+    Args:
+        servers: Mapping of server name → server config dict.
+        plugin_path: Plugin root (used for log context only).
+
+    Returns:
+        List of dicts consumable by ``MCPDependency.from_dict()``.
+    """
+    logger = logging.getLogger("apm")
+    deps: List[Dict[str, Any]] = []
+
+    for name, cfg in servers.items():
+        if not isinstance(cfg, dict):
+            logger.warning("Skipping non-dict MCP server config '%s'", name)
+            continue
+
+        dep: Dict[str, Any] = {"name": name, "registry": False}
+
+        if "command" in cfg:
+            dep["transport"] = "stdio"
+            dep["command"] = cfg["command"]
+            if "args" in cfg:
+                dep["args"] = cfg["args"]
+        elif "url" in cfg:
+            raw_type = cfg.get("type", "http")
+            valid_transports = {"http", "sse", "streamable-http"}
+            dep["transport"] = raw_type if raw_type in valid_transports else "http"
+            dep["url"] = cfg["url"]
+            if "headers" in cfg:
+                dep["headers"] = cfg["headers"]
+        else:
+            logger.warning(
+                "Skipping MCP server '%s' from plugin '%s': no 'command' or 'url'",
+                name, plugin_path.name,
+            )
+            continue
+
+        if "env" in cfg:
+            dep["env"] = cfg["env"]
+        if "tools" in cfg:
+            dep["tools"] = cfg["tools"]
+
+        deps.append(dep)
+
+    return deps
+
+
 def _map_plugin_artifacts(plugin_path: Path, apm_dir: Path, manifest: Optional[Dict[str, Any]] = None) -> None:
     """Map plugin artifacts to .apm/ subdirectories and copy pass-through files.
 
@@ -305,6 +488,11 @@ def _generate_apm_yml(manifest: Dict[str, Any]) -> str:
     if manifest.get('dependencies'):
         apm_package['dependencies'] = {'apm': manifest['dependencies']}
 
+    # Inject MCP deps extracted from plugin mcpServers / .mcp.json
+    mcp_deps = manifest.get('_mcp_deps')
+    if mcp_deps:
+        apm_package.setdefault('dependencies', {})['mcp'] = mcp_deps
+
     # Install behavior is driven by file presence (SKILL.md, etc.), not this
     # field.  Default to hybrid so the standard pipeline handles all components.
     apm_package['type'] = 'hybrid'
diff --git a/src/apm_cli/integration/mcp_integrator.py b/src/apm_cli/integration/mcp_integrator.py
index 209596b5..7b9d8879 100644
--- a/src/apm_cli/integration/mcp_integrator.py
+++ b/src/apm_cli/integration/mcp_integrator.py
@@ -55,8 +55,10 @@ def collect_transitive(
         picking up stale/orphaned packages from previous installs.
         Falls back to scanning all apm.yml files if no lock file is available.
 
-        Self-defined servers (registry: false) from transitive packages are
-        skipped with a warning unless *trust_private* is True.
+        Self-defined servers (registry: false) from direct dependencies
+        (depth == 1) are auto-trusted.  Self-defined servers from transitive
+        dependencies (depth > 1) are skipped with a warning unless
+        *trust_private* is True.
         """
         if not apm_modules_dir.exists():
             return []
@@ -65,6 +67,8 @@ def collect_transitive(
 
         # Build set of expected apm.yml paths from apm.lock
         locked_paths = None
+        direct_paths: builtins.set = builtins.set()
+        lockfile = None
         if lock_path and lock_path.exists():
             lockfile = LockFile.read(lock_path)
             if lockfile is not None:
@@ -77,6 +81,8 @@ def collect_transitive(
                             else apm_modules_dir / dep.repo_url / "apm.yml"
                         )
                         locked_paths.add(yml.resolve())
+                        if dep.depth == 1:
+                            direct_paths.add(yml.resolve())
 
         # Prefer iterating lock-derived paths directly (existing files only).
         # Fall back to full scan only when lock parsing is unavailable.
@@ -91,9 +97,15 @@ def collect_transitive(
                 pkg = APMPackage.from_apm_yml(apm_yml_path)
                 mcp = pkg.get_mcp_dependencies()
                 if mcp:
+                    is_direct = apm_yml_path.resolve() in direct_paths
                     for dep in mcp:
                         if hasattr(dep, "is_self_defined") and dep.is_self_defined:
-                            if trust_private:
+                            if is_direct:
+                                _rich_info(
+                                    f"Trusting direct dependency MCP '{dep.name}' "
+                                    f"from '{pkg.name}'"
+                                )
+                            elif trust_private:
                                 _rich_info(
                                     f"Trusting self-defined MCP server '{dep.name}' "
                                     f"from transitive package '{pkg.name}' (--trust-transitive-mcp)"
@@ -157,6 +169,17 @@ def _build_self_defined_info(dep) -> dict:
         """
         info: dict = {"name": dep.name}
 
+        # For stdio self-defined deps, store raw command/args so adapters
+        # can bypass registry-specific formatting (npm, docker, etc.).
+        if dep.transport == "stdio" or (
+            dep.transport not in ("http", "sse", "streamable-http") and dep.command
+        ):
+            info["_raw_stdio"] = {
+                "command": dep.command or dep.name,
+                "args": list(dep.args) if dep.args else [],
+                "env": dict(dep.env) if dep.env else {},
+            }
+
         if dep.transport in ("http", "sse", "streamable-http"):
             # Build as a remote endpoint
             remote = {
@@ -192,7 +215,7 @@ def _build_self_defined_info(dep) -> dict:
                 {
                     "runtime_hint": dep.command or dep.name,
                     "name": dep.name,
-                    "registry_name": "",
+                    "registry_name": "self-defined",
                     "runtime_arguments": runtime_args,
                     "package_arguments": [],
                     "environment_variables": env_vars,
diff --git a/src/apm_cli/integration/skill_integrator.py b/src/apm_cli/integration/skill_integrator.py
index f4b544ff..e97789c2 100644
--- a/src/apm_cli/integration/skill_integrator.py
+++ b/src/apm_cli/integration/skill_integrator.py
@@ -436,7 +436,7 @@ def find_context_files(self, package_path: Path) -> List[Path]:
         return context_files
     
     @staticmethod
-    def _promote_sub_skills(sub_skills_dir: Path, target_skills_root: Path, parent_name: str, *, warn: bool = True) -> tuple[int, list[Path]]:
+    def _promote_sub_skills(sub_skills_dir: Path, target_skills_root: Path, parent_name: str, *, warn: bool = True, owned_by: dict[str, str] | None = None) -> tuple[int, list[Path]]:
         """Promote sub-skills from .apm/skills/ to top-level skill entries.
 
         Args:
@@ -444,6 +444,8 @@ def _promote_sub_skills(sub_skills_dir: Path, target_skills_root: Path, parent_n
             target_skills_root: Root skills directory (e.g. .github/skills/ or .claude/skills/).
             parent_name: Name of the parent skill (used in warning messages).
             warn: Whether to emit a warning on name collisions.
+            owned_by: Map of skill_name → owner_package_name from the lockfile.
+                When provided, warnings are suppressed for self-overwrites.
 
         Returns:
             tuple[int, list[Path]]: (count of promoted sub-skills, list of deployed dir paths)
@@ -462,7 +464,9 @@ def _promote_sub_skills(sub_skills_dir: Path, target_skills_root: Path, parent_n
             sub_name = raw_sub_name if is_valid else normalize_skill_name(raw_sub_name)
             target = target_skills_root / sub_name
             if target.exists():
-                if warn:
+                prev_owner = (owned_by or {}).get(sub_name)
+                is_self_overwrite = prev_owner is not None and prev_owner == parent_name
+                if warn and not is_self_overwrite:
                     try:
                         from apm_cli.cli import _rich_warning
                         _rich_warning(
@@ -477,6 +481,27 @@ def _promote_sub_skills(sub_skills_dir: Path, target_skills_root: Path, parent_n
             deployed.append(target)
         return promoted, deployed
 
+    @staticmethod
+    def _build_skill_ownership_map(project_root: Path) -> dict[str, str]:
+        """Build a map of skill_name → owner_package_name from the lockfile.
+
+        Used to distinguish self-overwrites (no warning) from cross-package
+        conflicts (warning) when promoting sub-skills.
+        """
+        from apm_cli.deps.lockfile import LockFile, get_lockfile_path
+
+        owned_by: dict[str, str] = {}
+        lockfile = LockFile.read(get_lockfile_path(project_root))
+        if not lockfile:
+            return owned_by
+        for dep in lockfile.get_all_dependencies():
+            owner = (dep.virtual_path or dep.repo_url).rsplit("/", 1)[-1]
+            for deployed_path in dep.deployed_files:
+                # e.g. ".github/skills/context-map" → "context-map"
+                skill_name = deployed_path.rstrip("/").rsplit("/", 1)[-1]
+                owned_by[skill_name] = owner
+        return owned_by
+
     def _promote_sub_skills_standalone(
         self, package_info, project_root: Path
     ) -> tuple[int, list[Path]]:
@@ -502,8 +527,9 @@ def _promote_sub_skills_standalone(
         parent_name = package_path.name
         github_skills_root = project_root / ".github" / "skills"
         github_skills_root.mkdir(parents=True, exist_ok=True)
+        owned_by = self._build_skill_ownership_map(project_root)
         count, deployed = self._promote_sub_skills(
-            sub_skills_dir, github_skills_root, parent_name, warn=True
+            sub_skills_dir, github_skills_root, parent_name, warn=True, owned_by=owned_by
         )
         all_deployed = list(deployed)
 
@@ -604,7 +630,8 @@ def _integrate_native_skill(
         # so we promote each sub-skill to an independent top-level entry.
         sub_skills_dir = package_path / ".apm" / "skills"
         github_skills_root = project_root / ".github" / "skills"
-        sub_skills_count, sub_deployed = self._promote_sub_skills(sub_skills_dir, github_skills_root, skill_name, warn=True)
+        owned_by = self._build_skill_ownership_map(project_root)
+        sub_skills_count, sub_deployed = self._promote_sub_skills(sub_skills_dir, github_skills_root, skill_name, warn=True, owned_by=owned_by)
         all_target_paths.extend(sub_deployed)
         
         # === T7: Copy to .claude/skills/ (secondary - compatibility) ===
diff --git a/tests/unit/test_mcp_lifecycle_e2e.py b/tests/unit/test_mcp_lifecycle_e2e.py
index 4dba2eb8..fb2647f1 100644
--- a/tests/unit/test_mcp_lifecycle_e2e.py
+++ b/tests/unit/test_mcp_lifecycle_e2e.py
@@ -547,9 +547,9 @@ def test_virtual_and_non_virtual_together(self, tmp_path):
 # ---------------------------------------------------------------------------
 class TestSelfDefinedMCPTrustGating:
     """Self-defined (non-registry) MCP servers from transitive packages are
-    gated behind trust_private."""
+    gated behind trust_private.  Direct dependencies (depth=1) are auto-trusted."""
 
-    def test_self_defined_skipped_by_default(self, tmp_path):
+    def test_self_defined_skipped_for_transitive(self, tmp_path):
         apm_modules = tmp_path / "apm_modules"
         _make_pkg_dir(apm_modules, "acme/infra-cloud", mcp=[
             "ghcr.io/acme/mcp-registry-server",
@@ -558,7 +558,7 @@ def test_self_defined_skipped_by_default(self, tmp_path):
 
         lock_path = tmp_path / "apm.lock"
         _write_lockfile(lock_path, [
-            LockedDependency(repo_url="acme/infra-cloud", depth=1, resolved_by="root"),
+            LockedDependency(repo_url="acme/infra-cloud", depth=2, resolved_by="some-dep"),
         ])
 
         result = MCPIntegrator.collect_transitive(apm_modules, lock_path, trust_private=False)
@@ -566,6 +566,24 @@ def test_self_defined_skipped_by_default(self, tmp_path):
         assert "ghcr.io/acme/mcp-registry-server" in names
         assert "private-srv" not in names
 
+    def test_direct_dep_self_defined_auto_trusted(self, tmp_path):
+        """Depth=1 packages have their self-defined MCPs auto-trusted."""
+        apm_modules = tmp_path / "apm_modules"
+        _make_pkg_dir(apm_modules, "acme/infra-cloud", mcp=[
+            "ghcr.io/acme/mcp-registry-server",
+            {"name": "private-srv", "registry": False, "transport": "http", "url": "https://private.example.com"},
+        ])
+
+        lock_path = tmp_path / "apm.lock"
+        _write_lockfile(lock_path, [
+            LockedDependency(repo_url="acme/infra-cloud", depth=1, resolved_by="root"),
+        ])
+
+        result = MCPIntegrator.collect_transitive(apm_modules, lock_path, trust_private=False)
+        names = [d.name for d in result]
+        assert "ghcr.io/acme/mcp-registry-server" in names
+        assert "private-srv" in names
+
     def test_self_defined_included_when_trusted(self, tmp_path):
         apm_modules = tmp_path / "apm_modules"
         _make_pkg_dir(apm_modules, "acme/infra-cloud", mcp=[
diff --git a/tests/unit/test_plugin_parser.py b/tests/unit/test_plugin_parser.py
index c1fdd4a5..eaa3aa4a 100644
--- a/tests/unit/test_plugin_parser.py
+++ b/tests/unit/test_plugin_parser.py
@@ -8,8 +8,10 @@
 import yaml
 
 from apm_cli.deps.plugin_parser import (
+    _extract_mcp_servers,
     _generate_apm_yml,
     _map_plugin_artifacts,
+    _mcp_servers_to_apm_deps,
     normalize_plugin_directory,
     parse_plugin_manifest,
     synthesize_apm_yml_from_plugin,
@@ -541,3 +543,298 @@ def test_validate_readme_only(self, tmp_path):
         (plugin_dir / "README.md").write_text("# Hello")
 
         assert validate_plugin_package(plugin_dir) is False
+
+
+class TestExtractMCPServers:
+    """Tests for _extract_mcp_servers() — Phase 1, Step 1."""
+
+    def test_mcpservers_inline_object(self, tmp_path):
+        """Dict in manifest → extracted directly."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        manifest = {
+            "name": "test",
+            "mcpServers": {
+                "my-server": {"command": "npx", "args": ["-y", "my-server"]},
+            },
+        }
+        result = _extract_mcp_servers(plugin_dir, manifest)
+        assert "my-server" in result
+        assert result["my-server"]["command"] == "npx"
+
+    def test_mcpservers_string_path(self, tmp_path):
+        """File path → reads file, extracts servers."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        mcp_data = {"mcpServers": {"file-srv": {"command": "node", "args": ["index.js"]}}}
+        (plugin_dir / "mcp-config.json").write_text(json.dumps(mcp_data))
+        manifest = {"name": "test", "mcpServers": "mcp-config.json"}
+
+        result = _extract_mcp_servers(plugin_dir, manifest)
+        assert "file-srv" in result
+        assert result["file-srv"]["command"] == "node"
+
+    def test_mcpservers_array_paths(self, tmp_path):
+        """Multiple file paths → merges, last-wins."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        file1 = {"mcpServers": {"srv-a": {"command": "a"}, "srv-b": {"command": "b1"}}}
+        file2 = {"mcpServers": {"srv-b": {"command": "b2"}, "srv-c": {"command": "c"}}}
+        (plugin_dir / "mcp1.json").write_text(json.dumps(file1))
+        (plugin_dir / "mcp2.json").write_text(json.dumps(file2))
+        manifest = {"name": "test", "mcpServers": ["mcp1.json", "mcp2.json"]}
+
+        result = _extract_mcp_servers(plugin_dir, manifest)
+        assert result["srv-a"]["command"] == "a"
+        assert result["srv-b"]["command"] == "b2"  # last-wins
+        assert result["srv-c"]["command"] == "c"
+
+    def test_default_mcp_json(self, tmp_path):
+        """No mcpServers field, but .mcp.json exists → auto-discovered."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        mcp_data = {"mcpServers": {"default-srv": {"command": "echo"}}}
+        (plugin_dir / ".mcp.json").write_text(json.dumps(mcp_data))
+        manifest = {"name": "test"}
+
+        result = _extract_mcp_servers(plugin_dir, manifest)
+        assert "default-srv" in result
+
+    def test_github_mcp_json_fallback(self, tmp_path):
+        """No .mcp.json but .github/.mcp.json → discovered."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        gh_dir = plugin_dir / ".github"
+        gh_dir.mkdir()
+        mcp_data = {"mcpServers": {"gh-srv": {"url": "https://example.com"}}}
+        (gh_dir / ".mcp.json").write_text(json.dumps(mcp_data))
+        manifest = {"name": "test"}
+
+        result = _extract_mcp_servers(plugin_dir, manifest)
+        assert "gh-srv" in result
+
+    def test_manifest_wins_over_default(self, tmp_path):
+        """mcpServers field takes precedence over .mcp.json file."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        # .mcp.json has different server
+        mcp_data = {"mcpServers": {"file-srv": {"command": "from-file"}}}
+        (plugin_dir / ".mcp.json").write_text(json.dumps(mcp_data))
+        manifest = {
+            "name": "test",
+            "mcpServers": {"inline-srv": {"command": "from-manifest"}},
+        }
+
+        result = _extract_mcp_servers(plugin_dir, manifest)
+        assert "inline-srv" in result
+        assert "file-srv" not in result
+
+    def test_missing_file_graceful(self, tmp_path):
+        """String path pointing to nonexistent file → empty dict, warning."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        manifest = {"name": "test", "mcpServers": "does-not-exist.json"}
+
+        result = _extract_mcp_servers(plugin_dir, manifest)
+        assert result == {}
+
+    def test_symlink_skipped(self, tmp_path):
+        """Symlinked file → skipped."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        external = tmp_path / "external.json"
+        external.write_text(json.dumps({"mcpServers": {"evil": {"command": "evil"}}}))
+        link = plugin_dir / "mcp.json"
+        try:
+            link.symlink_to(external)
+        except OSError:
+            pytest.skip("Symlinks not supported on this platform")
+        manifest = {"name": "test", "mcpServers": "mcp.json"}
+
+        result = _extract_mcp_servers(plugin_dir, manifest)
+        assert result == {}
+
+    def test_empty_manifest(self, tmp_path):
+        """No mcpServers and no .mcp.json → empty dict."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        manifest = {"name": "test"}
+
+        result = _extract_mcp_servers(plugin_dir, manifest)
+        assert result == {}
+
+    def test_plugin_root_substitution(self, tmp_path):
+        """${CLAUDE_PLUGIN_ROOT} replaced with absolute plugin path."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        manifest = {
+            "name": "test",
+            "mcpServers": {
+                "local-srv": {
+                    "command": "node",
+                    "args": ["${CLAUDE_PLUGIN_ROOT}/server.js"],
+                },
+            },
+        }
+
+        result = _extract_mcp_servers(plugin_dir, manifest)
+        abs_root = str(plugin_dir.resolve())
+        assert result["local-srv"]["args"] == [f"{abs_root}/server.js"]
+
+
+class TestMCPServersToDeps:
+    """Tests for _mcp_servers_to_apm_deps() — Phase 1, Step 2."""
+
+    def test_stdio_server(self, tmp_path):
+        """command present → transport=stdio, registry=false."""
+        servers = {"my-srv": {"command": "npx", "args": ["-y", "my-server"]}}
+        deps = _mcp_servers_to_apm_deps(servers, tmp_path)
+        assert len(deps) == 1
+        assert deps[0]["name"] == "my-srv"
+        assert deps[0]["transport"] == "stdio"
+        assert deps[0]["registry"] is False
+        assert deps[0]["command"] == "npx"
+        assert deps[0]["args"] == ["-y", "my-server"]
+
+    def test_http_server(self, tmp_path):
+        """url present → transport=http, registry=false."""
+        servers = {"web-srv": {"url": "https://example.com/mcp"}}
+        deps = _mcp_servers_to_apm_deps(servers, tmp_path)
+        assert len(deps) == 1
+        assert deps[0]["name"] == "web-srv"
+        assert deps[0]["transport"] == "http"
+        assert deps[0]["registry"] is False
+        assert deps[0]["url"] == "https://example.com/mcp"
+
+    def test_mixed_servers(self, tmp_path):
+        """Both stdio and http in one config."""
+        servers = {
+            "stdio-srv": {"command": "node", "args": ["index.js"]},
+            "http-srv": {"url": "https://example.com"},
+        }
+        deps = _mcp_servers_to_apm_deps(servers, tmp_path)
+        assert len(deps) == 2
+        names = {d["name"] for d in deps}
+        assert names == {"stdio-srv", "http-srv"}
+
+    def test_env_and_args_passthrough(self, tmp_path):
+        """env and args are passed through."""
+        servers = {
+            "srv": {
+                "command": "cmd",
+                "args": ["--flag"],
+                "env": {"KEY": "VAL"},
+            }
+        }
+        deps = _mcp_servers_to_apm_deps(servers, tmp_path)
+        assert deps[0]["env"] == {"KEY": "VAL"}
+        assert deps[0]["args"] == ["--flag"]
+
+    def test_invalid_server_skipped(self, tmp_path):
+        """No command or url → skipped."""
+        servers = {"bad-srv": {"env": {"KEY": "VAL"}}}
+        deps = _mcp_servers_to_apm_deps(servers, tmp_path)
+        assert len(deps) == 0
+
+    def test_sse_type_preserved(self, tmp_path):
+        """type field with valid transport is used."""
+        servers = {"sse-srv": {"url": "https://sse.example.com", "type": "sse"}}
+        deps = _mcp_servers_to_apm_deps(servers, tmp_path)
+        assert deps[0]["transport"] == "sse"
+
+    def test_tools_passthrough(self, tmp_path):
+        """tools field is passed through."""
+        servers = {"srv": {"command": "cmd", "tools": ["tool1", "tool2"]}}
+        deps = _mcp_servers_to_apm_deps(servers, tmp_path)
+        assert deps[0]["tools"] == ["tool1", "tool2"]
+
+    def test_headers_passthrough(self, tmp_path):
+        """headers field is passed through for http servers."""
+        servers = {
+            "srv": {
+                "url": "https://example.com",
+                "headers": {"Authorization": "Bearer token"},
+            }
+        }
+        deps = _mcp_servers_to_apm_deps(servers, tmp_path)
+        assert deps[0]["headers"] == {"Authorization": "Bearer token"}
+
+
+class TestGenerateApmYmlMCPDeps:
+    """Test _mcp_deps injection in generated apm.yml."""
+
+    def test_mcp_deps_in_generated_yml(self):
+        """_mcp_deps in manifest → dependencies.mcp in output."""
+        manifest = {
+            "name": "mcp-plugin",
+            "_mcp_deps": [
+                {"name": "my-srv", "registry": False, "transport": "stdio", "command": "echo"},
+            ],
+        }
+        yml_str = _generate_apm_yml(manifest)
+        parsed = yaml.safe_load(yml_str)
+        assert "mcp" in parsed["dependencies"]
+        assert len(parsed["dependencies"]["mcp"]) == 1
+        assert parsed["dependencies"]["mcp"][0]["name"] == "my-srv"
+
+    def test_mcp_deps_with_apm_deps(self):
+        """Both apm and mcp deps coexist."""
+        manifest = {
+            "name": "both-plugin",
+            "dependencies": {"dep-a": "^1.0"},
+            "_mcp_deps": [
+                {"name": "srv", "registry": False, "transport": "http", "url": "https://x"},
+            ],
+        }
+        yml_str = _generate_apm_yml(manifest)
+        parsed = yaml.safe_load(yml_str)
+        assert "apm" in parsed["dependencies"]
+        assert "mcp" in parsed["dependencies"]
+
+    def test_no_mcp_deps_no_section(self):
+        """No _mcp_deps → no mcp key in dependencies."""
+        manifest = {"name": "no-mcp"}
+        yml_str = _generate_apm_yml(manifest)
+        parsed = yaml.safe_load(yml_str)
+        assert "dependencies" not in parsed
+
+
+class TestSynthesizeMCPIntegration:
+    """End-to-end test: synthesize_apm_yml_from_plugin with MCP servers."""
+
+    def test_synthesize_with_mcp_json(self, tmp_path):
+        """Plugin with .mcp.json produces apm.yml with dependencies.mcp."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        mcp_data = {"mcpServers": {"test-srv": {"command": "echo", "args": ["hello"]}}}
+        (plugin_dir / ".mcp.json").write_text(json.dumps(mcp_data))
+
+        apm_yml = synthesize_apm_yml_from_plugin(plugin_dir, {"name": "test-plugin"})
+        parsed = yaml.safe_load(apm_yml.read_text())
+
+        assert "dependencies" in parsed
+        assert "mcp" in parsed["dependencies"]
+        mcp_deps = parsed["dependencies"]["mcp"]
+        assert len(mcp_deps) == 1
+        assert mcp_deps[0]["name"] == "test-srv"
+        assert mcp_deps[0]["transport"] == "stdio"
+        assert mcp_deps[0]["registry"] is False
+
+    def test_synthesize_with_inline_mcpservers(self, tmp_path):
+        """Plugin with inline mcpServers in manifest."""
+        plugin_dir = tmp_path / "plugin"
+        plugin_dir.mkdir()
+        manifest = {
+            "name": "inline-mcp",
+            "mcpServers": {
+                "web-srv": {"url": "https://api.example.com"},
+            },
+        }
+
+        apm_yml = synthesize_apm_yml_from_plugin(plugin_dir, manifest)
+        parsed = yaml.safe_load(apm_yml.read_text())
+
+        mcp_deps = parsed["dependencies"]["mcp"]
+        assert len(mcp_deps) == 1
+        assert mcp_deps[0]["name"] == "web-srv"
+        assert mcp_deps[0]["transport"] == "http"
diff --git a/tests/unit/test_transitive_mcp.py b/tests/unit/test_transitive_mcp.py
index b3528a26..dea4bb9f 100644
--- a/tests/unit/test_transitive_mcp.py
+++ b/tests/unit/test_transitive_mcp.py
@@ -309,6 +309,95 @@ def test_trust_private_false_is_default_behavior(self, tmp_path):
         result = MCPIntegrator.collect_transitive(tmp_path, trust_private=False)
         assert len(result) == 0
 
+    def test_direct_dep_self_defined_auto_trusted(self, tmp_path):
+        """Depth=1 package with self-defined MCP → collected without flag."""
+        apm_modules = tmp_path / "apm_modules"
+        pkg_dir = apm_modules / "org" / "direct-pkg"
+        pkg_dir.mkdir(parents=True)
+        (pkg_dir / "apm.yml").write_text(yaml.dump({
+            "name": "direct-pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": [
+                {"name": "private-srv", "registry": False,
+                 "transport": "http", "url": "https://private.example.com"},
+            ]},
+        }))
+        lock_path = tmp_path / "apm.lock"
+        lock_path.write_text(yaml.dump({
+            "lockfile_version": "1",
+            "dependencies": [
+                {"repo_url": "org/direct-pkg", "host": "github.com", "depth": 1},
+            ],
+        }))
+        result = MCPIntegrator.collect_transitive(apm_modules, lock_path)
+        assert len(result) == 1
+        assert result[0].name == "private-srv"
+
+    def test_transitive_dep_self_defined_still_skipped(self, tmp_path):
+        """Depth=2 package with self-defined MCP → skipped without flag."""
+        apm_modules = tmp_path / "apm_modules"
+        pkg_dir = apm_modules / "org" / "transitive-pkg"
+        pkg_dir.mkdir(parents=True)
+        (pkg_dir / "apm.yml").write_text(yaml.dump({
+            "name": "transitive-pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": [
+                {"name": "private-srv", "registry": False,
+                 "transport": "http", "url": "https://private.example.com"},
+            ]},
+        }))
+        lock_path = tmp_path / "apm.lock"
+        lock_path.write_text(yaml.dump({
+            "lockfile_version": "1",
+            "dependencies": [
+                {"repo_url": "org/transitive-pkg", "host": "github.com", "depth": 2},
+            ],
+        }))
+        result = MCPIntegrator.collect_transitive(apm_modules, lock_path)
+        assert len(result) == 0
+
+    def test_transitive_dep_trusted_with_flag(self, tmp_path):
+        """Depth=2 + trust_private=True → collected."""
+        apm_modules = tmp_path / "apm_modules"
+        pkg_dir = apm_modules / "org" / "transitive-pkg"
+        pkg_dir.mkdir(parents=True)
+        (pkg_dir / "apm.yml").write_text(yaml.dump({
+            "name": "transitive-pkg",
+            "version": "1.0.0",
+            "dependencies": {"mcp": [
+                {"name": "private-srv", "registry": False,
+                 "transport": "http", "url": "https://private.example.com"},
+            ]},
+        }))
+        lock_path = tmp_path / "apm.lock"
+        lock_path.write_text(yaml.dump({
+            "lockfile_version": "1",
+            "dependencies": [
+                {"repo_url": "org/transitive-pkg", "host": "github.com", "depth": 2},
+            ],
+        }))
+        result = MCPIntegrator.collect_transitive(apm_modules, lock_path, trust_private=True)
+        assert len(result) == 1
+        assert result[0].name == "private-srv"
+
+    def test_no_lockfile_conservative(self, tmp_path):
+        """No lockfile → all self-defined skipped (conservative)."""
+        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/registry/server",
+                {"name": "private-srv", "registry": False,
+                 "transport": "http", "url": "https://private.example.com"},
+            ]},
+        }))
+        # No lock_path provided
+        result = MCPIntegrator.collect_transitive(tmp_path)
+        assert len(result) == 1
+        assert result[0].name == "ghcr.io/registry/server"
+
 
 # ---------------------------------------------------------------------------
 # _deduplicate_mcp_deps
diff --git a/uv.lock b/uv.lock
index ebfd331e..72a12718 100644
--- a/uv.lock
+++ b/uv.lock
@@ -200,7 +200,7 @@ wheels = [
 
 [[package]]
 name = "apm-cli"
-version = "0.7.4"
+version = "0.7.5"
 source = { editable = "." }
 dependencies = [
     { name = "click", version = "8.1.8", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" },