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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 92 additions & 0 deletions src/specify_cli/artifacts/_mcp.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
"""MCP registration and inventory for the ``specify artifact`` hierarchy."""

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 ARTIFACT_LIST_OPERATION
from .mcp_list import register as register_list

ArtifactOperationId = Literal["artifact.list", "artifact.info", "artifact.lookup"]
ArtifactCliPath = Literal[
"specify artifact list",
"specify artifact info",
"specify artifact lookup",
]
ArtifactToolName = Literal[
"specify_artifact_list",
"specify_artifact_info",
"specify_artifact_lookup",
]
ArtifactDisposition = Literal["available", "unavailable"]


@dataclass(frozen=True)
class ArtifactToolInventory:
"""One authoritative MCP disposition for an artifact CLI leaf."""

operation_id: ArtifactOperationId
cli_path: ArtifactCliPath
mcp_tool_name: ArtifactToolName
contract_version: str | None
disposition: ArtifactDisposition
disposition_reason: str | None
capabilities: frozenset[Literal["local-read"]]
network_access: Literal["none"]


ARTIFACT_TOOLS = (
ArtifactToolInventory(
operation_id=ARTIFACT_LIST_OPERATION.operation_id,
cli_path="specify artifact list",
mcp_tool_name="specify_artifact_list",
contract_version=ARTIFACT_LIST_OPERATION.contract_version,
disposition="available",
disposition_reason=None,
capabilities=ARTIFACT_LIST_OPERATION.capabilities,
network_access=ARTIFACT_LIST_OPERATION.network_access,
),
ArtifactToolInventory(
operation_id="artifact.info",
cli_path="specify artifact info",
mcp_tool_name="specify_artifact_info",
contract_version=None,
disposition="unavailable",
disposition_reason=(
"The artifact.info CLI leaf does not yet have a shared typed operation."
),
capabilities=frozenset({"local-read"}),
network_access="none",
),
ArtifactToolInventory(
operation_id="artifact.lookup",
cli_path="specify artifact lookup",
mcp_tool_name="specify_artifact_lookup",
contract_version=None,
disposition="unavailable",
disposition_reason=(
"The artifact.lookup CLI leaf does not yet have a shared typed operation."
),
capabilities=frozenset({"local-read"}),
network_access="none",
),
)


def register(server: MCPServer, *, launch_directory: Path) -> None:
"""Register every available artifact tool from the explicit inventory."""
available = {
item.operation_id: item
for item in ARTIFACT_TOOLS
if item.disposition == "available"
}
list_tool = available[ARTIFACT_LIST_OPERATION.operation_id]
register_list(
server,
launch_directory=launch_directory,
tool_name=list_tool.mcp_tool_name,
)
91 changes: 85 additions & 6 deletions src/specify_cli/artifacts/_operation_list.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,20 +73,26 @@ class ArtifactListHookRow(TypedDict):


ArtifactListItem = ArtifactListRow | ArtifactListHookRow
ARTIFACT_LIST_MAX_LIMIT = 1000
ARTIFACT_LIST_CURSOR_MAX_LENGTH = 20


@dataclass(frozen=True)
class ArtifactListRequest:
"""Explicit project context for the ``artifact.list`` operation."""

project_directory: Path
limit: int | None = None
cursor: str | None = None


@dataclass(frozen=True)
class ArtifactListResult:
"""Transport-neutral artifact inventory."""

rows: tuple[ArtifactListItem, ...]
next_cursor: str | None = None
truncated: bool = False


class ArtifactListError(ArtifactError):
Expand All @@ -111,6 +117,17 @@ def __init__(
super().__init__(message)


class ArtifactListProjectDirectoryError(ArtifactListError):
"""The supplied project directory is not an absolute path."""

def __init__(self, project_directory: Path) -> None:
super().__init__(
code="invalid_project_directory",
message="project_directory must be an absolute path",
details={"project_directory": str(project_directory)},
)


class ArtifactListProjectError(ArtifactListError):
"""The supplied directory is not a Spec Kit project root."""

Expand All @@ -133,12 +150,33 @@ def __init__(self, project_directory: Path) -> None:
)


