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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 38 additions & 1 deletion docs/reference/mcp.md
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ read-only, local-only (no network access), and reject unexpected arguments.
| `specify_version` | Return the installed CLI version, runtime, system, and feature capabilities. Takes no 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 |

The artifact tools take an optional absolute `project_directory`; it defaults
to the directory the server was launched from.
Expand All @@ -50,7 +51,43 @@ tool error.
| `specify_run_command` | Run one supported command through the real Specify CLI |

The server does not expose mutations, installation or update operations,
workflows, confirmations, or access tiers.
workflow execution, confirmations, or access tiers.

## Installed workflow list

`specify_workflow_list` accepts an optional absolute `project_directory`
(default: the server launch directory) and `limit` (1–1000, default 100).
It reads only the installed workflow registry; it never loads custom steps,
fetches catalogs, repairs files, or executes workflows.

The result has ordered `items` and a `truncated` boolean. Workflow items have
`kind: "workflow"`, `id`, `name`, `version`, nullable `description`, and `enabled`.
Malformed individual registry entries produce `kind: "warning"` items with
`code: "invalid_registry_entry"`, `workflow_id`, and `message` in their original
positions. The limit counts both workflows and warnings. `truncated: true`
explicitly indicates omitted entries; increase `limit` to retrieve more, up to
1000. There is no continuation cursor. The CLI retains its complete listing.

Missing registry files produce an empty list. Corrupt or unreadable registry
files produce `workflow_registry_read_failed`; unsafe storage produces
`unsafe_workflow_storage`. Other expected errors are `invalid_project_directory`,
`not_a_spec_kit_project`, and `invalid_limit`. Errors contain `code`, `message`,
`details`, and `retryable` under `error`, with `isError: true`.
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. Registry exception details are not exposed through MCP.

Workflow listing reads registry content through the selected project's
`FileHelper` policy owner, rejecting Windows junctions and unsupported reparse
points at the read boundary. Ordinary workflow registry symlinks remain
disallowed. Other workflow registry callers are not migrated by this tool.
These checks do not provide an OS sandbox or race-free filesystem confinement;
the registry is still parsed in full before the entry limit is applied.

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.

## Input validation errors

Expand Down
2 changes: 2 additions & 0 deletions src/specify_cli/mcp_server/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

from ..artifacts._mcp import register as register_artifacts
from ..mcp_version import register as register_version
from ..workflows._mcp import register as register_workflows
from .catalog import (
CommandAdapterError,
CommandDescription,
Expand Down Expand Up @@ -89,6 +90,7 @@ def specify_run_command(command: str) -> VersionResult:

register_version(server)
register_artifacts(server, launch_directory=server_launch_directory)
register_workflows(server, launch_directory=server_launch_directory)

return server

Expand Down
17 changes: 7 additions & 10 deletions src/specify_cli/workflows/_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -183,16 +183,13 @@ def _reject_unsafe_dir(path: Path, label: str) -> None:

def _reject_unsafe_workflow_storage(project_root: Path) -> None:
"""Refuse symlinked workflow storage directories before workflow commands run."""
_reject_unsafe_dir(project_root / ".specify", ".specify")
_reject_unsafe_dir(project_root / ".specify" / "workflows", ".specify/workflows")
_reject_unsafe_dir(
project_root / ".specify" / "workflows" / "runs",
".specify/workflows/runs",
)
_reject_unsafe_dir(
project_root / ".specify" / "workflows" / "overlays",
".specify/workflows/overlays",
)
from ._storage import WorkflowStorageError, validate_workflow_storage

try:
validate_workflow_storage(project_root)
except WorkflowStorageError as exc:
err_console.print(f"[red]Error:[/red] {exc}")
raise typer.Exit(1) from exc


def _resolve_installed_workflow_ownership(
Expand Down
32 changes: 24 additions & 8 deletions src/specify_cli/workflows/_mcp.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Explicit MCP dispositions for the complete ``specify workflow`` hierarchy.

No workflow adapter is registered yet. These records describe the conservative
union of the current CLI paths, not a promise of a future MCP contract.
Only workflow.list has 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:

Expand All @@ -23,8 +23,14 @@
from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Literal

from mcp.server import MCPServer

from ._operation_list import WORKFLOW_LIST_OPERATION
from .mcp_list import register as register_list

WorkflowDisposition = Literal["available", "unavailable", "excluded"]
WorkflowCapability = Literal[
"local-read", "project-write", "execution", "self-modifying"
Expand Down Expand Up @@ -101,12 +107,11 @@ class WorkflowToolInventory:
operation_id="workflow.list",
cli_path="specify workflow list",
mcp_tool_name="specify_workflow_list",
disposition_reason=(
"WorkflowRegistry.list returns registry dictionaries rendered by the CLI; "
"no shared typed list outcome or MCP adapter."
),
capabilities=_LOCAL_READ,
network_access="none",
disposition_reason=None,
disposition="available",
contract_version=WORKFLOW_LIST_OPERATION.contract_version,
capabilities=WORKFLOW_LIST_OPERATION.capabilities,
network_access=WORKFLOW_LIST_OPERATION.network_access,
),
WorkflowToolInventory(
operation_id="workflow.remove",
Expand Down Expand Up @@ -375,3 +380,14 @@ class WorkflowToolInventory:
network_access="none",
),
)


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
)
149 changes: 149 additions & 0 deletions src/specify_cli/workflows/_operation_list.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
"""Shared, read-only application operation for installed workflow listing."""

