diff --git a/src/specify_cli/_operation_version.py b/src/specify_cli/_operation_version.py index 97318b6dda..a91f25f918 100644 --- a/src/specify_cli/_operation_version.py +++ b/src/specify_cli/_operation_version.py @@ -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 @@ -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 { diff --git a/src/specify_cli/mcp_server/server.py b/src/specify_cli/mcp_server/server.py index aebbfc3c9e..13d1a9e023 100644 --- a/src/specify_cli/mcp_server/server.py +++ b/src/specify_cli/mcp_server/server.py @@ -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, @@ -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", @@ -78,6 +78,8 @@ def specify_run_command(command: str) -> VersionResult: except CommandAdapterError as exc: return _tool_error(exc) + register_version(server) + return server diff --git a/src/specify_cli/mcp_version.py b/src/specify_cli/mcp_version.py new file mode 100644 index 0000000000..3e4e35e506 --- /dev/null +++ b/src/specify_cli/mcp_version.py @@ -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) diff --git a/tests/specify_cli/mcp_server/test_server.py b/tests/specify_cli/mcp_server/test_server.py index 41e08ea5e5..ffe61511ec 100644 --- a/tests/specify_cli/mcp_server/test_server.py +++ b/tests/specify_cli/mcp_server/test_server.py @@ -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" diff --git a/tests/specify_cli/mcp_server/test_stdio.py b/tests/specify_cli/mcp_server/test_stdio.py index 699cc723f9..b3ca22f05d 100644 --- a/tests/specify_cli/mcp_server/test_stdio.py +++ b/tests/specify_cli/mcp_server/test_stdio.py @@ -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"}, @@ -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", diff --git a/tests/specify_cli/test_mcp_version.py b/tests/specify_cli/test_mcp_version.py new file mode 100644 index 0000000000..c2e0132096 --- /dev/null +++ b/tests/specify_cli/test_mcp_version.py @@ -0,0 +1,392 @@ +"""Tests for the first-class ``specify_version`` MCP adapter.""" + +from __future__ import annotations + +import asyncio +import logging +from unittest.mock import Mock, patch + +import anyio +import pytest +from mcp import ClientSession +from mcp.server import MCPServer +from mcp.server.mcpserver.exceptions import ToolError +from mcp.shared.memory import create_client_server_memory_streams + +from specify_cli import app +from specify_cli._operation_version import ( + VERSION_OPERATION, + VersionResult, + VersionRuntime, + VersionSystem, +) +from specify_cli.mcp_server.server import create_server +from specify_cli.mcp_version import ( + VERSION_TOOL, + VersionToolResult, + register, + specify_version, +) + +TOOL_NAME = VERSION_TOOL.mcp_tool_name +TOOL_DESCRIPTION = ( + "Return the installed Spec Kit CLI version, runtime, system, and " + "feature capabilities." +) + +VERSION_RESULT = VersionResult( + cli_version="1.2.3", + runtime=VersionRuntime( + python="3.13.1", + openssl="OpenSSL 3.4.0", + ), + system=VersionSystem( + platform="ExampleOS", + architecture="example64", + os_version="ExampleOS 4.5", + ), + features={"workflow_catalog": True}, +) +VERSION_PAYLOAD = { + "cli_version": "1.2.3", + "runtime": { + "python": "3.13.1", + "openssl": "OpenSSL 3.4.0", + }, + "system": { + "platform": "ExampleOS", + "architecture": "example64", + "os_version": "ExampleOS 4.5", + }, + "features": {"workflow_catalog": True}, +} + + +def _run(coro): + return asyncio.run(coro) + + +def test_specify_version_dispatches_directly_to_shared_operation(): + command_runner = Mock(side_effect=AssertionError("subprocess executor reached")) + + with ( + patch( + "specify_cli.mcp_version.collect_version_result", + return_value=VERSION_RESULT, + ) as collect, + patch( + "specify_cli.mcp_server.executor._run_cli_process", + side_effect=AssertionError("subprocess executor reached"), + ) as run_cli_process, + patch( + "specify_cli.command_version.version", + side_effect=AssertionError("CLI adapter reached"), + ) as cli_adapter, + ): + server = create_server(command_runner=command_runner) + result = _run(server.call_tool(TOOL_NAME, {})) + + assert result.structured_content == VERSION_PAYLOAD + collect.assert_called_once_with() + command_runner.assert_not_called() + run_cli_process.assert_not_called() + cli_adapter.assert_not_called() + + +def test_specify_version_matches_shared_operation_and_cli_json_semantics(): + with patch( + "specify_cli.mcp_version.collect_version_result", + return_value=VERSION_RESULT, + ): + result = specify_version() + + assert isinstance(result, VersionToolResult) + assert result.model_dump() == VERSION_PAYLOAD + assert isinstance(result.cli_version, str) + assert isinstance(result.runtime.python, str) + assert isinstance(result.runtime.openssl, str) + assert isinstance(result.system.platform, str) + assert isinstance(result.system.architecture, str) + assert isinstance(result.system.os_version, str) + assert all(isinstance(value, bool) for value in result.features.values()) + + +def test_specify_version_preserves_unavailable_openssl_as_none(): + operation_result = VersionResult( + cli_version=VERSION_RESULT.cli_version, + runtime=VersionRuntime( + python=VERSION_RESULT.runtime.python, + openssl=None, + ), + system=VERSION_RESULT.system, + features=VERSION_RESULT.features, + ) + with patch( + "specify_cli.mcp_version.collect_version_result", + return_value=operation_result, + ): + result = specify_version() + + assert isinstance(result, VersionToolResult) + assert result.runtime.openssl is None + + +@pytest.mark.parametrize( + "operation_result", + [ + VersionResult( + cli_version="1.2.3", + runtime=None, + system=VERSION_RESULT.system, + features=VERSION_RESULT.features, + ), + VersionResult( + cli_version="1.2.3", + runtime=VERSION_RESULT.runtime, + system=None, + features=VERSION_RESULT.features, + ), + VersionResult( + cli_version="1.2.3", + runtime=VERSION_RESULT.runtime, + system=VERSION_RESULT.system, + features={"workflow_catalog": "yes"}, + ), + ], +) +def test_specify_version_rejects_invalid_operation_results(operation_result): + with patch( + "specify_cli.mcp_version.collect_version_result", + return_value=operation_result, + ): + result = specify_version() + + assert result.is_error is True + assert result.structured_content == { + "error": { + "code": "invalid_operation_result", + "message": "The version operation returned an invalid result.", + "details": {}, + } + } + assert result.content[0].text == ( + "invalid_operation_result: The version operation returned an invalid result." + ) + + +def test_specify_version_sanitizes_and_logs_unexpected_operation_failure(caplog): + unsafe = "SECRET_TOKEN=do-not-print /Users/example/private/project" + with ( + patch( + "specify_cli.mcp_version.collect_version_result", + side_effect=RuntimeError(unsafe), + ), + caplog.at_level(logging.ERROR, logger="specify_cli.mcp_version"), + ): + result = specify_version() + + assert result.is_error is True + assert result.structured_content == { + "error": { + "code": "internal_error", + "message": "Unable to collect version information.", + "details": {}, + } + } + rendered = result.model_dump_json(by_alias=True) + assert unsafe not in rendered + assert "RuntimeError" not in rendered + assert "Traceback" not in rendered + assert len(caplog.records) == 1 + assert caplog.records[0].message == ( + "Unexpected failure in the version MCP adapter." + ) + assert caplog.records[0].exc_info is not None + + +def test_version_inventory_matches_operation_cli_and_registration(): + callback_names = [ + command.callback.__name__ + for command in app.registered_commands + if command.callback is not None + ] + tool_names = [tool.name for tool in _run(create_server().list_tools())] + + assert VERSION_TOOL.operation_id == VERSION_OPERATION.operation_id + assert VERSION_TOOL.cli_path.split() == ["specify", VERSION_TOOL.operation_id] + assert callback_names.count(VERSION_TOOL.operation_id) == 1 + assert tool_names.count(VERSION_TOOL.mcp_tool_name) == 1 + assert VERSION_TOOL.contract_version == VERSION_OPERATION.contract_version + assert VERSION_TOOL.disposition == "available" + assert VERSION_TOOL.disposition_reason is None + assert VERSION_TOOL.capabilities == VERSION_OPERATION.capabilities + assert VERSION_TOOL.network_access == VERSION_OPERATION.network_access + + +def test_specify_version_discovery_has_no_argument_schema_and_typed_output(): + tools = _run(create_server().list_tools()) + tool = next(tool for tool in tools if tool.name == TOOL_NAME) + + assert tool.input_schema == { + "additionalProperties": False, + "properties": {}, + "title": "specify_versionArguments", + "type": "object", + } + assert tool.description == TOOL_DESCRIPTION + assert tool.output_schema == { + "$defs": { + "VersionRuntimeResult": { + "additionalProperties": False, + "description": "Runtime portion of the MCP version result.", + "properties": { + "python": { + "title": "Python", + "type": "string", + }, + "openssl": { + "anyOf": [ + {"type": "string"}, + {"type": "null"}, + ], + "title": "Openssl", + }, + }, + "required": ["python", "openssl"], + "title": "VersionRuntimeResult", + "type": "object", + }, + "VersionSystemResult": { + "additionalProperties": False, + "description": "System portion of the MCP version result.", + "properties": { + "platform": { + "title": "Platform", + "type": "string", + }, + "architecture": { + "title": "Architecture", + "type": "string", + }, + "os_version": { + "title": "Os Version", + "type": "string", + }, + }, + "required": [ + "platform", + "architecture", + "os_version", + ], + "title": "VersionSystemResult", + "type": "object", + }, + }, + "additionalProperties": False, + "description": "Typed structured result returned by ``specify_version``.", + "properties": { + "cli_version": { + "title": "Cli Version", + "type": "string", + }, + "runtime": { + "$ref": "#/$defs/VersionRuntimeResult", + }, + "system": { + "$ref": "#/$defs/VersionSystemResult", + }, + "features": { + "additionalProperties": { + "type": "boolean", + }, + "title": "Features", + "type": "object", + }, + }, + "required": [ + "cli_version", + "runtime", + "system", + "features", + ], + "title": "VersionToolResult", + "type": "object", + } + assert tool.annotations.read_only_hint is True + assert tool.annotations.destructive_hint is None + assert tool.annotations.idempotent_hint is True + assert tool.annotations.open_world_hint is False + + +def test_specify_version_rejects_unexpected_arguments_before_dispatch(): + collect = Mock(side_effect=AssertionError("operation reached")) + with ( + patch("specify_cli.mcp_version.collect_version_result", collect), + pytest.raises(ToolError, match="Extra inputs are not permitted"), + ): + _run(create_server().call_tool(TOOL_NAME, {"unexpected": True})) + + collect.assert_not_called() + + +def test_register_adds_one_tool_and_rejects_name_collision(): + server = MCPServer(name="test") + + register(server) + tools = _run(server.list_tools()) + + assert [tool.name for tool in tools] == [TOOL_NAME] + with pytest.raises(ValueError, match=f"MCP tool name collision: {TOOL_NAME}"): + register(server) + assert [tool.name for tool in _run(server.list_tools())] == [TOOL_NAME] + + +def test_in_memory_client_preserves_success_and_failure_wire_shapes(): + async def exercise(): + server = create_server() + async with ( + create_client_server_memory_streams() as ( + client_streams, + server_streams, + ), + anyio.create_task_group() as task_group, + ): + task_group.start_soon( + server._lowlevel_server.run, + server_streams[0], + server_streams[1], + server._lowlevel_server.create_initialization_options(), + ) + async with ClientSession(*client_streams) as session: + await session.initialize() + with patch( + "specify_cli.mcp_version.collect_version_result", + return_value=VERSION_RESULT, + ): + success = await session.call_tool(TOOL_NAME, {}) + with patch( + "specify_cli.mcp_version.collect_version_result", + return_value=VersionResult( + cli_version="1.2.3", + runtime=None, + system=None, + features={}, + ), + ): + failure = await session.call_tool(TOOL_NAME, {}) + schema_failure = await session.call_tool( + TOOL_NAME, + {"unexpected": True}, + ) + task_group.cancel_scope.cancel() + return success, failure, schema_failure + + success, failure, schema_failure = _run(exercise()) + + assert success.is_error is False + assert success.structured_content == VERSION_PAYLOAD + assert failure.is_error is True + assert failure.structured_content["error"]["code"] == "invalid_operation_result" + assert schema_failure.is_error is True + assert schema_failure.structured_content is None + assert "Extra inputs are not permitted" in schema_failure.content[0].text diff --git a/tests/specify_cli/test_operation_version.py b/tests/specify_cli/test_operation_version.py index 802d627b8a..56c71d5744 100644 --- a/tests/specify_cli/test_operation_version.py +++ b/tests/specify_cli/test_operation_version.py @@ -6,6 +6,7 @@ from specify_cli import _operation_version from specify_cli._operation_version import ( + VERSION_OPERATION, VersionResult, VersionRuntime, VersionSystem, @@ -23,6 +24,14 @@ } +def test_version_operation_descriptor_is_stable(): + """Adapters share one stable operation identity and capability contract.""" + assert VERSION_OPERATION.operation_id == "version" + assert VERSION_OPERATION.contract_version == "1" + assert VERSION_OPERATION.capabilities == frozenset({"local-read"}) + assert VERSION_OPERATION.network_access == "none" + + def test_collect_version_result_returns_complete_typed_result(): """The operation owns complete semantic version collection.""" with (