feat(workflows): expose workflow catalog sources through MCP

docs/overview/core.md

@@ -175,7 +175,8 @@ specify mcp
 
 Starts the experimental stdio-only MCP server. Its first-class read-only tools
 expose version information (`specify_version`), artifact inventory and composition
-(`specify_artifact_list`, `specify_artifact_info`), and installed workflows
-(`specify_workflow_list`). The generic list, describe, and run tools support only
-the stable `version` JSON command. See the [MCP Server reference](mcp.md) for the
+(`specify_artifact_list`, `specify_artifact_info`), installed workflows
+(`specify_workflow_list`), and workflow catalog sources
+(`specify_workflow_catalog_list`). The generic list, describe, and run tools
+support only the stable `version` JSON command. See the [MCP Server reference](mcp.md) for the
 tool names, result contracts, and current limitations.

docs/overview/mcp.md

@@ -26,6 +26,7 @@ read-only, local-only (no network access), and reject unexpected arguments.
 | `specify_artifact_list` | Return one paginated page of a project's artifact inventory (`command`, `template`, `script`, `hook`) |
 | `specify_artifact_info` | Return one artifact and its complete composition stack |
 | `specify_workflow_list` | List installed workflows and skipped-entry warnings in registry order |
+| `specify_workflow_catalog_list` | List active workflow catalog sources in priority order |
 
 The artifact tools take an optional absolute `project_directory`; it defaults
 to the directory the server was launched from.
@@ -93,6 +94,35 @@ Operation outcomes exceeding the requested entry limit, or claiming truncation
 without filling that limit, are rejected as `invalid_operation_result`. The
 output schema also caps `items` at 1000.
 