from __future__ import annotations

from dataclasses import dataclass
from itertools import islice
from pathlib import Path
import stat
from typing import Literal

from .._file_policy import ProjectPolicyError, project_files
from ._storage import WorkflowStorageError, validate_workflow_storage
from .catalog import WorkflowRegistry

WORKFLOW_LIST_MAX_LIMIT = 1000


@dataclass(frozen=True)
class WorkflowListRequest:
project_directory: Path
# CLI retains its complete listing. MCP supplies an explicit bounded limit.
limit: int | None = None


@dataclass(frozen=True)
class WorkflowListEntry:
id: str
name: str
version: str
description: str | None
enabled: bool
kind: Literal["workflow"] = "workflow"


@dataclass(frozen=True)
class WorkflowListWarning:
workflow_id: str
message: str
code: Literal["invalid_registry_entry"] = "invalid_registry_entry"
kind: Literal["warning"] = "warning"


@dataclass(frozen=True)
class WorkflowListResult:
# Warnings stay in registry order, including an all-invalid registry.
items: tuple[WorkflowListEntry | WorkflowListWarning, ...]
truncated: bool = False


class WorkflowListError(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 WorkflowListOperationDescriptor:
operation_id: Literal["workflow.list"]
contract_version: Literal["1"]
request_type: type[WorkflowListRequest]
result_type: type[WorkflowListResult]
warning_types: tuple[type[WorkflowListWarning], ...]
error_types: tuple[type[WorkflowListError], ...]
capabilities: frozenset[Literal["local-read"]]
network_access: Literal["none"]


WORKFLOW_LIST_OPERATION = WorkflowListOperationDescriptor(
operation_id="workflow.list",
contract_version="1",
request_type=WorkflowListRequest,
result_type=WorkflowListResult,
warning_types=(WorkflowListWarning,),
error_types=(WorkflowListError,),
capabilities=frozenset({"local-read"}),
network_access="none",
)


def list_workflows(request: WorkflowListRequest) -> WorkflowListResult:
"""Read the registry without importing custom steps or repairing its contents."""
root = request.project_directory
if not isinstance(root, Path) or not root.is_absolute():
raise WorkflowListError(
"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_LIST_MAX_LIMIT
):
raise WorkflowListError(
"invalid_limit", f"limit must be between 1 and {WORKFLOW_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):
root_is_directory = False
Comment on lines +102 to +103
if not root_is_directory:
raise WorkflowListError(
"not_a_spec_kit_project",
"Not a Spec Kit project (no .specify/ directory)",
)
files = project_files(root)
if not (root / ".specify").is_dir():
raise WorkflowListError(
"not_a_spec_kit_project",
"Not a Spec Kit project (no .specify/ directory)",
)
validate_workflow_storage(root)
except (ProjectPolicyError, WorkflowStorageError, OSError) as exc:
raise WorkflowListError("unsafe_workflow_storage", str(exc)) from exc

try:
installed = WorkflowRegistry(root, read_files=files).list()
except OSError as exc:
raise WorkflowListError(
"workflow_registry_read_failed", f"Failed to read workflow registry: {exc}"
) from exc

items: list[WorkflowListEntry | WorkflowListWarning] = []
selected = installed.items() if limit is None else islice(installed.items(), limit)
for workflow_id, metadata in selected:
if not isinstance(metadata, dict):
items.append(
WorkflowListWarning(
workflow_id=workflow_id,
message=f"Skipping corrupted registry entry '{workflow_id}'.",
)
)
continue
description = metadata.get("description", "")
items.append(
WorkflowListEntry(
id=workflow_id,
name=str(metadata.get("name", workflow_id)),
version=str(metadata.get("version", "?")),
description=str(description) if description else None,
enabled=bool(metadata.get("enabled", True)),
)
)
return WorkflowListResult(
items=tuple(items), truncated=limit is not None and len(installed) > limit
)
22 changes: 22 additions & 0 deletions src/specify_cli/workflows/_storage.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
"""Transport-neutral validation of existing workflow storage boundaries."""

from pathlib import Path


class WorkflowStorageError(ValueError):
"""A workflow storage directory is symlinked or has the wrong type."""


def validate_workflow_storage(project_root: Path) -> None:
"""Check storage without creating directories or following child symlinks."""
for suffix in (
".specify",
".specify/workflows",
".specify/workflows/runs",
".specify/workflows/overlays",
):
path = project_root / suffix
if path.is_symlink():
raise WorkflowStorageError(f"Refusing to use symlinked {suffix} path")
if path.exists() and not path.is_dir():
raise WorkflowStorageError(f"{suffix} exists but is not a directory")
30 changes: 22 additions & 8 deletions src/specify_cli/workflows/catalog/_domain.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
MAX_JSON_CATALOG_BYTES as MAX_JSON_CATALOG_BYTES,
read_response_limited,
)
from ...file_helper import FileHelper, FileHelperError
from .._catalog_cache import read_cache_json, write_cache_json

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -82,10 +83,15 @@ class WorkflowRegistry:
REGISTRY_FILE = "workflow-registry.json"
SCHEMA_VERSION = "1.0"

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.registry_path = self.workflows_dir / self.REGISTRY_FILE
# Optional read-policy handoff for workflow.list. Existing callers and
# all mutation paths retain their established behavior.
self._read_files = read_files
self.data = self._load()

def _has_symlinked_parent(self) -> bool:
Expand Down Expand Up @@ -116,14 +122,22 @@ def _load(self) -> dict[str, Any]:
f"Refusing to read workflow registry at {self.registry_path}: "
"a parent directory or the registry file itself is a symlink"
)
if self.registry_path.exists():
if self._read_files is not None or self.registry_path.exists():
try:
with open(self.registry_path, encoding="utf-8") as f:
data = json.load(f)
except OSError as exc:
# The real data may still be intact on disk. Fail closed at
# construction rather than fabricating an empty registry that
# a read-only caller could mistake for "nothing installed."
if self._read_files is None:
with open(self.registry_path, encoding="utf-8") as f:
data = json.load(f)
else:
try:
text = self._read_files.read_text(self.registry_path)
except FileNotFoundError:
return default_registry
# Match the legacy text stream's universal-newline handling,
# including JSON error offsets for malformed CRLF files.
text = text.replace("\r\n", "\n").replace("\r", "\n")
data = json.loads(text)
except (OSError, FileHelperError) as exc:
# Do not fabricate an empty registry on access/policy failures.
raise OSError(
f"Failed to read workflow registry at {self.registry_path}: {exc}"
) from exc
Expand Down
Loading
Loading