diff --git a/changelog/1117.added.md b/changelog/1117.added.md
new file mode 100644
index 00000000..b68b9f0f
--- /dev/null
+++ b/changelog/1117.added.md
@@ -0,0 +1 @@
+Added a `--dependencies` flag to `infrahubctl marketplace get`. When downloading a schema or a collection, it now also resolves and downloads the schemas they depend on, via the marketplace API. Dependencies are grouped by the collection they belong to: prerequisite collections (and, for a single schema, dependencies that are members of a collection) are placed in their own `<collection>/` directory, while dependencies that belong to no collection land in the output root. Resolution is transitive and cycle-safe, and referenced kinds the marketplace cannot resolve are reported as unresolved dependencies. A schema that already exists in the output directory is reconciled to a single file rather than duplicated across directories — kept by default, or overwritten with the new `-y`/`--yes` flag.
diff --git a/docs/docs/infrahubctl/infrahubctl-marketplace.mdx b/docs/docs/infrahubctl/infrahubctl-marketplace.mdx
index 4b9ebc5f..41b3e20e 100644
--- a/docs/docs/infrahubctl/infrahubctl-marketplace.mdx
+++ b/docs/docs/infrahubctl/infrahubctl-marketplace.mdx
@@ -39,6 +39,8 @@ $ infrahubctl marketplace get [OPTIONS] IDENTIFIER
 
 * `-v, --version TEXT`: Specific schema version, for example 1.2.0. Default: latest published.
 * `-c, --collection`: Force collection download. Default: auto-detect whether the identifier is a schema or collection.
+* `--dependencies`: Also download the schemas this schema or collection depends on.
+* `-y, --yes`: Overwrite schemas that already exist in the output directory without prompting.
 * `-s, --stdout`: Print content to stdout instead of writing to disk. Status messages go to stderr.
 * `-o, --output-dir PATH`: Directory to save downloaded files.  [default: schemas]
 * `--marketplace-url TEXT`: Base URL of the Infrahub Marketplace. Overrides configuration and environment.
diff --git a/infrahub_sdk/ctl/marketplace.py b/infrahub_sdk/ctl/marketplace.py
index 1b12e279..fd651d5c 100644
--- a/infrahub_sdk/ctl/marketplace.py
+++ b/infrahub_sdk/ctl/marketplace.py
@@ -2,6 +2,7 @@
 
 import asyncio
 import sys
+from collections import Counter, deque
 from enum import Enum
 from pathlib import Path
 from typing import Any, Literal, NamedTuple, NoReturn
@@ -74,6 +75,10 @@ def _collection_url(base_url: str, namespace: str, name: str) -> str:
     return f"{base_url}/api/v1/collections/{namespace}/{name}"
 
 
+def _schema_detail_url(base_url: str, namespace: str, name: str) -> str:
+    return f"{base_url}/api/v1/schemas/{namespace}/{name}"
+
+
 def _is_transport_failure(r: object) -> bool:
     if isinstance(r, Exception):
         return True
@@ -87,6 +92,53 @@ def _mkdir_or_fail(path: Path) -> None:
         _fail(_ErrorClass.INVALID_INPUT, f"Cannot write to '{path}': {exc}")
 
 
+class _WriteContext(NamedTuple):
+    """Controls overwriting when a schema already exists in the output tree.
+
+    ``preexisting`` maps a schema filename (``<name>.yml``) to the path where it was found in
+    the output directory *before* this command ran, so a schema being written now that already
+    exists elsewhere (e.g. a dependency at the root vs. a copy under a collection directory) is
+    reconciled to a single file instead of duplicated. Only files present before the run are
+    considered, so schemas written during this same run never trigger a prompt.
+    """
+
+    assume_yes: bool
+    preexisting: dict[str, Path]
+
+
+def _snapshot_existing_schemas(output_root: Path) -> dict[str, Path]:
+    """Map ``<name>.yml`` -> existing path for every schema already under ``output_root``."""
+    existing: dict[str, Path] = {}
+    if output_root.exists():
+        for path in sorted(output_root.rglob("*.yml")):
+            if path.is_file():
+                existing.setdefault(path.name, path)
+    return existing
+
+
+def _print_unresolved(status: Console, unresolved: set[str] | list[str]) -> None:
+    """Report referenced kinds the marketplace could not resolve to a schema, if any."""
+    if unresolved:
+        status.print(
+            "[yellow]Unresolved dependencies (referenced kinds the marketplace could not resolve to a schema): "
+            + ", ".join(sorted(unresolved))
+        )
+
+
+def _confirm_overwrite(prompt: str, *, assume_yes: bool) -> bool:
+    """Return whether to overwrite an existing schema file.
+
+    ``--yes`` overwrites unconditionally; an interactive terminal is prompted; a
+    non-interactive run without ``--yes`` declines (keep the existing file) so scripts and CI
+    never block or clobber.
+    """
+    if assume_yes:
+        return True
+    if not sys.stdin.isatty():
+        return False
+    return typer.confirm(prompt, default=False)
+
+
 def _make_http_client(sdk_cfg: _SdkConfig) -> httpx.AsyncClient:
     """Build an httpx client that inherits the SDK's proxy and TLS configuration."""
     proxy_kwargs: dict[str, Any] = {}
@@ -157,9 +209,14 @@ async def _download_schema(
     prefetched: httpx.Response | None = None,
     schema_confirmed_exists: bool = False,
     needs_separator: bool = False,
-) -> None:
+    soft_fail: bool = False,
+    write_ctx: _WriteContext | None = None,
+) -> bool:
     """Download a single schema and write it to disk or stdout.
 
+    Returns ``True`` when the schema was written/streamed and ``False`` when it was skipped
+    (only possible with ``soft_fail``).
+
     When ``prefetched`` is supplied and ``version`` is None, reuses the response
     instead of re-fetching the unversioned download URL.
     ``schema_confirmed_exists`` signals that the schema is known to exist (e.g. from
@@ -168,13 +225,36 @@ async def _download_schema(
     ``needs_separator`` inserts a ``---`` document separator before the content in
     stdout mode when it is missing, so multiple schemas streamed back-to-back (e.g.
     from a collection) form a valid multi-document YAML stream.
+    ``soft_fail`` downgrades a 404 to an informational note and a ``False`` return instead of
+    aborting — used for resolved dependencies so one missing dependency does not fail the
+    whole download. ``write_ctx`` reconciles against schemas already present on disk: an
+    already-present schema is overwritten (``--yes``/prompt) or kept, decided before fetching
+    so a kept schema costs no download.
     """