+## Workflow catalog list
+
+`specify_workflow_catalog_list` accepts an optional absolute
+`project_directory` (default: the server launch directory) and `limit` (1–1000,
+default 100). It resolves the same catalog stack as
+`specify workflow catalog list`: the `SPECKIT_WORKFLOW_CATALOG_URL` override, the
+project's `.specify/workflow-catalogs.yml`, the user's
+`~/.specify/workflow-catalogs.yml`, then the built-in defaults. The override and
+the user config come from the server process, not the selected project. The tool
+never fetches catalogs or writes caches.
+
+The result has `items` in priority order and a `truncated` boolean. Each item has
+`name`, `url`, `priority`, `install_allowed`, and nullable `description`.
+`truncated: true` indicates omitted sources; increase `limit` to retrieve more, up
+to 1000. There is no continuation cursor.
+
+Returned URLs contain only the scheme, host, port, and path. Userinfo, query
+strings, and fragments are removed because they can carry credentials, so a
+returned URL may not be fetchable as is, and a secret placed in the URL path is
+not detected. The CLI still prints URLs as configured.
+
+Expected errors are `invalid_project_directory`, `not_a_spec_kit_project`,
+`unsafe_workflow_storage` (including a project catalog config symlink that the
+project's `allow_symlinks` policy denies), and
+`invalid_catalog_config`. Error messages never quote configured values.
+Unexpected failures use sanitized `internal_error`; invalid operation outcomes use
+`invalid_operation_result`. Responses exceeding 1 MiB produce a retryable
+`response_too_large` error instead of partial success; retry with a smaller limit.
+
 ## Input validation errors
 
 The first-class tools validate arguments against their input schema before

docs/workflows/catalogs.md

@@ -12,6 +12,8 @@ discovery continues using the fetched catalog.
 
 > **A project's `.specify/workflow-catalogs.yml` can point `add` and `search` at a catalog you didn't choose.** Before running a workflow from an unfamiliar project, run `specify workflow catalog list` (and `specify workflow step catalog list` for the step catalogs its steps can pull in) — a project supplying that config is not evidence its workflows or steps were vetted. Maintainers do not audit workflow code; review both `run` and `env` in its shell steps yourself, including strict steps (see [Reviewing shell execution](steps.md#reviewing-shell-execution) and [Who maintains workflows?](overview.md#who-maintains-workflows)).
 
+`specify workflow catalog list` applies the project's symlink policy to `.specify/workflow-catalogs.yml`: it refuses a symlinked config unless `allow_symlinks` is `true` in `.specify/init-options.json`, and even then it only follows links to regular files inside the project. Other commands, including `search`, `add`, and `catalog remove`, do not apply this check and still read through the link. If `catalog list` refuses the config, it can't show the sources or the indexes `catalog remove` expects, so inspect the link and its target yourself before running those commands.
+
 ## Discovery and trust
 
 Workflow catalogs advertise `workflow.yml` files or complete packages. Step catalogs advertise executable Python packages through individual file URLs. They are independent stacks with built-in `default` (installation allowed) and `community` (discovery-only) sources. Users must vet sources before installation and use; maintainers do not audit or endorse component code.

src/specify_cli/workflows/_mcp.py

@@ -1,7 +1,8 @@
 """Explicit MCP dispositions for the complete ``specify workflow`` hierarchy.
 
-Only workflow.list has a shared operation and adapter. Unavailable records
-describe the conservative union of CLI paths, not a future MCP contract.
+Only workflow.list and workflow.catalog.list have a shared operation and
+adapter. Unavailable records describe the conservative union of CLI paths, not
+a future MCP contract.
 
 Effect boundaries worth retaining when adapters are introduced:
 
@@ -29,6 +30,8 @@
 from mcp.server import MCPServer
 
 from ._operation_list import WORKFLOW_LIST_OPERATION
+from .catalog._operation_list import WORKFLOW_CATALOG_LIST_OPERATION
+from .catalog.mcp_list import register as register_catalog_list
 from .mcp_list import register as register_list
 
 WorkflowDisposition = Literal["available", "unavailable", "excluded"]
@@ -218,12 +221,11 @@ class WorkflowToolInventory:
         operation_id="workflow.catalog.list",
         cli_path="specify workflow catalog list",
         mcp_tool_name="specify_workflow_catalog_list",
-        disposition_reason=(
-            "WorkflowCatalog.get_catalog_configs returns dictionaries rendered by the "
-            "CLI; no shared typed catalog-list outcome or MCP adapter."
-        ),
-        capabilities=_LOCAL_READ,
-        network_access="none",
+        disposition_reason=None,
+        disposition="available",
+        contract_version=WORKFLOW_CATALOG_LIST_OPERATION.contract_version,
+        capabilities=WORKFLOW_CATALOG_LIST_OPERATION.capabilities,
+        network_access=WORKFLOW_CATALOG_LIST_OPERATION.network_access,
     ),
     WorkflowToolInventory(
         operation_id="workflow.catalog.remove",
@@ -395,11 +397,15 @@ class WorkflowToolInventory:
 
 
 def register(server: MCPServer, *, launch_directory: Path) -> None:
-    """Register only the workflow list operation from its owning hierarchy."""
-    record = next(
-        item for item in WORKFLOW_TOOLS if item.operation_id == "workflow.list"
-    )
-    if record.disposition == "available":
-        register_list(
-            server, launch_directory=launch_directory, tool_name=record.mcp_tool_name
-        )
+    """Register the available workflow operations from their owning hierarchy."""
+    adapters = {
+        "workflow.list": register_list,
+        "workflow.catalog.list": register_catalog_list,
+    }
+    for record in WORKFLOW_TOOLS:
+        if record.disposition == "available":
+            adapters[record.operation_id](
+                server,
+                launch_directory=launch_directory,
+                tool_name=record.mcp_tool_name,
+            )

src/specify_cli/workflows/catalog/_domain.py

@@ -52,6 +52,62 @@ class WorkflowValidationError(WorkflowCatalogError):
     """Validation error for catalog config or workflow data."""
 
 
+# ---------------------------------------------------------------------------
+# Config YAML
+# ---------------------------------------------------------------------------
+
+# How far aliases may unfold a catalog config beyond the document's own length.
+_MAX_CONFIG_ALIAS_EXPANSION = 1024 * 1024
+
+
+def _expanded_size(node: yaml.Node, sizes: dict[int, int], active: set[int]) -> int:
+    """Measure a composed node with aliases unfolded, visiting shared nodes once."""
+    if isinstance(node, yaml.ScalarNode):
+        return len(node.value)
+    key = id(node)
+    if key in sizes:
+        return sizes[key]
+    if key in active:
+        raise WorkflowValidationError(
+            "Catalog config must not contain recursive YAML aliases."
+        )
+    active.add(key)
+    if isinstance(node, yaml.MappingNode):
+        children = [child for pair in node.value for child in pair]
+    else:
+        children = node.value
+    size = 1 + sum(_expanded_size(child, sizes, active) for child in children)
+    active.remove(key)
+    sizes[key] = size
+    return size
+
+
+def _safe_load_catalog_config(text: str) -> Any:
+    """``yaml.safe_load`` that refuses documents whose aliases unfold without bound.
+
+    An alias shares its anchor's node, so a few hundred bytes can describe
+    terabytes that merge keys or a later ``str()`` would materialize. The size is
+    checked before anything is constructed; a document without aliases never
+    reaches the limit.
+    """
+    loader = yaml.SafeLoader(text)
+    try:
+        node = loader.get_single_node()
+        if node is None:
+            return None
+        if _expanded_size(node, {}, set()) > len(text) + _MAX_CONFIG_ALIAS_EXPANSION:
+            raise WorkflowValidationError(
+                "Catalog config expands beyond the size limit through YAML aliases."
+            )
+        return loader.construct_document(node)
+    except RecursionError:
+        raise WorkflowValidationError(
+            "Catalog config is nested too deeply."
+        ) from None
+    finally:
+        loader.dispose()
+
+
 # ---------------------------------------------------------------------------
 # CatalogEntry
 # ---------------------------------------------------------------------------
@@ -321,10 +377,15 @@ class WorkflowCatalog:
     )
     CACHE_DURATION = 3600  # 1 hour
 
-    def __init__(self, project_root: Path) -> None:
+    def __init__(
+        self, project_root: Path, *, read_files: FileHelper | None = None
+    ) -> None:
         self.project_root = project_root
         self.workflows_dir = project_root / ".specify" / "workflows"
         self.cache_dir = self.workflows_dir / ".cache"
+        # Optional read-policy handoff for workflow.catalog.list. Existing callers
+        # and all mutation paths retain their established behavior.
+        self._read_files = read_files
 
     # -- Catalog resolution -----------------------------------------------
 
@@ -360,13 +421,20 @@ def _validate_catalog_url(self, url: str) -> None:
             )
 
     def _load_catalog_config(
-        self, config_path: Path
+        self, config_path: Path, *, read_files: FileHelper | None = None
     ) -> list[WorkflowCatalogEntry] | None:
         """Load catalog stack configuration from a YAML file."""
-        if not config_path.exists():
+        if read_files is None and not config_path.exists():
             return None
         try:
-            data = yaml.safe_load(config_path.read_text(encoding="utf-8"))
+            if read_files is None:
+                text = config_path.read_text(encoding="utf-8")
+            else:
+                try:
+                    text = read_files.read_text(config_path)
+                except FileNotFoundError:
+                    return None
+            data = _safe_load_catalog_config(text)
         except (yaml.YAMLError, OSError, UnicodeError) as exc:
             raise WorkflowValidationError(
                 f"Failed to read catalog config {config_path}: {exc}"
@@ -492,7 +560,9 @@ def get_active_catalogs(self) -> list[WorkflowCatalogEntry]:
 
         # 2. Project-level config
         project_config = self.project_root / ".specify" / "workflow-catalogs.yml"
-        project_entries = self._load_catalog_config(project_config)
+        project_entries = self._load_catalog_config(
+            project_config, read_files=self._read_files
+        )
         if project_entries is not None:
             return project_entries
 
@@ -775,7 +845,9 @@ def add_catalog(self, url: str, name: str | None = None) -> str:
         data: dict[str, Any] = {"catalogs": []}
         if config_path.exists():
             try:
-                raw = yaml.safe_load(config_path.read_text(encoding="utf-8"))
+                raw = _safe_load_catalog_config(
+                    config_path.read_text(encoding="utf-8")
+                )
             except (yaml.YAMLError, OSError, UnicodeDecodeError) as exc:
                 raise WorkflowValidationError(
                     f"Catalog config file is unreadable or malformed: {exc}"
@@ -860,7 +932,7 @@ def remove_catalog(self, index: int) -> str:
             raise WorkflowValidationError("No catalog config file found.")
 
         try:
-            data = yaml.safe_load(config_path.read_text(encoding="utf-8"))
+            data = _safe_load_catalog_config(config_path.read_text(encoding="utf-8"))
         except (yaml.YAMLError, OSError, UnicodeDecodeError) as exc:
             raise WorkflowValidationError(
                 f"Catalog config file is unreadable or malformed: {exc}"

src/specify_cli/workflows/catalog/_operation_list.py

@@ -0,0 +1,132 @@
+"""Shared, read-only application operation for configured workflow catalogs."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+import stat
+from typing import Literal
+
+from ..._file_policy import ProjectPolicyError, project_files
+from ...file_helper import FileHelperError
+from .._storage import WorkflowStorageError, validate_workflow_storage
+from ._domain import WorkflowCatalog, WorkflowCatalogError
+
+WORKFLOW_CATALOG_LIST_MAX_LIMIT = 1000
+
+
+@dataclass(frozen=True)
+class WorkflowCatalogListRequest:
+    project_directory: Path
+    # CLI retains its complete listing. MCP supplies an explicit bounded limit.
+    limit: int | None = None
+
+
+@dataclass(frozen=True)
+class WorkflowCatalogSource:
+    name: str
+    # Exactly as configured, so adapters decide what they may disclose.
+    url: str
+    priority: int
+    install_allowed: bool
+    description: str | None
+
+
+@dataclass(frozen=True)
+class WorkflowCatalogListResult:
+    items: tuple[WorkflowCatalogSource, ...]
+    truncated: bool = False
+
+
+class WorkflowCatalogListError(Exception):
+    """Expected operation failure; adapters own presentation and sanitization."""
+
+    def __init__(self, code: str, message: str) -> None:
+        self.code = code
+        self.details: dict[str, str] = {}
+        self.retryable = False
+        super().__init__(message)
+
+
+@dataclass(frozen=True)
+class WorkflowCatalogListOperationDescriptor:
+    operation_id: Literal["workflow.catalog.list"]
+    contract_version: Literal["1"]
+    request_type: type[WorkflowCatalogListRequest]
+    result_type: type[WorkflowCatalogListResult]
+    warning_types: tuple[type[object], ...]
+    error_types: tuple[type[WorkflowCatalogListError], ...]
+    capabilities: frozenset[Literal["local-read"]]
+    network_access: Literal["none"]
+
+
+WORKFLOW_CATALOG_LIST_OPERATION = WorkflowCatalogListOperationDescriptor(
+    operation_id="workflow.catalog.list",
+    contract_version="1",
+    request_type=WorkflowCatalogListRequest,
+    result_type=WorkflowCatalogListResult,
+    warning_types=(),
+    error_types=(WorkflowCatalogListError,),
+    capabilities=frozenset({"local-read"}),
+    network_access="none",
+)
+
+
+def list_catalog_sources(
+    request: WorkflowCatalogListRequest,
+) -> WorkflowCatalogListResult:
+    """Resolve the active catalog stack without fetching or caching catalogs."""
+    root = request.project_directory
+    if not isinstance(root, Path) or not root.is_absolute():
+        raise WorkflowCatalogListError(
+            "invalid_project_directory", "project_directory must be an absolute path"
+        )
+    limit = request.limit
+    if limit is not None and (
+        type(limit) is not int or not 1 <= limit <= WORKFLOW_CATALOG_LIST_MAX_LIMIT
+    ):
+        raise WorkflowCatalogListError(
+            "invalid_limit",
+            f"limit must be between 1 and {WORKFLOW_CATALOG_LIST_MAX_LIMIT}",
+        )
+    try:
+        # Unlike is_dir(), stat() preserves permission errors on Python 3.14.
+        try:
+            root_is_directory = stat.S_ISDIR(root.stat().st_mode)
+        except (FileNotFoundError, NotADirectoryError, ValueError):
+            root_is_directory = False
+        if not root_is_directory:
+            raise WorkflowCatalogListError(
+                "not_a_spec_kit_project",
+                "Not a Spec Kit project (no .specify/ directory)",
+            )
+        files = project_files(root)
+        if not (root / ".specify").is_dir():
+            raise WorkflowCatalogListError(
+                "not_a_spec_kit_project",
+                "Not a Spec Kit project (no .specify/ directory)",
+            )
+        validate_workflow_storage(root)
+        configs = WorkflowCatalog(root, read_files=files).get_catalog_configs()
+    except (ProjectPolicyError, WorkflowStorageError, FileHelperError, OSError) as exc:
+        raise WorkflowCatalogListError("unsafe_workflow_storage", str(exc)) from exc
+    except WorkflowCatalogError as exc:
+        # Parser and URL diagnostics can quote a configured URL with its credentials.
+        raise WorkflowCatalogListError(
+            "invalid_catalog_config", "The workflow catalog configuration is invalid."
+        ) from exc
+
+    selected = configs if limit is None else configs[:limit]
+    return WorkflowCatalogListResult(
+        items=tuple(
+            WorkflowCatalogSource(
+                name=config["name"],
+                url=config["url"],
+                priority=config["priority"],
+                install_allowed=config["install_allowed"],
+                description=config["description"] or None,
+            )
+            for config in selected
+        ),
+        truncated=limit is not None and len(configs) > limit,
+    )