class ArtifactListPaginationError(ArtifactListError):
"""The requested artifact page is invalid."""

def __init__(self, *, field: Literal["limit", "cursor"], value: object) -> None:
if field == "limit":
message = (
f"artifact list limit must be between 1 and {ARTIFACT_LIST_MAX_LIMIT}"
)
else:
message = "artifact list cursor must be a canonical decimal offset"
super().__init__(
code="invalid_pagination",
message=message,
details={
"field": field,
"value": value,
"max_limit": ARTIFACT_LIST_MAX_LIMIT,
},
)


@dataclass(frozen=True)
class ArtifactListOperationDescriptor:
"""Stable metadata shared by delivery adapters for ``artifact.list``."""

operation_id: Literal["artifact.list"]
contract_version: Literal["1"]
contract_version: Literal["2"]
request_type: type[ArtifactListRequest]
result_type: type[ArtifactListResult]
warning_types: tuple[type[object], ...]
Expand All @@ -149,21 +187,52 @@ class ArtifactListOperationDescriptor:

ARTIFACT_LIST_OPERATION = ArtifactListOperationDescriptor(
operation_id="artifact.list",
contract_version="1",
contract_version="2",
request_type=ArtifactListRequest,
result_type=ArtifactListResult,
warning_types=(),
error_types=(ArtifactListProjectError, ArtifactListResolutionError),
error_types=(
ArtifactListProjectDirectoryError,
ArtifactListProjectError,
ArtifactListResolutionError,
ArtifactListPaginationError,
Comment thread
mnriem marked this conversation as resolved.
),
capabilities=frozenset({"local-read"}),
network_access="none",
)


def _pagination_offset(request: ArtifactListRequest) -> int:
limit = request.limit
if limit is not None and (
isinstance(limit, bool)
or not isinstance(limit, int)
or limit < 1
or limit > ARTIFACT_LIST_MAX_LIMIT
):
raise ArtifactListPaginationError(field="limit", value=limit)

cursor = request.cursor
if cursor is None:
return 0
if (
not isinstance(cursor, str)
or not cursor
or len(cursor) > ARTIFACT_LIST_CURSOR_MAX_LENGTH
or not cursor.isascii()
or not cursor.isdecimal()
or (len(cursor) > 1 and cursor.startswith("0"))
):
raise ArtifactListPaginationError(field="cursor", value=cursor)
return int(cursor)


def list_artifacts(request: ArtifactListRequest) -> ArtifactListResult:
"""Return the complete typed artifact inventory for an explicit project."""
"""Return one typed artifact inventory page for an explicit project."""
project_directory = Path(request.project_directory)
if not project_directory.is_absolute():
raise ArtifactListProjectError(project_directory)
raise ArtifactListProjectDirectoryError(project_directory)
offset = _pagination_offset(request)

try:
rows = ArtifactCatalog(project_directory).list_artifacts_with_stack()
Expand All @@ -172,6 +241,16 @@ def list_artifacts(request: ArtifactListRequest) -> ArtifactListResult:
except (ArtifactResolutionError, OSError, PresetError) as exc:
raise ArtifactListResolutionError(project_directory) from exc

all_rows = tuple(cast(ArtifactListItem, row) for row in rows)
page_rows = (
all_rows[offset:]
if request.limit is None
else all_rows[offset : offset + request.limit]
)
page_end = offset + len(page_rows)
truncated = page_end < len(all_rows)
return ArtifactListResult(
rows=tuple(cast(ArtifactListItem, row) for row in rows),
rows=page_rows,
next_cursor=str(page_end) if truncated else None,
truncated=truncated,
)
Loading
Loading