+    filename = f"{name}.yml"
+
+    # Reconcile with a pre-existing copy before fetching (disk mode only).
+    existing_path = write_ctx.preexisting.get(filename) if write_ctx is not None and not stdout else None
+    if existing_path is not None and not _confirm_overwrite(
+        f"{namespace}/{name} already exists at {existing_path}. Overwrite?",
+        assume_yes=write_ctx.assume_yes if write_ctx else False,
+    ):
+        _status_console(stdout).print(
+            f"[yellow]Kept existing {existing_path}; skipped {namespace}/{name} (pass --yes to overwrite)."
+        )
+        return False
+
     if prefetched is not None and version is None:
         resp = prefetched
     else:
         resp = await client.get(_schema_url(base_url, namespace, name, version=version))
 
     if resp.status_code == 404:
+        if soft_fail:
+            _status_console(stdout).print(
+                f"[yellow]Note: dependency {namespace}/{name} could not be downloaded (not found); skipping."
+            )
+            return False
         if version and schema_confirmed_exists:
             _fail(
                 _ErrorClass.NOT_FOUND,
@@ -196,14 +276,475 @@ async def _download_schema(
         if not resp.text.endswith("\n"):
             sys.stdout.write("\n")
         err_console.print(f"[green]Fetched schema {namespace}/{name} v{resolved_version}")
-        return
+        return True
+
+    if existing_path is not None:
+        existing_path.write_text(resp.text, encoding="utf-8")
+        console.print(f"[green]Updated schema {namespace}/{name} v{resolved_version} -> {existing_path}")
+        return True
 
-    filename = f"{name}.yml"
     _mkdir_or_fail(output_dir)
     file_path = output_dir / filename
     file_path.write_text(resp.text, encoding="utf-8")
 
     console.print(f"[green]Downloaded schema {namespace}/{name} v{resolved_version} -> {file_path}")
+    return True
+
+
+def _collection_members(payload: Any, status: Console) -> list[dict[str, Any]]:
+    """Extract downloadable members from a collection metadata payload.
+
+    Returns a list of ``{"namespace", "name", "version"}`` dicts. Members missing a namespace
+    or name are skipped with a warning rather than aborting the download.
+    """
+    items = payload.get("items", []) if isinstance(payload, dict) else []
+    schemas = [item.get("schema") for item in items if isinstance(item, dict)]
+    members: list[dict[str, Any]] = []
+    for schema in schemas:
+        if not isinstance(schema, dict):
+            continue
+        member_namespace = schema.get("namespace")
+        member_name = schema.get("name")
+        if not member_namespace or not member_name:
+            status.print("[yellow]Warning: skipping a collection member with missing namespace or name.")
+            continue
+        version = (schema.get("latest_version") or {}).get("semver")
+        members.append({"namespace": member_namespace, "name": member_name, "version": version})
+    return members
+
+
+async def _read_schema_dependencies(
+    client: httpx.AsyncClient,
+    base_url: str,
+    namespace: str,
+    name: str,
+    *,
+    status: Console,
+) -> tuple[list[tuple[str, str, str]], list[str]]:
+    """Read a schema's latest-version dependencies from the marketplace API.
+
+    Returns ``(resolved, unresolved_kinds)`` where ``resolved`` is a list of
+    ``(namespace, name, schema_id)`` for dependencies the marketplace can supply, and
+    ``unresolved_kinds`` are referenced kinds that are not available (including ones hidden by
+    visibility). A read failure is reported as an informational note and treated as "no
+    dependencies" so the rest of the resolution continues.
+    """
+    try:
+        resp = await client.get(_schema_detail_url(base_url, namespace, name))
+        resp.raise_for_status()
+        payload = resp.json()
+    except (httpx.HTTPError, ValueError) as exc:
+        detail = str(exc) or type(exc).__name__
+        status.print(f"[yellow]Note: could not read dependencies for {namespace}/{name}: {detail}")
+        return [], []
+
+    versions = (payload.get("versions") or []) if isinstance(payload, dict) else []
+    latest_id = (payload.get("latest_version") or {}).get("id") if isinstance(payload, dict) else None
+    deps_source = next((v for v in versions if isinstance(v, dict) and v.get("id") == latest_id), None) or (
+        versions[0] if versions and isinstance(versions[0], dict) else {}
+    )
+
+    resolved: list[tuple[str, str, str]] = []
+    unresolved: list[str] = []
+    for dep in deps_source.get("dependencies") or []:
+        if not isinstance(dep, dict):
+            continue
+        resolved_schema = dep.get("resolved_schema")
+        if dep.get("is_resolved") and isinstance(resolved_schema, dict):
+            dep_namespace = resolved_schema.get("namespace")
+            dep_name = resolved_schema.get("name")
+            if dep_namespace and dep_name:
+                resolved.append((dep_namespace, dep_name, resolved_schema.get("id") or ""))
+                continue
+        referenced_kind = dep.get("referenced_kind")
+        if referenced_kind:
+            unresolved.append(referenced_kind)
+    return resolved, unresolved
+
+
+async def _resolve_dependency_closure(
+    client: httpx.AsyncClient,
+    base_url: str,
+    members: list[dict[str, Any]],
+    *,
+    status: Console,
+) -> tuple[list[dict[str, Any]], list[str]]:
+    """Walk schema dependencies transitively, starting from the collection's members.
+
+    Returns ``(schemas, unresolved_kinds)`` where ``schemas`` is the full ordered download set
+    (members first, then discovered dependencies, each downloaded at its latest version) and
+    ``unresolved_kinds`` is the sorted set of referenced kinds not available in the
+    marketplace. A ``seen`` set of ``(namespace, name)`` makes the walk cycle-safe and ensures
+    each schema appears once even when reachable through multiple paths.
+    """
+    seen: set[tuple[str, str]] = set()
+    ordered: list[dict[str, Any]] = []
+    unresolved: set[str] = set()
+    queue: deque[dict[str, Any]] = deque()
+
+    for member in members:
+        key = (member["namespace"], member["name"])
+        if key in seen:
+            continue
+        seen.add(key)
+        ordered.append(member)
+        queue.append(member)
+
+    while queue:
+        current = queue.popleft()
+        resolved, kinds = await _read_schema_dependencies(
+            client, base_url, current["namespace"], current["name"], status=status
+        )
+        unresolved.update(kinds)
+        for dep_namespace, dep_name, dep_id in resolved:
+            key = (dep_namespace, dep_name)
+            if key in seen:
+                continue
+            seen.add(key)
+            entry = {"namespace": dep_namespace, "name": dep_name, "version": None, "schema_id": dep_id}
+            ordered.append(entry)
+            queue.append(entry)
+
+    return ordered, sorted(unresolved)
+
+
+async def _download_schema_set(
+    client: httpx.AsyncClient,
+    base_url: str,
+    schemas: list[dict[str, Any]],
+    target_dir: Path,
+    *,
+    stdout: bool,
+    seen: set[tuple[str, str]] | None = None,
+    already_written: int = 0,
+    soft_fail: bool = False,
+    reserved_names: set[str] | None = None,
+    write_ctx: _WriteContext | None = None,
+) -> int:
+    """Download a resolved set of schemas into ``target_dir``, returning the count written.
+
+    Schemas sharing a name across namespaces are disambiguated into per-namespace
+    subdirectories so they do not overwrite each other. ``soft_fail`` downgrades a missing
+    schema to a note instead of aborting — used for loose schema dependencies so one missing
+    dependency does not fail the whole download (collection members download strictly).
+    ``seen`` deduplicates ``(namespace, name)`` across multiple calls so a schema already
+    downloaded (e.g. as a member of another collection) is skipped. ``already_written`` is the
+    running total written by prior calls, used so the ``---`` stdout separator is inserted
+    before every document except the very first across the whole download. ``reserved_names``
+    are names already written into ``target_dir`` by a prior call (e.g. the requested schema),
+    so a pending schema sharing one is disambiguated into a namespace subdirectory rather than
+    overwriting it.
+    """
+    if seen is None:
+        seen = set()
+    pending = []
+    for schema in schemas:
+        key = (schema["namespace"], schema["name"])
+        if key in seen:
+            continue
+        seen.add(key)
+        pending.append(schema)
+
+    name_counts = Counter(schema["name"] for schema in pending)
+    for reserved in reserved_names or ():
+        name_counts[reserved] += 1
+
+    written_here = 0
+    for schema in pending:
+        member_name = schema["name"]
+        member_dir = target_dir / schema["namespace"] if name_counts[member_name] > 1 else target_dir
+        written = await _download_schema(
+            client=client,
+            base_url=base_url,
+            namespace=schema["namespace"],
+            name=member_name,
+            version=schema.get("version"),
+            output_dir=member_dir,
+            stdout=stdout,
+            schema_confirmed_exists=True,
+            needs_separator=already_written + written_here > 0,
+            soft_fail=soft_fail,
+            write_ctx=write_ctx,
+        )
+        if written:
+            written_here += 1
+    return written_here
+
+
+def _collection_dependency_targets(
+    payload: Any,
+) -> tuple[list[tuple[str, str]], list[dict[str, Any]], list[str]]:
+    """Extract a collection's declared dependencies from its detail payload.
+
+    Returns ``(prerequisite_collections, standalone_schemas, unresolved_kinds)``: prerequisite
+    collections as ``(namespace, name)`` tuples, standalone schemas as member-shaped dependency
+    dicts, and unresolved kinds the marketplace could not resolve to a schema.
+    """
+    dep = (payload.get("dependencies") or {}) if isinstance(payload, dict) else {}
+    collections: list[tuple[str, str]] = [
+        (str(entry["namespace"]), str(entry["name"]))
+        for entry in dep.get("collections") or []
+        if isinstance(entry, dict) and entry.get("namespace") and entry.get("name")
+    ]
+    schemas: list[dict[str, Any]] = [
+        {"namespace": str(entry["namespace"]), "name": str(entry["name"]), "version": None}
+        for entry in dep.get("schemas") or []
+        if isinstance(entry, dict) and entry.get("namespace") and entry.get("name")
+    ]
+    unresolved: list[str] = [str(kind) for kind in dep.get("unresolved_kinds") or [] if kind]
+    return collections, schemas, unresolved
+
+
+async def _fetch_collection_payload(
+    client: httpx.AsyncClient,
+    base_url: str,
+    namespace: str,
+    name: str,
+    *,
+    status: Console,
+) -> Any | None:
+    """Fetch a prerequisite collection's detail, returning None (with a note) on any failure.
+
+    Prerequisite collections are dependencies, so an unreachable one is reported and skipped
+    rather than aborting the whole download.
+    """
+    try:
+        resp = await client.get(_collection_url(base_url, namespace, name))
+        if resp.status_code == 404:
+            status.print(f"[yellow]Note: prerequisite collection {namespace}/{name} not found; skipping.")
+            return None
+        resp.raise_for_status()
+        return resp.json()
+    except (httpx.HTTPError, ValueError) as exc:
+        detail = str(exc) or type(exc).__name__
+        status.print(f"[yellow]Note: could not fetch prerequisite collection {namespace}/{name}: {detail}")
+        return None
+
+
+_CollectionTargets = tuple[list[tuple[str, str]], list[dict[str, Any]], list[str]]
+_CollectionRecord = tuple[str, str, list[dict[str, Any]], _CollectionTargets]
+
+
+async def _walk_collection_graph(
+    client: httpx.AsyncClient,
+    base_url: str,
+    namespace: str,
+    name: str,
+    payload: Any,
+    *,
+    status: Console,
+) -> list[_CollectionRecord]:
+    """Walk the collection dependency graph cycle-safe, one record per collection.
+
+    Each record is ``(namespace, name, members, dependency_targets)``. Prerequisite collections
+    from ``dependencies.collections`` are visited transitively; ``seen`` prevents revisiting a
+    collection in a cycle.
+    """
+    seen: set[tuple[str, str]] = {(namespace, name)}
+    records: list[_CollectionRecord] = []
+    queue: deque[tuple[str, str, Any]] = deque([(namespace, name, payload)])
+    while queue:
+        current_namespace, current_name, current_payload = queue.popleft()
+        if current_payload is None:
+            current_payload = await _fetch_collection_payload(
+                client, base_url, current_namespace, current_name, status=status
+            )
+            if current_payload is None:
+                continue
+        members = _collection_members(current_payload, status)
+        targets = _collection_dependency_targets(current_payload)
+        records.append((current_namespace, current_name, members, targets))
+        for dep_namespace, dep_name in targets[0]:
+            if (dep_namespace, dep_name) not in seen:
+                seen.add((dep_namespace, dep_name))
+                queue.append((dep_namespace, dep_name, None))
+    return records
+
+
+def _report_collection_tree(
+    status: Console,
+    namespace: str,
+    name: str,
+    total_written: int,
+    requested_member_count: int,
+    prerequisites: list[str],
+    unresolved: set[str],
+) -> None:
+    dependency_count = total_written - requested_member_count
+    noun = "dependency" if dependency_count == 1 else "dependencies"
+    status.print(
+        f"\n[green]Collection {namespace}/{name}: {total_written} schemas downloaded "
+        f"({dependency_count} {noun} resolved)"
+    )
+    if prerequisites:
+        status.print("[green]Prerequisite collections: " + ", ".join(prerequisites))
+    _print_unresolved(status, unresolved)
+
+
+async def _download_collection_tree(
+    client: httpx.AsyncClient,
+    base_url: str,
+    namespace: str,
+    name: str,
+    payload: Any,
+    output_dir: Path,
+    *,
+    stdout: bool,
+    write_ctx: _WriteContext | None = None,
+) -> None:
+    """Download a collection together with its dependencies, grouped by source collection.
+
+    Layout:
+
+    - requested collection members -> ``output_dir/<requested name>/``
+    - each transitive prerequisite collection -> ``output_dir/<collection name>/``
+    - standalone dependency schemas (not part of any prerequisite collection) -> ``output_dir/``
+
+    ``seen_schemas`` ensures each schema is written once even when reachable through several
+    collections or paths; standalone schemas are resolved transitively via the per-schema walk.
+    """
+    status = _status_console(stdout)
+    records = await _walk_collection_graph(client, base_url, namespace, name, payload, status=status)
+
+    seen_schemas: set[tuple[str, str]] = set()
+    unresolved: set[str] = set()
+    standalone_seed: list[dict[str, Any]] = []
+    prerequisites: list[str] = []
+    total_written = 0
+    requested_member_count = 0
+
+    for rec_namespace, rec_name, members, targets in records:
+        if (rec_namespace, rec_name) == (namespace, name):
+            requested_member_count = len(members)
+        else:
+            prerequisites.append(f"{rec_namespace}/{rec_name}")
+        standalone_seed.extend(targets[1])
+        unresolved.update(targets[2])
+        # Collection members (requested or prerequisite) download strictly: a curated
+        # collection that lists a missing member is an error, not something to skip.
+        total_written += await _download_schema_set(
+            client,
+            base_url,
+            members,
+            output_dir / rec_name,
+            stdout=stdout,
+            seen=seen_schemas,
+            already_written=total_written,
+            soft_fail=False,
+            write_ctx=write_ctx,
+        )
+
+    # Loose schema dependencies (standalone + transitively discovered) soft-fail: a referenced
+    # schema that cannot be retrieved is reported and skipped, not fatal (FR-014).
+    standalone, standalone_unresolved = await _resolve_dependency_closure(
+        client, base_url, standalone_seed, status=status
+    )
+    unresolved.update(standalone_unresolved)
+    total_written += await _download_schema_set(
+        client,
+        base_url,
+        standalone,
+        output_dir,
+        stdout=stdout,
+        seen=seen_schemas,
+        already_written=total_written,
+        soft_fail=True,
+        write_ctx=write_ctx,
+    )
+
+    _report_collection_tree(status, namespace, name, total_written, requested_member_count, prerequisites, unresolved)
+
+
+async def _owning_collection_name(
+    client: httpx.AsyncClient,
+    base_url: str,
+    schema_id: str,
+) -> str | None:
+    """Return the name of a collection the schema belongs to, if any (first by name).
+
+    Used to group a schema's dependencies under their collection's directory, mirroring a
+    collection download. A lookup failure is treated as "no collection" (write to the root).
+    """
+    if not schema_id:
+        return None
+    try:
+        resp = await client.get(f"{base_url}/api/v1/collections/for-schema/{schema_id}")
+        resp.raise_for_status()
+        items = resp.json().get("items") or []
+    except (httpx.HTTPError, ValueError, AttributeError):
+        return None
+    names = sorted(entry["name"] for entry in items if isinstance(entry, dict) and entry.get("name"))
+    return names[0] if names else None
+
+
+async def _download_schema_tree(
+    client: httpx.AsyncClient,
+    base_url: str,
+    namespace: str,
+    name: str,
+    version: str | None,
+    output_dir: Path,
+    *,
+    stdout: bool,
+    prefetched: httpx.Response | None = None,
+    schema_confirmed_exists: bool = False,
+    write_ctx: _WriteContext | None = None,
+) -> None:
+    """Download a single schema together with its transitive dependencies.
+
+    The requested schema downloads strictly (it is the primary target) to the output root. Its
+    transitively resolved dependency schemas soft-fail and are grouped by the collection they
+    belong to (``output_dir/<collection>/``, like a collection download); dependencies not in
+    any collection go to the output root. Referenced kinds that cannot be resolved are reported.
+    """
+    status = _status_console(stdout)
+    requested_written = await _download_schema(
+        client=client,
+        base_url=base_url,
+        namespace=namespace,
+        name=name,
+        version=version,
+        output_dir=output_dir,
+        stdout=stdout,
+        prefetched=prefetched,
+        schema_confirmed_exists=schema_confirmed_exists,
+        write_ctx=write_ctx,
+    )
+    total_written = 1 if requested_written else 0
+
+    # Resolve the transitive closure from the requested schema (it is already downloaded, so
+    # ``seen`` skips it), then bucket each dependency under its owning collection's directory.
+    seen: set[tuple[str, str]] = {(namespace, name)}
+    seed = [{"namespace": namespace, "name": name, "version": version}]
+    closure, unresolved = await _resolve_dependency_closure(client, base_url, seed, status=status)
+
+    buckets: dict[Path, list[dict[str, Any]]] = {}
+    for dep in closure:
+        if (dep["namespace"], dep["name"]) == (namespace, name):
+            continue
+        owning = await _owning_collection_name(client, base_url, dep.get("schema_id", ""))
+        buckets.setdefault(output_dir / owning if owning else output_dir, []).append(dep)
+
+    for target, group in buckets.items():
+        total_written += await _download_schema_set(
+            client,
+            base_url,
+            group,
+            target,
+            stdout=stdout,
+            seen=seen,
+            already_written=total_written,
+            soft_fail=True,
+            reserved_names={name} if target == output_dir and requested_written else None,
+            write_ctx=write_ctx,
+        )
+
+    dependency_count = total_written - (1 if requested_written else 0)
+    noun = "dependency" if dependency_count == 1 else "dependencies"
+    status.print(
+        f"\n[green]Schema {namespace}/{name}: {total_written} schemas downloaded ({dependency_count} {noun} resolved)"
+    )
+    _print_unresolved(status, unresolved)
 
 
 async def _download_collection(
@@ -215,6 +756,8 @@ async def _download_collection(
     *,
     stdout: bool,
     prefetched: httpx.Response | None = None,
+    with_dependencies: bool = False,
+    write_ctx: _WriteContext | None = None,
 ) -> None:
     """Fetch every schema in a collection, writing to disk or stdout.
 
@@ -247,37 +790,16 @@ async def _download_collection(
             f"Response from {_collection_url(base_url, namespace, name)} is not valid JSON",
         )
 
-    items = payload.get("items", []) if isinstance(payload, dict) else []
-    schemas = [item.get("schema") for item in items if isinstance(item, dict)]
-    members: list[dict[str, Any]] = [schema for schema in schemas if isinstance(schema, dict)]
     status = _status_console(stdout)
-    target_dir = output_dir / name
 
-    member_names = [schema.get("name") for schema in members if schema.get("namespace") and schema.get("name")]
-    duplicated_names = {member_name for member_name in member_names if member_names.count(member_name) > 1}
-
-    downloaded = 0
-    for schema in members:
-        member_namespace = schema.get("namespace")
-        member_name = schema.get("name")
-        if not member_namespace or not member_name:
-            status.print("[yellow]Warning: skipping a collection member with missing namespace or name.")
-            continue
-        version = (schema.get("latest_version") or {}).get("semver")
-        member_dir = target_dir / member_namespace if member_name in duplicated_names else target_dir
-        await _download_schema(
-            client=client,
-            base_url=base_url,
-            namespace=member_namespace,
-            name=member_name,
-            version=version,
-            output_dir=member_dir,
-            stdout=stdout,
-            schema_confirmed_exists=True,
-            needs_separator=downloaded > 0,
+    if with_dependencies:
+        await _download_collection_tree(
+            client, base_url, namespace, name, payload, output_dir, stdout=stdout, write_ctx=write_ctx
         )
-        downloaded += 1
+        return
 
+    members = _collection_members(payload, status)
+    downloaded = await _download_schema_set(client, base_url, members, output_dir / name, stdout=stdout)
     status.print(f"\n[green]Collection {namespace}/{name}: {downloaded} schemas downloaded")
 
 
@@ -295,6 +817,17 @@ async def get(
         is_flag=True,
         help="Force collection download. Default: auto-detect whether the identifier is a schema or collection.",
     ),
+    dependencies: bool = typer.Option(
+        False,
+        "--dependencies",
+        help="Also download the schemas this schema or collection depends on.",
+    ),
+    yes: bool = typer.Option(
+        False,
+        "--yes",
+        "-y",
+        help="Overwrite schemas that already exist in the output directory without prompting.",
+    ),
     stdout: bool = typer.Option(
         False,
         "--stdout",
@@ -321,6 +854,12 @@ async def get(
     sdk_cfg = _SdkConfig()
     resolved_url = (marketplace_url or SETTINGS.active.marketplace_url).rstrip("/")
 
+    # When resolving dependencies to disk, reconcile against schemas already present so a
+    # dependency is not duplicated across directories; overwriting is gated by --yes/prompt.
+    write_ctx: _WriteContext | None = None
+    if dependencies and not stdout:
+        write_ctx = _WriteContext(assume_yes=yes, preexisting=_snapshot_existing_schemas(output_dir))
+
     async with _make_http_client(sdk_cfg) as client:
         prefetched: httpx.Response | None = None
         schema_confirmed_exists = False
@@ -346,6 +885,21 @@ async def get(
                     output_dir=output_dir,
                     stdout=stdout,
                     prefetched=prefetched,
+                    with_dependencies=dependencies,
+                    write_ctx=write_ctx,
+                )
+            elif dependencies:
+                await _download_schema_tree(
+                    client=client,
+                    base_url=resolved_url,
+                    namespace=namespace,
+                    name=name,
+                    version=version,
+                    output_dir=output_dir,
+                    stdout=stdout,
+                    prefetched=prefetched,
+                    schema_confirmed_exists=schema_confirmed_exists,
+                    write_ctx=write_ctx,
                 )
             else:
                 await _download_schema(
diff --git a/tests/unit/ctl/test_marketplace_app.py b/tests/unit/ctl/test_marketplace_app.py
index 4f302a51..f3da93ba 100644
--- a/tests/unit/ctl/test_marketplace_app.py
+++ b/tests/unit/ctl/test_marketplace_app.py
@@ -19,17 +19,77 @@
 """
 
 
-def _collection_json(members: list[tuple[str, str, str]]) -> dict:
+def _collection_json(members: list[tuple[str, str, str]], dependencies: dict | None = None) -> dict:
     """Build collection metadata mimicking the marketplace endpoint.
 
-    ``members`` is a list of ``(namespace, name, semver)`` tuples.
+    ``members`` is a list of ``(namespace, name, semver)`` tuples. ``dependencies`` is the
+    optional derived-dependency object the detail endpoint returns.
     """
-    return {
+    payload: dict = {
         "items": [
             {"schema": {"namespace": ns, "name": name, "latest_version": {"semver": semver}}}
             for ns, name, semver in members
         ]
     }
+    if dependencies is not None:
+        payload["dependencies"] = dependencies
+    return payload
+
+
+def _deps(
+    *,
+    schemas: list[tuple[str, str]] | None = None,
+    collections: list[tuple[str, str]] | None = None,
+    unresolved: list[str] | None = None,
+) -> dict:
+    """Build a collection ``dependencies`` object (prerequisite collections + standalone schemas)."""
+    return {
+        "schemas": [{"id": f"s-{ns}-{nm}", "namespace": ns, "name": nm} for ns, nm in (schemas or [])],
+        "collections": [{"id": f"c-{ns}-{nm}", "namespace": ns, "name": nm} for ns, nm in (collections or [])],
+        "unresolved_kinds": unresolved or [],
+        "hidden_count": 0,
+    }
+
+
+def _resolved_dep(namespace: str, name: str, kind: str | None = None) -> dict:
+    return {
+        "referenced_kind": kind or f"{namespace.capitalize()}{name.capitalize()}",
+        "resolved_schema": {"id": f"s-{namespace}-{name}", "namespace": namespace, "name": name},
+        "is_resolved": True,
+        "multi_resolved": False,
+        "hidden_due_to_visibility": False,
+    }
+
+
+def _unresolved_dep(kind: str, *, hidden: bool = False) -> dict:
+    return {
+        "referenced_kind": kind,
+        "resolved_schema": None,
+        "is_resolved": False,
+        "multi_resolved": False,
+        "hidden_due_to_visibility": hidden,
+    }
+
+
+def _schema_detail(namespace: str, name: str, *, semver: str = "1.0.0", deps: list[dict] | None = None) -> dict:
+    """Build a schema-detail response mimicking GET /api/v1/schemas/{ns}/{name}."""
+    version_id = f"v-{namespace}-{name}"
+    return {
+        "namespace": namespace,
+        "name": name,
+        "latest_version": {"id": version_id, "semver": semver},
+        "versions": [{"id": version_id, "semver": semver, "dependencies": deps or []}],
+    }
+
+
+def _stub_for_schema(httpx_mock: HTTPXMock, ns: str, name: str, collection: str | None = None) -> None:
+    """Stub GET /collections/for-schema/{id} for a dependency (id matches _resolved_dep)."""
+    items = [{"namespace": "infrahub", "name": collection}] if collection else []
+    httpx_mock.add_response(
+        method="GET",
+        url=f"https://marketplace.infrahub.app/api/v1/collections/for-schema/s-{ns}-{name}",
+        json={"items": items},
+    )
 
 
 def test_download_schema_specific_version(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
@@ -589,6 +649,7 @@ async def test_collection_false_autodetects_schema(httpx_mock: HTTPXMock, tmp_pa
         identifier="acme/network-base",
         version=None,
         collection=False,
+        dependencies=False,
         stdout=False,
         output_dir=tmp_path,
         marketplace_url="https://marketplace.infrahub.app",
@@ -596,3 +657,478 @@ async def test_collection_false_autodetects_schema(httpx_mock: HTTPXMock, tmp_pa
     )
 
     assert (tmp_path / "network-base.yml").read_text() == SCHEMA_YAML
+
+
+def test_dependencies_groups_prerequisite_collections_and_standalone(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """C1: members, prerequisite collections, and transitive standalone schemas land in the right dirs."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/starter-pack",
+        json=_collection_json(
+            [("acme", "app", "1.0.0")],
+            _deps(collections=[("acme", "base")], schemas=[("acme", "extra")]),
+        ),
+    )
+    # Prerequisite collection → its own directory.
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/base",
+        json=_collection_json([("acme", "dcim", "1.0.0"), ("acme", "ipam", "2.0.0")]),
+    )
+    # Standalone schema walk: extra → more (transitive), to the output root.
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/extra",
+        json=_schema_detail("acme", "extra", deps=[_resolved_dep("acme", "more")]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/more",
+        json=_schema_detail("acme", "more", deps=[]),
+    )
+    for url in (
+        "schemas/acme/app/versions/1.0.0/download",
+        "schemas/acme/dcim/versions/1.0.0/download",
+        "schemas/acme/ipam/versions/2.0.0/download",
+        "schemas/acme/extra/download",
+        "schemas/acme/more/download",
+    ):
+        httpx_mock.add_response(method="GET", url=f"https://marketplace.infrahub.app/api/v1/{url}", text=SCHEMA_YAML)
+    result = runner.invoke(app, ["get", "acme/starter-pack", "-c", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "5 schemas downloaded" in result.output
+    assert "4 dependencies resolved" in result.output
+    assert "Prerequisite collections: acme/base" in result.output
+    # Requested collection members in its own dir.
+    assert (tmp_path / "starter-pack" / "app.yml").exists()
+    # Prerequisite collection members in their collection's dir.
+    assert (tmp_path / "base" / "dcim.yml").exists()
+    assert (tmp_path / "base" / "ipam.yml").exists()
+    # Standalone dependency schemas at the output root.
+    assert (tmp_path / "extra.yml").exists()
+    assert (tmp_path / "more.yml").exists()
+
+
+def test_dependencies_not_requested_skips_resolution(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """C2: without --dependencies, declared dependencies are ignored (backward compatible)."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/starter-pack",
+        json=_collection_json([("acme", "app", "1.0.0")], _deps(collections=[("acme", "base")])),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/app/versions/1.0.0/download",
+        text=SCHEMA_YAML,
+    )
+    # No /collections/acme/base mock — pytest-httpx fails if dependency resolution runs.
+    result = runner.invoke(app, ["get", "acme/starter-pack", "-c", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "1 schemas downloaded" in result.output
+    assert "dependencies resolved" not in result.output
+    assert (tmp_path / "starter-pack" / "app.yml").exists()
+
+
+def test_dependencies_collection_cycle_safe(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """A collection cycle A→B→A resolves each collection once."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/a",
+        json=_collection_json([("acme", "sa", "1.0.0")], _deps(collections=[("acme", "b")])),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/b",
+        json=_collection_json([("acme", "sb", "1.0.0")], _deps(collections=[("acme", "a")])),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/sa/versions/1.0.0/download",
+        text=SCHEMA_YAML,
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/sb/versions/1.0.0/download",
+        text=SCHEMA_YAML,
+    )
+    result = runner.invoke(app, ["get", "acme/a", "-c", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "2 schemas downloaded" in result.output
+    assert "1 dependency resolved" in result.output
+    assert "Prerequisite collections: acme/b" in result.output
+    assert (tmp_path / "a" / "sa.yml").exists()
+    assert (tmp_path / "b" / "sb.yml").exists()
+
+
+def test_dependencies_member_also_standalone_downloaded_once(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """A schema that is both a member and a standalone dependency is written once (as a member)."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/pair",
+        json=_collection_json([("acme", "a", "1.0.0"), ("acme", "b", "2.0.0")], _deps(schemas=[("acme", "b")])),
+    )
+    # The standalone walk reads b's dependencies but b is already a member, so it is not re-downloaded.
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/b",
+        json=_schema_detail("acme", "b", deps=[]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/a/versions/1.0.0/download",
+        text=SCHEMA_YAML,
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/b/versions/2.0.0/download",
+        text=SCHEMA_YAML,
+    )
+    result = runner.invoke(app, ["get", "acme/pair", "-c", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "2 schemas downloaded" in result.output
+    assert "0 dependencies resolved" in result.output
+    assert (tmp_path / "pair" / "b.yml").exists()
+
+
+def test_dependencies_stdout_streams_all_documents(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """C4: --dependencies --stdout streams members, prerequisite-collection, and standalone schemas; no files."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/starter-pack",
+        json=_collection_json(
+            [("acme", "a", "1.0.0")],
+            _deps(collections=[("acme", "base")], schemas=[("acme", "ext")]),
+        ),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/base",
+        json=_collection_json([("acme", "bb", "1.0.0")]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/ext",
+        json=_schema_detail("acme", "ext", deps=[]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/a/versions/1.0.0/download",
+        text=SCHEMA_YAML,
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/bb/versions/1.0.0/download",
+        text=SCHEMA_YAML,
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/ext/download",
+        text=SCHEMA_YAML,
+    )
+    result = runner.invoke(app, ["get", "acme/starter-pack", "-c", "--dependencies", "--stdout", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert result.output.count(SCHEMA_YAML) == 3
+    assert "Fetched schema acme/a v1.0.0" in result.output
+    assert "Fetched schema acme/bb v1.0.0" in result.output
+    assert "Fetched schema acme/ext" in result.output
+    assert "2 dependencies resolved" in result.output
+    assert not any(tmp_path.iterdir())
+
+
+def test_dependencies_unresolved_kinds_reported(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """C6: referenced kinds the marketplace cannot resolve are listed as unresolved dependencies."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/starter-pack",
+        json=_collection_json([("acme", "app", "1.0.0")], _deps(unresolved=["BuiltinTag", "BuiltinIPHost"])),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/app/versions/1.0.0/download",
+        text=SCHEMA_YAML,
+    )
+    result = runner.invoke(app, ["get", "acme/starter-pack", "-c", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "0 dependencies resolved" in result.output
+    assert "Unresolved dependencies" in result.output
+    assert "BuiltinTag" in result.output
+    assert "BuiltinIPHost" in result.output
+
+
+def test_dependencies_missing_standalone_download_is_soft(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """C7: a standalone dependency that 404s on download is skipped with a note; the rest succeed."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/needy",
+        json=_collection_json([("acme", "app", "1.0.0")], _deps(schemas=[("acme", "gone")])),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/gone",
+        json=_schema_detail("acme", "gone", deps=[]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/app/versions/1.0.0/download",
+        text=SCHEMA_YAML,
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/gone/download",
+        status_code=404,
+        json={"detail": "Schema not found"},
+    )
+    result = runner.invoke(app, ["get", "acme/needy", "-c", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "dependency acme/gone could not be downloaded" in result.output
+    assert "1 schemas downloaded" in result.output
+    assert (tmp_path / "needy" / "app.yml").exists()
+    assert not (tmp_path / "gone.yml").exists()
+
+
+def test_dependencies_missing_prerequisite_member_fails_strictly(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """A prerequisite collection that lists a member with no published version is a hard error."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/app-pack",
+        json=_collection_json([("acme", "app", "1.0.0")], _deps(collections=[("acme", "base")])),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/base",
+        json=_collection_json([("acme", "missing", "9.9.9")]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/app/versions/1.0.0/download",
+        text=SCHEMA_YAML,
+    )
+    # The prerequisite collection's member cannot be fetched at its pinned version.
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/missing/versions/9.9.9/download",
+        status_code=404,
+        json={"detail": "Version not found"},
+    )
+    result = runner.invoke(app, ["get", "acme/app-pack", "-c", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 1
+    assert "acme/missing" in result.output
+
+
+def test_dependencies_on_schema_resolves_transitively(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """--dependencies on a single schema downloads it plus its transitive dependency schemas."""
+    # Auto-detect: identifier resolves to a schema.
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/app/download",
+        text=SCHEMA_YAML,
+        headers={"x-schema-version": "1.0.0"},
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/app",
+        status_code=404,
+        json={"detail": "Collection not found"},
+    )
+    # Dependency walk: app → ipam → location.
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/app",
+        json=_schema_detail("acme", "app", deps=[_resolved_dep("acme", "ipam")]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/ipam",
+        json=_schema_detail("acme", "ipam", deps=[_resolved_dep("acme", "location")]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/location",
+        json=_schema_detail("acme", "location", deps=[]),
+    )
+    # Dependency downloads (latest, unversioned) land in the output root next to the schema.
+    for dep in ("ipam", "location"):
+        httpx_mock.add_response(
+            method="GET",
+            url=f"https://marketplace.infrahub.app/api/v1/schemas/acme/{dep}/download",
+            text=SCHEMA_YAML,
+        )
+    # Neither dependency belongs to a collection → both stay at the output root.
+    _stub_for_schema(httpx_mock, "acme", "ipam")
+    _stub_for_schema(httpx_mock, "acme", "location")
+    result = runner.invoke(app, ["get", "acme/app", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "3 schemas downloaded" in result.output
+    assert "2 dependencies resolved" in result.output
+    assert (tmp_path / "app.yml").exists()
+    assert (tmp_path / "ipam.yml").exists()
+    assert (tmp_path / "location.yml").exists()
+
+
+def test_dependencies_on_schema_groups_deps_by_collection(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """A dependency that belongs to a collection is placed under that collection's directory."""
+    _stub_schema_get(httpx_mock, "acme", "app", deps=[_resolved_dep("acme", "dcim")])
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/dcim",
+        json=_schema_detail("acme", "dcim", deps=[]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/dcim/download",
+        text=SCHEMA_YAML,
+    )
+    # dcim belongs to the base-schemas collection → grouped under base-schemas/.
+    _stub_for_schema(httpx_mock, "acme", "dcim", collection="base-schemas")
+    result = runner.invoke(app, ["get", "acme/app", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "1 dependency resolved" in result.output
+    assert (tmp_path / "app.yml").exists()  # requested schema at the root
+    assert (tmp_path / "base-schemas" / "dcim.yml").exists()  # dependency grouped by collection
+    assert not (tmp_path / "dcim.yml").exists()
+
+
+def test_dependencies_on_schema_with_no_dependencies(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """--dependencies on a leaf schema downloads just the schema and reports zero dependencies."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/leaf/download",
+        text=SCHEMA_YAML,
+        headers={"x-schema-version": "1.0.0"},
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/leaf",
+        status_code=404,
+        json={"detail": "Collection not found"},
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/leaf",
+        json=_schema_detail("acme", "leaf", deps=[]),
+    )
+    result = runner.invoke(app, ["get", "acme/leaf", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "1 schemas downloaded" in result.output
+    assert "0 dependencies resolved" in result.output
+    assert (tmp_path / "leaf.yml").exists()
+
+
+def test_dependencies_on_schema_disambiguates_same_name_dependency(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """A dependency sharing the requested schema's name (different namespace) does not overwrite it."""
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/thing/download",
+        text=SCHEMA_YAML,
+        headers={"x-schema-version": "1.0.0"},
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/collections/acme/thing",
+        status_code=404,
+        json={"detail": "Collection not found"},
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/thing",
+        json=_schema_detail("acme", "thing", deps=[_resolved_dep("other", "thing")]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/other/thing",
+        json=_schema_detail("other", "thing", deps=[]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/other/thing/download",
+        text=SCHEMA_YAML,
+    )
+    _stub_for_schema(httpx_mock, "other", "thing")  # not in a collection → stays at root
+    result = runner.invoke(app, ["get", "acme/thing", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "2 schemas downloaded" in result.output
+    # Requested schema at the root; the same-named dependency disambiguated into its namespace.
+    assert (tmp_path / "thing.yml").exists()
+    assert (tmp_path / "other" / "thing.yml").exists()
+
+
+def _stub_schema_get(httpx_mock: HTTPXMock, ns: str, name: str, *, deps: list[dict] | None = None) -> None:
+    """Register the auto-detect (download + collection-404) and dependency-read stubs for a schema."""
+    httpx_mock.add_response(
+        method="GET",
+        url=f"https://marketplace.infrahub.app/api/v1/schemas/{ns}/{name}/download",
+        text=SCHEMA_YAML,
+        headers={"x-schema-version": "1.0.0"},
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url=f"https://marketplace.infrahub.app/api/v1/collections/{ns}/{name}",
+        status_code=404,
+        json={"detail": "Collection not found"},
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url=f"https://marketplace.infrahub.app/api/v1/schemas/{ns}/{name}",
+        json=_schema_detail(ns, name, deps=deps or []),
+    )
+
+
+def test_dependencies_existing_file_kept_without_yes(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """A schema already present in the output tree is kept (not overwritten/duplicated) without --yes."""
+    # Pre-existing copy from a prior download, in a collection subdirectory.
+    (tmp_path / "base-schemas").mkdir()
+    existing = tmp_path / "base-schemas" / "dcim.yml"
+    existing.write_text("OLD CONTENT\n")
+
+    _stub_schema_get(httpx_mock, "acme", "app", deps=[_resolved_dep("acme", "dcim")])
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/dcim",
+        json=_schema_detail("acme", "dcim", deps=[]),
+    )
+    _stub_for_schema(httpx_mock, "acme", "dcim", collection="base-schemas")
+    # Non-interactive (CliRunner stdin is not a tty) and no --yes → dcim is kept, not re-fetched.
+    result = runner.invoke(app, ["get", "acme/app", "--dependencies", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "Kept existing" in result.output
+    assert existing.read_text() == "OLD CONTENT\n"  # untouched
+    assert not (tmp_path / "dcim.yml").exists()  # no duplicate created at the root
+
+
+def test_dependencies_existing_file_overwritten_with_yes(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
+    """With --yes, an existing schema is overwritten in place rather than duplicated."""
+    (tmp_path / "base-schemas").mkdir()
+    existing = tmp_path / "base-schemas" / "dcim.yml"
+    existing.write_text("OLD CONTENT\n")
+
+    _stub_schema_get(httpx_mock, "acme", "app", deps=[_resolved_dep("acme", "dcim")])
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/dcim",
+        json=_schema_detail("acme", "dcim", deps=[]),
+    )
+    httpx_mock.add_response(
+        method="GET",
+        url="https://marketplace.infrahub.app/api/v1/schemas/acme/dcim/download",
+        text=SCHEMA_YAML,
+    )
+    _stub_for_schema(httpx_mock, "acme", "dcim", collection="base-schemas")
+    result = runner.invoke(app, ["get", "acme/app", "--dependencies", "--yes", "-o", str(tmp_path)])
+
+    assert result.exit_code == 0
+    assert "Updated schema acme/dcim" in result.output
+    assert existing.read_text() == SCHEMA_YAML  # overwritten in place
+    assert not (tmp_path / "dcim.yml").exists()  # still no duplicate at the root