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
19 changes: 19 additions & 0 deletions src/specify_cli/_operation_version.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import platform
from collections.abc import Callable
from dataclasses import dataclass
from typing import Literal

from ._assets import get_speckit_version

Expand Down Expand Up @@ -36,6 +37,24 @@ class VersionResult:
features: dict[str, bool]


@dataclass(frozen=True)
class VersionOperationDescriptor:
"""Stable metadata shared by delivery adapters for ``version``."""

operation_id: Literal["version"]
contract_version: Literal["1"]
capabilities: frozenset[Literal["local-read"]]
network_access: Literal["none"]


VERSION_OPERATION = VersionOperationDescriptor(
operation_id="version",
contract_version="1",
capabilities=frozenset({"local-read"}),
network_access="none",
)


def _feature_capabilities() -> dict[str, bool]:
"""Return stable local CLI capability flags for humans and agents."""
return {
Expand Down
6 changes: 4 additions & 2 deletions src/specify_cli/mcp_server/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from mcp.server import MCPServer
from mcp.types import CallToolResult, TextContent

from ..mcp_version import register as register_version
from .catalog import (
CommandAdapterError,
CommandDescription,
Expand Down Expand Up @@ -42,8 +43,7 @@ def create_server(
name="specify",
title="Spec Kit CLI",
description=(
"Experimental stdio-only MCP adapter for stable Specify CLI JSON "
"commands."
"Experimental stdio-only MCP adapter for stable Specify CLI JSON commands."
),
version="experimental",
log_level="ERROR",
Expand Down Expand Up @@ -78,6 +78,8 @@ def specify_run_command(command: str) -> VersionResult:
except CommandAdapterError as exc:
return _tool_error(exc)

register_version(server)

return server


Expand Down
164 changes: 164 additions & 0 deletions src/specify_cli/mcp_version.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,164 @@
"""MCP adapter for the shared ``version`` operation."""

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Literal

from mcp.server import MCPServer
from mcp.types import CallToolResult, TextContent, ToolAnnotations
from pydantic import BaseModel, ConfigDict, ValidationError

from ._operation_version import VERSION_OPERATION, VersionResult, collect_version_result

logger = logging.getLogger(__name__)
_INVALID_RESULT_MESSAGE = "The version operation returned an invalid result."
_INTERNAL_ERROR_MESSAGE = "Unable to collect version information."
_TOOL_DESCRIPTION = (
"Return the installed Spec Kit CLI version, runtime, system, and "
"feature capabilities."
)


@dataclass(frozen=True)
class VersionToolInventory:
"""Authoritative MCP inventory record for the version operation."""

operation_id: Literal["version"]
cli_path: Literal["specify version"]
mcp_tool_name: Literal["specify_version"]
contract_version: Literal["1"]
disposition: Literal["available"]
disposition_reason: None
capabilities: frozenset[Literal["local-read"]]
network_access: Literal["none"]


VERSION_TOOL = VersionToolInventory(
operation_id=VERSION_OPERATION.operation_id,
cli_path="specify version",
mcp_tool_name="specify_version",
contract_version=VERSION_OPERATION.contract_version,
disposition="available",
disposition_reason=None,
capabilities=VERSION_OPERATION.capabilities,
network_access=VERSION_OPERATION.network_access,
)


class VersionRuntimeResult(BaseModel):
"""Runtime portion of the MCP version result."""

model_config = ConfigDict(extra="forbid")

python: str
openssl: str | None


class VersionSystemResult(BaseModel):
"""System portion of the MCP version result."""

model_config = ConfigDict(extra="forbid")

platform: str
architecture: str
os_version: str


class VersionToolResult(BaseModel):
"""Typed structured result returned by ``specify_version``."""

model_config = ConfigDict(extra="forbid")

cli_version: str
runtime: VersionRuntimeResult
system: VersionSystemResult
features: dict[str, bool]


class _InvalidOperationResult(Exception):
"""The shared operation returned an incomplete or invalid typed result."""


def _tool_error(code: str, message: str) -> CallToolResult:
payload = {
"error": {
"code": code,
"message": message,
"details": {},
}
}
return CallToolResult(
content=[TextContent(type="text", text=f"{code}: {message}")],
structuredContent=payload,
isError=True,
)


def _convert_result(result: VersionResult) -> VersionToolResult:
if result.runtime is None or result.system is None:
raise _InvalidOperationResult

try:
return VersionToolResult.model_validate(
{
"cli_version": result.cli_version,
"runtime": {
"python": result.runtime.python,
"openssl": result.runtime.openssl,
},
"system": {
"platform": result.system.platform,
"architecture": result.system.architecture,
"os_version": result.system.os_version,
},
"features": result.features,
},
strict=True,
)
except ValidationError as exc:
raise _InvalidOperationResult from exc


def specify_version() -> VersionToolResult:
"""Return installed CLI, runtime, system, and feature information."""
try:
return _convert_result(collect_version_result())
except _InvalidOperationResult:
return _tool_error("invalid_operation_result", _INVALID_RESULT_MESSAGE)
except Exception:
logger.exception("Unexpected failure in the version MCP adapter.")
return _tool_error("internal_error", _INTERNAL_ERROR_MESSAGE)


def _forbid_unexpected_arguments(server: MCPServer) -> None:
tool = server._tool_manager.get_tool(VERSION_TOOL.mcp_tool_name)
if tool is None: # pragma: no cover - registration immediately precedes this
raise RuntimeError(f"Tool registration failed: {VERSION_TOOL.mcp_tool_name}")

# MCP SDK argument models ignore extras by default, despite publishing an
# empty schema for a no-argument callable.
argument_model = tool.fn_metadata.arg_model
argument_model.model_config["extra"] = "forbid"
argument_model.model_rebuild(force=True)
tool.parameters = argument_model.model_json_schema(by_alias=True)


def register(server: MCPServer) -> None:
"""Register the first-class ``specify_version`` MCP tool exactly once."""
if server._tool_manager.get_tool(VERSION_TOOL.mcp_tool_name) is not None:
raise ValueError(f"MCP tool name collision: {VERSION_TOOL.mcp_tool_name}")

server.add_tool(
specify_version,
name=VERSION_TOOL.mcp_tool_name,
description=_TOOL_DESCRIPTION,
annotations=ToolAnnotations(
readOnlyHint=True,
idempotentHint=True,
openWorldHint=False,
),
structured_output=True,
)
_forbid_unexpected_arguments(server)
5 changes: 4 additions & 1 deletion tests/specify_cli/mcp_server/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,16 +23,19 @@ def _run(coro):
return asyncio.run(coro)


def test_tool_discovery_exposes_only_generic_surface_with_typed_inputs():
def test_tool_discovery_exposes_first_class_and_transitional_tools():
tools = _run(create_server().list_tools())

assert [tool.name for tool in tools] == [
"specify_list_commands",
"specify_describe_command",
"specify_run_command",
"specify_version",
]
schemas = {tool.name: tool.input_schema for tool in tools}
assert schemas["specify_list_commands"]["properties"] == {}
assert schemas["specify_version"]["properties"] == {}
assert schemas["specify_version"]["additionalProperties"] is False
for name in ("specify_describe_command", "specify_run_command"):
assert schemas[name]["required"] == ["command"]
assert schemas[name]["properties"]["command"]["type"] == "string"
Expand Down
9 changes: 9 additions & 0 deletions tests/specify_cli/mcp_server/test_stdio.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ async def exercise() -> None:
initialized = await session.initialize()
tools = await session.list_tools()
listed = await session.call_tool("specify_list_commands", {})
version = await session.call_tool("specify_version", {})
ran = await session.call_tool(
"specify_run_command",
{"command": "version"},
Expand All @@ -49,8 +50,16 @@ async def exercise() -> None:
"specify_list_commands",
"specify_describe_command",
"specify_run_command",
"specify_version",
]
assert listed.structured_content["commands"][0]["command"] == "version"
assert set(version.structured_content) == {
"cli_version",
"runtime",
"system",
"features",
}
assert version.is_error is False
assert set(ran.structured_content) == {
"cli_version",
"runtime",
Expand Down
Loading
Loading