From df5e416e793742bd12de0f1ac13bedc216c6113e Mon Sep 17 00:00:00 2001 From: Manfred Riem <15701806+mnriem@users.noreply.github.com> Date: Wed, 7 Oct 2026 09:03:08 -0500 Subject: [PATCH 1/5] feat(mcp): add artifact list tool Assisted-by: GitHub Copilot (model: GPT-5.6 Sol, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/specify_cli/artifacts/_mcp.py | 92 +++ src/specify_cli/artifacts/mcp_list.py | 238 ++++++ src/specify_cli/mcp_server/server.py | 12 +- tests/specify_cli/artifacts/test_mcp_list.py | 772 +++++++++++++++++++ tests/specify_cli/mcp_server/test_server.py | 7 +- tests/specify_cli/mcp_server/test_stdio.py | 20 +- 6 files changed, 1137 insertions(+), 4 deletions(-) create mode 100644 src/specify_cli/artifacts/_mcp.py create mode 100644 src/specify_cli/artifacts/mcp_list.py create mode 100644 tests/specify_cli/artifacts/test_mcp_list.py diff --git a/src/specify_cli/artifacts/_mcp.py b/src/specify_cli/artifacts/_mcp.py new file mode 100644 index 0000000000..1a835def04 --- /dev/null +++ b/src/specify_cli/artifacts/_mcp.py @@ -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, + ) diff --git a/src/specify_cli/artifacts/mcp_list.py b/src/specify_cli/artifacts/mcp_list.py new file mode 100644 index 0000000000..e067057e2c --- /dev/null +++ b/src/specify_cli/artifacts/mcp_list.py @@ -0,0 +1,238 @@ +"""MCP adapter for the shared ``artifact.list`` operation.""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from pathlib import Path +from typing import Annotated, Literal + +from mcp.server import MCPServer +from mcp.types import CallToolResult, TextContent, ToolAnnotations +from pydantic import BaseModel, ConfigDict, Field, ValidationError + +from ._operation_list import ( + ArtifactListError, + ArtifactListRequest, + ArtifactListResult, + list_artifacts, +) + +logger = logging.getLogger(__name__) +_INVALID_RESULT_MESSAGE = "The artifact list operation returned an invalid result." +_INTERNAL_ERROR_MESSAGE = "Unable to list Spec Kit artifacts." +_TOOL_DESCRIPTION = ( + "List every command, template, script, and hook Spec Kit exposes for a project." +) + + +class ArtifactListToolInput(BaseModel): + """Typed input accepted by ``specify_artifact_list``.""" + + model_config = ConfigDict(extra="forbid") + + project_directory: str | None = None + + +class ArtifactListStackEntryResult(BaseModel): + """Composition entry for a named artifact.""" + + model_config = ConfigDict(extra="forbid") + + id: str + layer: Literal["project", "preset", "extension"] | None + sourceId: str | None + presetId: str | None + presetName: str | None + strategy: Literal["replace", "wrap", "prepend", "append"] + active: bool + hidden: bool + manifestPath: str | None + lookupId: str | None + sourcePath: str | None + + +class ArtifactListHookStackEntryResult(BaseModel): + """Composition entry for a hook artifact.""" + + model_config = ConfigDict(extra="forbid") + + id: str + layer: Literal["preset", "extension"] + sourceId: str + presetId: str | None + presetName: str | None + strategy: Literal["additive"] + active: bool + hidden: bool + manifestPath: str + lookupId: str + sourcePath: None + priority: int + optional: bool + + +class ArtifactListRowResult(BaseModel): + """One named artifact inventory row.""" + + model_config = ConfigDict(extra="forbid") + + id: str + name: str + kind: Literal["command", "template", "script"] + description: str + stack: list[ArtifactListStackEntryResult] + + +class ArtifactListHookRowResult(BaseModel): + """One hook inventory row.""" + + model_config = ConfigDict(extra="forbid") + + id: str + name: str + kind: Literal["hook"] + description: str + eventName: str + targetCommand: str + registered: bool + stack: list[ArtifactListHookStackEntryResult] + + +ArtifactListItemResult = Annotated[ + ArtifactListRowResult | ArtifactListHookRowResult, + Field(discriminator="kind"), +] + + +class ArtifactListToolResult(BaseModel): + """Typed structured result returned by ``specify_artifact_list``.""" + + model_config = ConfigDict(extra="forbid") + + rows: list[ArtifactListItemResult] + + +class _InvalidOperationResult(Exception): + """The shared operation returned an incomplete or invalid typed result.""" + + +ArtifactListTool = Callable[[str | None], ArtifactListToolResult | CallToolResult] + + +def _tool_error( + code: str, + message: str, + *, + details: dict[str, object] | None = None, + retryable: bool = False, +) -> CallToolResult: + payload = { + "error": { + "code": code, + "message": message, + "details": details or {}, + "retryable": retryable, + } + } + return CallToolResult( + content=[TextContent(type="text", text=f"{code}: {message}")], + structuredContent=payload, + isError=True, + ) + + +def _convert_result(result: ArtifactListResult) -> ArtifactListToolResult: + try: + rows = result.rows + except AttributeError as exc: + raise _InvalidOperationResult from exc + + try: + return ArtifactListToolResult.model_validate( + {"rows": list(rows)}, + strict=True, + ) + except (TypeError, ValidationError) as exc: + raise _InvalidOperationResult from exc + + +def create_artifact_list_tool(*, launch_directory: Path) -> ArtifactListTool: + """Create a tool bound to the immutable MCP server launch directory.""" + launch_directory = Path(launch_directory) + if not launch_directory.is_absolute(): + raise ValueError("MCP server launch directory must be absolute") + + def specify_artifact_list( + project_directory: str | None = None, + ) -> ArtifactListToolResult: + """List every artifact exposed by the selected Spec Kit project.""" + tool_input = ArtifactListToolInput.model_validate( + {"project_directory": project_directory}, + strict=True, + ) + selected_directory = ( + launch_directory + if tool_input.project_directory is None + else Path(tool_input.project_directory) + ) + try: + return _convert_result( + list_artifacts( + ArtifactListRequest(project_directory=selected_directory), + ) + ) + except ArtifactListError as exc: + return _tool_error( + exc.code, + exc.message, + details=exc.details, + retryable=exc.retryable, + ) + except _InvalidOperationResult: + return _tool_error( + "invalid_operation_result", + _INVALID_RESULT_MESSAGE, + ) + except Exception: + logger.exception("Unexpected failure in the artifact list MCP adapter.") + return _tool_error("internal_error", _INTERNAL_ERROR_MESSAGE) + + return specify_artifact_list + + +def _forbid_unexpected_arguments(server: MCPServer, tool_name: str) -> None: + tool = server._tool_manager.get_tool(tool_name) + if tool is None: # pragma: no cover - registration immediately precedes this + raise RuntimeError(f"Tool registration failed: {tool_name}") + + # MCP SDK argument models ignore extras by default even when discovery + # advertises a closed command-specific schema. + 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, + *, + launch_directory: Path, + tool_name: str = "specify_artifact_list", +) -> None: + """Register the first-class artifact-list MCP tool exactly once.""" + if server._tool_manager.get_tool(tool_name) is not None: + raise ValueError(f"MCP tool name collision: {tool_name}") + + server.add_tool( + create_artifact_list_tool(launch_directory=launch_directory), + name=tool_name, + description=_TOOL_DESCRIPTION, + annotations=ToolAnnotations( + readOnlyHint=True, + idempotentHint=True, + openWorldHint=False, + ), + structured_output=True, + ) + _forbid_unexpected_arguments(server, tool_name) diff --git a/src/specify_cli/mcp_server/server.py b/src/specify_cli/mcp_server/server.py index 13d1a9e023..c7b5b6dae9 100644 --- a/src/specify_cli/mcp_server/server.py +++ b/src/specify_cli/mcp_server/server.py @@ -3,10 +3,12 @@ from __future__ import annotations from collections.abc import Callable +from pathlib import Path from mcp.server import MCPServer from mcp.types import CallToolResult, TextContent +from ..artifacts._mcp import register as register_artifacts from ..mcp_version import register as register_version from .catalog import ( CommandAdapterError, @@ -37,8 +39,15 @@ def _tool_error(exc: CommandAdapterError) -> CallToolResult: def create_server( *, command_runner: CommandRunner = run_command, + launch_directory: Path | None = None, ) -> MCPServer: - """Create the experimental version-only MCP server.""" + """Create the experimental local stdio MCP server.""" + server_launch_directory = ( + Path.cwd() if launch_directory is None else Path(launch_directory) + ) + if not server_launch_directory.is_absolute(): + raise ValueError("MCP server launch directory must be absolute") + server = MCPServer( name="specify", title="Spec Kit CLI", @@ -79,6 +88,7 @@ def specify_run_command(command: str) -> VersionResult: return _tool_error(exc) register_version(server) + register_artifacts(server, launch_directory=server_launch_directory) return server diff --git a/tests/specify_cli/artifacts/test_mcp_list.py b/tests/specify_cli/artifacts/test_mcp_list.py new file mode 100644 index 0000000000..1def92bbd4 --- /dev/null +++ b/tests/specify_cli/artifacts/test_mcp_list.py @@ -0,0 +1,772 @@ +"""Tests for the first-class ``specify_artifact_list`` MCP adapter.""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path +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.artifacts import _commands, _mcp, _operation_list +from specify_cli.artifacts._operation_list import ( + ARTIFACT_LIST_OPERATION, + ArtifactListRequest, + ArtifactListResult, +) +from specify_cli.artifacts.mcp_list import ( + ArtifactListToolResult, + create_artifact_list_tool, +) +from specify_cli.mcp_server.server import create_server +from specify_cli.presets import PresetError + +TOOL_NAME = "specify_artifact_list" +TOOL_DESCRIPTION = ( + "List every command, template, script, and hook Spec Kit exposes for a project." +) + +ARTIFACT_ROWS = ( + { + "id": "template:réview", + "name": "réview", + "kind": "template", + "description": "Réview checklist", + "stack": [ + { + "id": "template:réview", + "layer": "extension", + "sourceId": "quality", + "presetId": None, + "presetName": None, + "strategy": "append", + "active": True, + "hidden": False, + "manifestPath": ".specify/extensions/quality/extension.yml", + "lookupId": "extension:quality:template:réview", + "sourcePath": ( + ".specify/extensions/quality/templates/réview-checklist.md" + ), + } + ], + }, + { + "id": "hook:before_plan:quality", + "name": "quality", + "kind": "hook", + "description": "Validate quality", + "eventName": "before_plan", + "targetCommand": "speckit.plan", + "registered": True, + "stack": [ + { + "id": "hook:before_plan:quality", + "layer": "extension", + "sourceId": "quality", + "presetId": None, + "presetName": None, + "strategy": "additive", + "active": True, + "hidden": False, + "manifestPath": ".specify/extensions/quality/extension.yml", + "lookupId": "extension:quality:hook:before_plan:quality", + "sourcePath": None, + "priority": 10, + "optional": False, + } + ], + }, +) + +ARTIFACT_PAYLOAD = {"rows": list(ARTIFACT_ROWS)} + +EXPECTED_OUTPUT_SCHEMA = { + "$defs": { + "ArtifactListHookRowResult": { + "additionalProperties": False, + "description": "One hook inventory row.", + "properties": { + "id": {"title": "Id", "type": "string"}, + "name": {"title": "Name", "type": "string"}, + "kind": {"const": "hook", "title": "Kind", "type": "string"}, + "description": {"title": "Description", "type": "string"}, + "eventName": {"title": "Eventname", "type": "string"}, + "targetCommand": {"title": "Targetcommand", "type": "string"}, + "registered": {"title": "Registered", "type": "boolean"}, + "stack": { + "items": {"$ref": "#/$defs/ArtifactListHookStackEntryResult"}, + "title": "Stack", + "type": "array", + }, + }, + "required": [ + "id", + "name", + "kind", + "description", + "eventName", + "targetCommand", + "registered", + "stack", + ], + "title": "ArtifactListHookRowResult", + "type": "object", + }, + "ArtifactListHookStackEntryResult": { + "additionalProperties": False, + "description": "Composition entry for a hook artifact.", + "properties": { + "id": {"title": "Id", "type": "string"}, + "layer": { + "enum": ["preset", "extension"], + "title": "Layer", + "type": "string", + }, + "sourceId": {"title": "Sourceid", "type": "string"}, + "presetId": { + "anyOf": [{"type": "string"}, {"type": "null"}], + "title": "Presetid", + }, + "presetName": { + "anyOf": [{"type": "string"}, {"type": "null"}], + "title": "Presetname", + }, + "strategy": { + "const": "additive", + "title": "Strategy", + "type": "string", + }, + "active": {"title": "Active", "type": "boolean"}, + "hidden": {"title": "Hidden", "type": "boolean"}, + "manifestPath": {"title": "Manifestpath", "type": "string"}, + "lookupId": {"title": "Lookupid", "type": "string"}, + "sourcePath": {"title": "Sourcepath", "type": "null"}, + "priority": {"title": "Priority", "type": "integer"}, + "optional": {"title": "Optional", "type": "boolean"}, + }, + "required": [ + "id", + "layer", + "sourceId", + "presetId", + "presetName", + "strategy", + "active", + "hidden", + "manifestPath", + "lookupId", + "sourcePath", + "priority", + "optional", + ], + "title": "ArtifactListHookStackEntryResult", + "type": "object", + }, + "ArtifactListRowResult": { + "additionalProperties": False, + "description": "One named artifact inventory row.", + "properties": { + "id": {"title": "Id", "type": "string"}, + "name": {"title": "Name", "type": "string"}, + "kind": { + "enum": ["command", "template", "script"], + "title": "Kind", + "type": "string", + }, + "description": {"title": "Description", "type": "string"}, + "stack": { + "items": {"$ref": "#/$defs/ArtifactListStackEntryResult"}, + "title": "Stack", + "type": "array", + }, + }, + "required": ["id", "name", "kind", "description", "stack"], + "title": "ArtifactListRowResult", + "type": "object", + }, + "ArtifactListStackEntryResult": { + "additionalProperties": False, + "description": "Composition entry for a named artifact.", + "properties": { + "id": {"title": "Id", "type": "string"}, + "layer": { + "anyOf": [ + { + "enum": ["project", "preset", "extension"], + "type": "string", + }, + {"type": "null"}, + ], + "title": "Layer", + }, + "sourceId": { + "anyOf": [{"type": "string"}, {"type": "null"}], + "title": "Sourceid", + }, + "presetId": { + "anyOf": [{"type": "string"}, {"type": "null"}], + "title": "Presetid", + }, + "presetName": { + "anyOf": [{"type": "string"}, {"type": "null"}], + "title": "Presetname", + }, + "strategy": { + "enum": ["replace", "wrap", "prepend", "append"], + "title": "Strategy", + "type": "string", + }, + "active": {"title": "Active", "type": "boolean"}, + "hidden": {"title": "Hidden", "type": "boolean"}, + "manifestPath": { + "anyOf": [{"type": "string"}, {"type": "null"}], + "title": "Manifestpath", + }, + "lookupId": { + "anyOf": [{"type": "string"}, {"type": "null"}], + "title": "Lookupid", + }, + "sourcePath": { + "anyOf": [{"type": "string"}, {"type": "null"}], + "title": "Sourcepath", + }, + }, + "required": [ + "id", + "layer", + "sourceId", + "presetId", + "presetName", + "strategy", + "active", + "hidden", + "manifestPath", + "lookupId", + "sourcePath", + ], + "title": "ArtifactListStackEntryResult", + "type": "object", + }, + }, + "additionalProperties": False, + "description": "Typed structured result returned by ``specify_artifact_list``.", + "properties": { + "rows": { + "items": { + "discriminator": { + "mapping": { + "command": "#/$defs/ArtifactListRowResult", + "hook": "#/$defs/ArtifactListHookRowResult", + "script": "#/$defs/ArtifactListRowResult", + "template": "#/$defs/ArtifactListRowResult", + }, + "propertyName": "kind", + }, + "oneOf": [ + {"$ref": "#/$defs/ArtifactListRowResult"}, + {"$ref": "#/$defs/ArtifactListHookRowResult"}, + ], + }, + "title": "Rows", + "type": "array", + } + }, + "required": ["rows"], + "title": "ArtifactListToolResult", + "type": "object", +} + + +def _run(coro): + return asyncio.run(coro) + + +def _expected_error( + code: str, + message: str, + *, + details: dict[str, object] | None = None, +) -> dict[str, object]: + return { + "error": { + "code": code, + "message": message, + "details": details or {}, + "retryable": False, + } + } + + +def test_artifact_list_dispatches_directly_to_shared_operation( + spec_kit_project: Path, +): + command_runner = Mock(side_effect=AssertionError("subprocess executor reached")) + + with ( + patch( + "specify_cli.artifacts.mcp_list.list_artifacts", + return_value=ArtifactListResult(rows=ARTIFACT_ROWS), + ) as operation, + patch( + "specify_cli.mcp_server.executor._run_cli_process", + side_effect=AssertionError("subprocess executor reached"), + ) as run_cli_process, + patch( + "specify_cli.artifacts.command_list.artifact_list", + side_effect=AssertionError("CLI adapter reached"), + ) as cli_adapter, + ): + server = create_server( + command_runner=command_runner, + launch_directory=spec_kit_project, + ) + result = _run(server.call_tool(TOOL_NAME, {})) + + assert result.structured_content == ARTIFACT_PAYLOAD + request = operation.call_args.args[0] + assert request == ArtifactListRequest(project_directory=spec_kit_project) + command_runner.assert_not_called() + run_cli_process.assert_not_called() + cli_adapter.assert_not_called() + + +def test_artifact_list_preserves_complete_typed_rows_unicode_and_order( + spec_kit_project: Path, +): + tool = create_artifact_list_tool(launch_directory=spec_kit_project) + with patch( + "specify_cli.artifacts.mcp_list.list_artifacts", + return_value=ArtifactListResult(rows=ARTIFACT_ROWS), + ): + result = tool() + + assert isinstance(result, ArtifactListToolResult) + assert result.model_dump() == ARTIFACT_PAYLOAD + assert [row.id for row in result.rows] == [ + "template:réview", + "hook:before_plan:quality", + ] + assert result.rows[0].stack[0].sourcePath.endswith("réview-checklist.md") + assert result.rows[1].stack[0].sourcePath is None + + +def test_artifact_list_uses_server_launch_directory_by_default( + spec_kit_project: Path, + non_project: Path, + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.chdir(spec_kit_project) + server = create_server() + monkeypatch.chdir(non_project) + operation = Mock(return_value=ArtifactListResult(rows=())) + with patch("specify_cli.artifacts.mcp_list.list_artifacts", operation): + result = _run(server.call_tool(TOOL_NAME, {})) + + assert result.structured_content == {"rows": []} + assert operation.call_args.args[0].project_directory == spec_kit_project + + +def test_artifact_list_accepts_explicit_absolute_project_directory( + spec_kit_project: Path, + non_project: Path, +): + operation = Mock(return_value=ArtifactListResult(rows=())) + with patch("specify_cli.artifacts.mcp_list.list_artifacts", operation): + result = _run( + create_server(launch_directory=non_project).call_tool( + TOOL_NAME, + {"project_directory": str(spec_kit_project)}, + ) + ) + + assert result.structured_content == {"rows": []} + assert operation.call_args.args[0].project_directory == spec_kit_project + + +def test_artifact_list_isolates_sequential_project_contexts( + spec_kit_project: Path, + non_project: Path, +): + invocation_cwd = Path.cwd() + + def operation(request: ArtifactListRequest) -> ArtifactListResult: + return ArtifactListResult( + rows=( + { + **ARTIFACT_ROWS[0], + "id": f"template:{request.project_directory.name}", + "name": request.project_directory.name, + }, + ) + ) + + with patch("specify_cli.artifacts.mcp_list.list_artifacts", operation): + server = create_server(launch_directory=spec_kit_project) + first = _run( + server.call_tool( + TOOL_NAME, + {"project_directory": str(spec_kit_project)}, + ) + ) + second = _run( + server.call_tool( + TOOL_NAME, + {"project_directory": str(non_project)}, + ) + ) + + assert first.structured_content["rows"][0]["name"] == spec_kit_project.name + assert second.structured_content["rows"][0]["name"] == non_project.name + assert Path.cwd() == invocation_cwd + + +def test_artifact_list_discovery_has_exact_contract( + spec_kit_project: Path, +): + tools = _run(create_server(launch_directory=spec_kit_project).list_tools()) + tool = next(tool for tool in tools if tool.name == TOOL_NAME) + + assert tool.name == TOOL_NAME + assert tool.description == TOOL_DESCRIPTION + assert tool.input_schema == { + "additionalProperties": False, + "properties": { + "project_directory": { + "anyOf": [{"type": "string"}, {"type": "null"}], + "default": None, + "title": "Project Directory", + } + }, + "title": "specify_artifact_listArguments", + "type": "object", + } + assert tool.output_schema == EXPECTED_OUTPUT_SCHEMA + 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_artifact_list_rejects_unknown_arguments_before_dispatch( + spec_kit_project: Path, +): + operation = Mock(side_effect=AssertionError("operation reached")) + with ( + patch("specify_cli.artifacts.mcp_list.list_artifacts", operation), + pytest.raises(ToolError, match="Extra inputs are not permitted"), + ): + _run( + create_server(launch_directory=spec_kit_project).call_tool( + TOOL_NAME, + {"unexpected": True}, + ) + ) + + operation.assert_not_called() + + +def test_artifact_list_rejects_invalid_project_directory_type_before_dispatch( + spec_kit_project: Path, +): + operation = Mock(side_effect=AssertionError("operation reached")) + with ( + patch("specify_cli.artifacts.mcp_list.list_artifacts", operation), + pytest.raises(ToolError, match="Input should be a valid string"), + ): + _run( + create_server(launch_directory=spec_kit_project).call_tool( + TOOL_NAME, + {"project_directory": 123}, + ) + ) + + operation.assert_not_called() + + +def test_artifact_list_rejects_relative_project_directory( + spec_kit_project: Path, +): + result = _run( + create_server(launch_directory=spec_kit_project).call_tool( + TOOL_NAME, + {"project_directory": "relative-project"}, + ) + ) + + assert result.is_error is True + assert result.structured_content == _expected_error( + "not_a_spec_kit_project", + "not a Spec Kit project: no .specify/ directory found", + details={"project_directory": "relative-project"}, + ) + + +def test_artifact_list_maps_non_project_error(non_project: Path): + result = _run(create_server(launch_directory=non_project).call_tool(TOOL_NAME, {})) + + assert result.is_error is True + assert result.structured_content == _expected_error( + "not_a_spec_kit_project", + "not a Spec Kit project: no .specify/ directory found", + details={"project_directory": str(non_project)}, + ) + assert result.content[0].text == ( + "not_a_spec_kit_project: not a Spec Kit project: no .specify/ directory found" + ) + + +def test_artifact_list_maps_corrupt_registry_error(spec_kit_project: Path): + registry = spec_kit_project / ".specify" / "extensions" / ".registry" + registry.write_text("{invalid", encoding="utf-8") + + result = _run( + create_server(launch_directory=spec_kit_project).call_tool(TOOL_NAME, {}) + ) + + assert result.is_error is True + assert result.structured_content == _expected_error( + "artifact_resolution_failed", + "artifact resolution failed", + details={"project_directory": str(spec_kit_project)}, + ) + + +@pytest.mark.parametrize( + "operation_failure", + [ + OSError("unreadable"), + PresetError("broken preset"), + ], +) +def test_artifact_list_preserves_operation_owned_resolution_mapping( + spec_kit_project: Path, + monkeypatch: pytest.MonkeyPatch, + operation_failure: Exception, +): + catalog = Mock() + catalog.list_artifacts_with_stack.side_effect = operation_failure + monkeypatch.setattr( + _operation_list, + "ArtifactCatalog", + Mock(return_value=catalog), + ) + + result = _run( + create_server(launch_directory=spec_kit_project).call_tool(TOOL_NAME, {}) + ) + + assert result.is_error is True + assert result.structured_content == _expected_error( + "artifact_resolution_failed", + "artifact resolution failed", + details={"project_directory": str(spec_kit_project)}, + ) + + +@pytest.mark.parametrize( + "operation_result", + [ + None, + ArtifactListResult(rows=None), + ArtifactListResult( + rows=( + { + "id": "template:incomplete", + "name": "incomplete", + "kind": "template", + "description": "Missing stack", + }, + ) + ), + ArtifactListResult( + rows=( + { + **ARTIFACT_ROWS[0], + "unexpected": True, + }, + ) + ), + ], +) +def test_artifact_list_rejects_invalid_operation_results( + spec_kit_project: Path, + operation_result: object, +): + with patch( + "specify_cli.artifacts.mcp_list.list_artifacts", + return_value=operation_result, + ): + result = _run( + create_server(launch_directory=spec_kit_project).call_tool(TOOL_NAME, {}) + ) + + assert result.is_error is True + assert result.structured_content == _expected_error( + "invalid_operation_result", + "The artifact list operation returned an invalid result.", + ) + + +def test_artifact_list_sanitizes_and_logs_unexpected_failure( + spec_kit_project: Path, + caplog: pytest.LogCaptureFixture, +): + unsafe = "SECRET_TOKEN=do-not-print /Users/example/private/project" + with ( + patch( + "specify_cli.artifacts.mcp_list.list_artifacts", + side_effect=RuntimeError(unsafe), + ), + caplog.at_level(logging.ERROR, logger="specify_cli.artifacts.mcp_list"), + ): + result = _run( + create_server(launch_directory=spec_kit_project).call_tool(TOOL_NAME, {}) + ) + + assert result.is_error is True + assert result.structured_content == _expected_error( + "internal_error", + "Unable to list Spec Kit artifacts.", + ) + 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 artifact list MCP adapter." + ) + assert caplog.records[0].exc_info is not None + + +def test_artifact_inventory_matches_cli_leaves_and_operation_contract( + spec_kit_project: Path, +): + cli_operation_ids = { + f"artifact.{command.name}" + for command in _commands.artifact_app.registered_commands + } + inventory_operation_ids = [item.operation_id for item in _mcp.ARTIFACT_TOOLS] + inventory_tool_names = [item.mcp_tool_name for item in _mcp.ARTIFACT_TOOLS] + registered_tool_names = { + tool.name + for tool in _run(create_server(launch_directory=spec_kit_project).list_tools()) + } + + assert set(inventory_operation_ids) == cli_operation_ids + assert len(inventory_operation_ids) == len(set(inventory_operation_ids)) + assert len(inventory_tool_names) == len(set(inventory_tool_names)) + assert all( + item.cli_path == f"specify {item.operation_id.replace('.', ' ')}" + for item in _mcp.ARTIFACT_TOOLS + ) + assert all( + item.mcp_tool_name == f"specify_{item.operation_id.replace('.', '_')}" + for item in _mcp.ARTIFACT_TOOLS + ) + + list_tool = next( + item + for item in _mcp.ARTIFACT_TOOLS + if item.operation_id == ARTIFACT_LIST_OPERATION.operation_id + ) + assert list_tool.contract_version == ARTIFACT_LIST_OPERATION.contract_version + assert list_tool.disposition == "available" + assert list_tool.disposition_reason is None + assert list_tool.capabilities == ARTIFACT_LIST_OPERATION.capabilities + assert list_tool.network_access == ARTIFACT_LIST_OPERATION.network_access + assert list_tool.mcp_tool_name in registered_tool_names + + unavailable = [ + item for item in _mcp.ARTIFACT_TOOLS if item.disposition == "unavailable" + ] + assert {item.operation_id for item in unavailable} == { + "artifact.info", + "artifact.lookup", + } + assert all(item.contract_version is None for item in unavailable) + assert all(item.disposition_reason for item in unavailable) + assert all(item.mcp_tool_name not in registered_tool_names for item in unavailable) + + +def test_artifact_registration_adds_available_tool_once_and_rejects_collision( + spec_kit_project: Path, +): + server = MCPServer(name="test") + + _mcp.register(server, launch_directory=spec_kit_project) + + available_names = [ + item.mcp_tool_name + for item in _mcp.ARTIFACT_TOOLS + if item.disposition == "available" + ] + assert [tool.name for tool in _run(server.list_tools())] == available_names + with pytest.raises(ValueError, match=f"MCP tool name collision: {TOOL_NAME}"): + _mcp.register(server, launch_directory=spec_kit_project) + assert [tool.name for tool in _run(server.list_tools())] == available_names + + +def test_artifact_tool_requires_absolute_server_launch_directory(): + with pytest.raises( + ValueError, + match="MCP server launch directory must be absolute", + ): + create_artifact_list_tool(launch_directory=Path("relative")) + + +def test_in_memory_client_preserves_artifact_success_and_failure_wire_shapes( + spec_kit_project: Path, +): + async def exercise(): + server = create_server(launch_directory=spec_kit_project) + 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() + tools = await session.list_tools() + with patch( + "specify_cli.artifacts.mcp_list.list_artifacts", + return_value=ArtifactListResult(rows=ARTIFACT_ROWS), + ): + success = await session.call_tool(TOOL_NAME, {}) + with patch( + "specify_cli.artifacts.mcp_list.list_artifacts", + return_value=None, + ): + failure = await session.call_tool(TOOL_NAME, {}) + schema_failure = await session.call_tool( + TOOL_NAME, + {"unexpected": True}, + ) + task_group.cancel_scope.cancel() + return tools, success, failure, schema_failure + + tools, success, failure, schema_failure = _run(exercise()) + + discovered = next(tool for tool in tools.tools if tool.name == TOOL_NAME) + assert discovered.input_schema["additionalProperties"] is False + assert success.is_error is False + assert success.structured_content == ARTIFACT_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/mcp_server/test_server.py b/tests/specify_cli/mcp_server/test_server.py index ffe61511ec..88808ec787 100644 --- a/tests/specify_cli/mcp_server/test_server.py +++ b/tests/specify_cli/mcp_server/test_server.py @@ -23,19 +23,22 @@ def _run(coro): return asyncio.run(coro) -def test_tool_discovery_exposes_first_class_and_transitional_tools(): - tools = _run(create_server().list_tools()) +def test_tool_discovery_exposes_first_class_and_transitional_tools(tmp_path): + tools = _run(create_server(launch_directory=tmp_path).list_tools()) assert [tool.name for tool in tools] == [ "specify_list_commands", "specify_describe_command", "specify_run_command", "specify_version", + "specify_artifact_list", ] 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 + assert schemas["specify_artifact_list"]["additionalProperties"] is False + assert set(schemas["specify_artifact_list"]["properties"]) == {"project_directory"} 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 b3ca22f05d..606126a5e4 100644 --- a/tests/specify_cli/mcp_server/test_stdio.py +++ b/tests/specify_cli/mcp_server/test_stdio.py @@ -11,7 +11,7 @@ _TEST_TIMEOUT_SECONDS = 30 -def test_real_stdio_server_initializes_discovers_and_runs_version(): +def test_real_stdio_server_initializes_discovers_and_calls_first_class_tools(): async def exercise() -> None: repo_root = Path(__file__).resolve().parents[3] parameters = StdioServerParameters( @@ -34,6 +34,11 @@ async def exercise() -> None: tools = await session.list_tools() listed = await session.call_tool("specify_list_commands", {}) version = await session.call_tool("specify_version", {}) + artifacts = await session.call_tool("specify_artifact_list", {}) + artifact_failure = await session.call_tool( + "specify_artifact_list", + {"project_directory": str(repo_root / "tests")}, + ) ran = await session.call_tool( "specify_run_command", {"command": "version"}, @@ -51,6 +56,7 @@ async def exercise() -> None: "specify_describe_command", "specify_run_command", "specify_version", + "specify_artifact_list", ] assert listed.structured_content["commands"][0]["command"] == "version" assert set(version.structured_content) == { @@ -60,6 +66,18 @@ async def exercise() -> None: "features", } assert version.is_error is False + assert artifacts.is_error is False + assert isinstance(artifacts.structured_content["rows"], list) + assert artifacts.structured_content["rows"] + assert artifact_failure.is_error is True + assert artifact_failure.structured_content == { + "error": { + "code": "not_a_spec_kit_project", + "message": "not a Spec Kit project: no .specify/ directory found", + "details": {"project_directory": str(repo_root / "tests")}, + "retryable": False, + } + } assert set(ran.structured_content) == { "cli_version", "runtime", From 280bf593fbb522e3d69024528477044b9d27afc7 Mon Sep 17 00:00:00 2001 From: Manfred Riem <15701806+mnriem@users.noreply.github.com> Date: Wed, 7 Oct 2026 10:50:10 -0500 Subject: [PATCH 2/5] fix(mcp): bound artifact list responses Assisted-by: GitHub Copilot (model: GPT-5.6 Sol, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/specify_cli/artifacts/_operation_list.py | 73 ++++++- src/specify_cli/artifacts/mcp_list.py | 84 ++++++++- tests/specify_cli/artifacts/test_mcp_list.py | 178 +++++++++++++++++- .../artifacts/test_operation_list.py | 89 +++++++++ tests/specify_cli/mcp_server/test_server.py | 6 +- tests/specify_cli/mcp_server/test_stdio.py | 2 + 6 files changed, 411 insertions(+), 21 deletions(-) diff --git a/src/specify_cli/artifacts/_operation_list.py b/src/specify_cli/artifacts/_operation_list.py index 55cf4beb51..4d1a319acf 100644 --- a/src/specify_cli/artifacts/_operation_list.py +++ b/src/specify_cli/artifacts/_operation_list.py @@ -73,6 +73,8 @@ class ArtifactListHookRow(TypedDict): ArtifactListItem = ArtifactListRow | ArtifactListHookRow +ARTIFACT_LIST_MAX_LIMIT = 1000 +ARTIFACT_LIST_CURSOR_MAX_LENGTH = 20 @dataclass(frozen=True) @@ -80,6 +82,8 @@ class ArtifactListRequest: """Explicit project context for the ``artifact.list`` operation.""" project_directory: Path + limit: int | None = None + cursor: str | None = None @dataclass(frozen=True) @@ -87,6 +91,8 @@ class ArtifactListResult: """Transport-neutral artifact inventory.""" rows: tuple[ArtifactListItem, ...] + next_cursor: str | None = None + truncated: bool = False class ArtifactListError(ArtifactError): @@ -133,6 +139,27 @@ 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``.""" @@ -153,17 +180,47 @@ class ArtifactListOperationDescriptor: request_type=ArtifactListRequest, result_type=ArtifactListResult, warning_types=(), - error_types=(ArtifactListProjectError, ArtifactListResolutionError), + error_types=( + ArtifactListProjectError, + ArtifactListResolutionError, + ArtifactListPaginationError, + ), 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) + offset = _pagination_offset(request) try: rows = ArtifactCatalog(project_directory).list_artifacts_with_stack() @@ -172,6 +229,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, ) diff --git a/src/specify_cli/artifacts/mcp_list.py b/src/specify_cli/artifacts/mcp_list.py index e067057e2c..3a84de7c39 100644 --- a/src/specify_cli/artifacts/mcp_list.py +++ b/src/specify_cli/artifacts/mcp_list.py @@ -9,9 +9,11 @@ from mcp.server import MCPServer from mcp.types import CallToolResult, TextContent, ToolAnnotations -from pydantic import BaseModel, ConfigDict, Field, ValidationError +from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator from ._operation_list import ( + ARTIFACT_LIST_CURSOR_MAX_LENGTH, + ARTIFACT_LIST_MAX_LIMIT, ArtifactListError, ArtifactListRequest, ArtifactListResult, @@ -19,12 +21,29 @@ ) logger = logging.getLogger(__name__) +_DEFAULT_LIMIT = 100 +_MAX_RESPONSE_BYTES = 1024 * 1024 _INVALID_RESULT_MESSAGE = "The artifact list operation returned an invalid result." _INTERNAL_ERROR_MESSAGE = "Unable to list Spec Kit artifacts." +_RESPONSE_TOO_LARGE_MESSAGE = ( + "The artifact list result exceeds the MCP response-size limit." +) _TOOL_DESCRIPTION = ( "List every command, template, script, and hook Spec Kit exposes for a project." ) +ArtifactListLimit = Annotated[ + int, + Field(ge=1, le=ARTIFACT_LIST_MAX_LIMIT), +] +ArtifactListCursor = Annotated[ + str, + Field( + max_length=ARTIFACT_LIST_CURSOR_MAX_LENGTH, + pattern=r"^(0|[1-9][0-9]*)$", + ), +] + class ArtifactListToolInput(BaseModel): """Typed input accepted by ``specify_artifact_list``.""" @@ -32,6 +51,8 @@ class ArtifactListToolInput(BaseModel): model_config = ConfigDict(extra="forbid") project_directory: str | None = None + limit: ArtifactListLimit = _DEFAULT_LIMIT + cursor: ArtifactListCursor | None = None class ArtifactListStackEntryResult(BaseModel): @@ -111,13 +132,29 @@ class ArtifactListToolResult(BaseModel): model_config = ConfigDict(extra="forbid") rows: list[ArtifactListItemResult] + next_cursor: ArtifactListCursor | None + truncated: bool + + @model_validator(mode="after") + def validate_continuation(self) -> ArtifactListToolResult: + """Require continuation metadata to agree with truncation state.""" + if self.truncated != (self.next_cursor is not None): + raise ValueError("inconsistent artifact list continuation metadata") + return self class _InvalidOperationResult(Exception): """The shared operation returned an incomplete or invalid typed result.""" -ArtifactListTool = Callable[[str | None], ArtifactListToolResult | CallToolResult] +class _ResponseTooLarge(Exception): + """The serialized structured result exceeds the MCP response budget.""" + + +ArtifactListTool = Callable[ + [str | None, ArtifactListLimit, ArtifactListCursor | None], + ArtifactListToolResult | CallToolResult, +] def _tool_error( @@ -150,25 +187,41 @@ def _convert_result(result: ArtifactListResult) -> ArtifactListToolResult: try: return ArtifactListToolResult.model_validate( - {"rows": list(rows)}, + { + "rows": list(rows), + "next_cursor": result.next_cursor, + "truncated": result.truncated, + }, strict=True, ) except (TypeError, ValidationError) as exc: raise _InvalidOperationResult from exc -def create_artifact_list_tool(*, launch_directory: Path) -> ArtifactListTool: +def create_artifact_list_tool( + *, + launch_directory: Path, + max_response_bytes: int = _MAX_RESPONSE_BYTES, +) -> ArtifactListTool: """Create a tool bound to the immutable MCP server launch directory.""" launch_directory = Path(launch_directory) if not launch_directory.is_absolute(): raise ValueError("MCP server launch directory must be absolute") + if max_response_bytes < 1: + raise ValueError("MCP response-size limit must be positive") def specify_artifact_list( project_directory: str | None = None, + limit: ArtifactListLimit = _DEFAULT_LIMIT, + cursor: ArtifactListCursor | None = None, ) -> ArtifactListToolResult: """List every artifact exposed by the selected Spec Kit project.""" tool_input = ArtifactListToolInput.model_validate( - {"project_directory": project_directory}, + { + "project_directory": project_directory, + "limit": limit, + "cursor": cursor, + }, strict=True, ) selected_directory = ( @@ -177,11 +230,18 @@ def specify_artifact_list( else Path(tool_input.project_directory) ) try: - return _convert_result( + result = _convert_result( list_artifacts( - ArtifactListRequest(project_directory=selected_directory), + ArtifactListRequest( + project_directory=selected_directory, + limit=tool_input.limit, + cursor=tool_input.cursor, + ), ) ) + if len(result.model_dump_json().encode("utf-8")) > max_response_bytes: + raise _ResponseTooLarge + return result except ArtifactListError as exc: return _tool_error( exc.code, @@ -194,6 +254,16 @@ def specify_artifact_list( "invalid_operation_result", _INVALID_RESULT_MESSAGE, ) + except _ResponseTooLarge: + return _tool_error( + "response_too_large", + _RESPONSE_TOO_LARGE_MESSAGE, + details={ + "max_bytes": max_response_bytes, + "requested_limit": tool_input.limit, + }, + retryable=True, + ) except Exception: logger.exception("Unexpected failure in the artifact list MCP adapter.") return _tool_error("internal_error", _INTERNAL_ERROR_MESSAGE) diff --git a/tests/specify_cli/artifacts/test_mcp_list.py b/tests/specify_cli/artifacts/test_mcp_list.py index 1def92bbd4..c94f40b2ee 100644 --- a/tests/specify_cli/artifacts/test_mcp_list.py +++ b/tests/specify_cli/artifacts/test_mcp_list.py @@ -84,7 +84,11 @@ }, ) -ARTIFACT_PAYLOAD = {"rows": list(ARTIFACT_ROWS)} +ARTIFACT_PAYLOAD = { + "rows": list(ARTIFACT_ROWS), + "next_cursor": None, + "truncated": False, +} EXPECTED_OUTPUT_SCHEMA = { "$defs": { @@ -275,9 +279,24 @@ }, "title": "Rows", "type": "array", - } + }, + "next_cursor": { + "anyOf": [ + { + "maxLength": 20, + "pattern": "^(0|[1-9][0-9]*)$", + "type": "string", + }, + {"type": "null"}, + ], + "title": "Next Cursor", + }, + "truncated": { + "title": "Truncated", + "type": "boolean", + }, }, - "required": ["rows"], + "required": ["rows", "next_cursor", "truncated"], "title": "ArtifactListToolResult", "type": "object", } @@ -292,13 +311,14 @@ def _expected_error( message: str, *, details: dict[str, object] | None = None, + retryable: bool = False, ) -> dict[str, object]: return { "error": { "code": code, "message": message, "details": details or {}, - "retryable": False, + "retryable": retryable, } } @@ -330,7 +350,11 @@ def test_artifact_list_dispatches_directly_to_shared_operation( assert result.structured_content == ARTIFACT_PAYLOAD request = operation.call_args.args[0] - assert request == ArtifactListRequest(project_directory=spec_kit_project) + assert request == ArtifactListRequest( + project_directory=spec_kit_project, + limit=100, + cursor=None, + ) command_runner.assert_not_called() run_cli_process.assert_not_called() cli_adapter.assert_not_called() @@ -368,8 +392,16 @@ def test_artifact_list_uses_server_launch_directory_by_default( with patch("specify_cli.artifacts.mcp_list.list_artifacts", operation): result = _run(server.call_tool(TOOL_NAME, {})) - assert result.structured_content == {"rows": []} - assert operation.call_args.args[0].project_directory == spec_kit_project + assert result.structured_content == { + "rows": [], + "next_cursor": None, + "truncated": False, + } + assert operation.call_args.args[0] == ArtifactListRequest( + project_directory=spec_kit_project, + limit=100, + cursor=None, + ) def test_artifact_list_accepts_explicit_absolute_project_directory( @@ -385,8 +417,72 @@ def test_artifact_list_accepts_explicit_absolute_project_directory( ) ) - assert result.structured_content == {"rows": []} - assert operation.call_args.args[0].project_directory == spec_kit_project + assert result.structured_content == { + "rows": [], + "next_cursor": None, + "truncated": False, + } + assert operation.call_args.args[0] == ArtifactListRequest( + project_directory=spec_kit_project, + limit=100, + cursor=None, + ) + + +def test_artifact_list_forwards_pagination_and_returns_continuation( + spec_kit_project: Path, +): + operation = Mock( + return_value=ArtifactListResult( + rows=(ARTIFACT_ROWS[1],), + next_cursor="7", + truncated=True, + ) + ) + with patch("specify_cli.artifacts.mcp_list.list_artifacts", operation): + result = _run( + create_server(launch_directory=spec_kit_project).call_tool( + TOOL_NAME, + {"limit": 2, "cursor": "5"}, + ) + ) + + assert result.structured_content == { + "rows": [ARTIFACT_ROWS[1]], + "next_cursor": "7", + "truncated": True, + } + assert operation.call_args.args[0] == ArtifactListRequest( + project_directory=spec_kit_project, + limit=2, + cursor="5", + ) + + +def test_artifact_list_rejects_oversized_structured_page( + spec_kit_project: Path, +): + tool = create_artifact_list_tool( + launch_directory=spec_kit_project, + max_response_bytes=64, + ) + with patch( + "specify_cli.artifacts.mcp_list.list_artifacts", + return_value=ArtifactListResult(rows=ARTIFACT_ROWS), + ): + result = tool() + + assert result.is_error is True + assert result.structured_content == _expected_error( + "response_too_large", + "The artifact list result exceeds the MCP response-size limit.", + details={"max_bytes": 64, "requested_limit": 100}, + retryable=True, + ) + assert result.content[0].text == ( + "response_too_large: " + "The artifact list result exceeds the MCP response-size limit." + ) def test_artifact_list_isolates_sequential_project_contexts( @@ -441,7 +537,26 @@ def test_artifact_list_discovery_has_exact_contract( "anyOf": [{"type": "string"}, {"type": "null"}], "default": None, "title": "Project Directory", - } + }, + "limit": { + "default": 100, + "maximum": 1000, + "minimum": 1, + "title": "Limit", + "type": "integer", + }, + "cursor": { + "anyOf": [ + { + "maxLength": 20, + "pattern": "^(0|[1-9][0-9]*)$", + "type": "string", + }, + {"type": "null"}, + ], + "default": None, + "title": "Cursor", + }, }, "title": "specify_artifact_listArguments", "type": "object", @@ -489,6 +604,33 @@ def test_artifact_list_rejects_invalid_project_directory_type_before_dispatch( operation.assert_not_called() +@pytest.mark.parametrize( + "arguments", + [ + {"limit": 0}, + {"limit": 1001}, + {"cursor": "01"}, + ], +) +def test_artifact_list_rejects_invalid_pagination_before_dispatch( + spec_kit_project: Path, + arguments: dict[str, object], +): + operation = Mock(side_effect=AssertionError("operation reached")) + with ( + patch("specify_cli.artifacts.mcp_list.list_artifacts", operation), + pytest.raises(ToolError), + ): + _run( + create_server(launch_directory=spec_kit_project).call_tool( + TOOL_NAME, + arguments, + ) + ) + + operation.assert_not_called() + + def test_artifact_list_rejects_relative_project_directory( spec_kit_project: Path, ): @@ -592,6 +734,9 @@ def test_artifact_list_preserves_operation_owned_resolution_mapping( }, ) ), + ArtifactListResult(rows=(), next_cursor=1, truncated=True), + ArtifactListResult(rows=(), next_cursor=None, truncated=True), + ArtifactListResult(rows=(), next_cursor="1", truncated=False), ], ) def test_artifact_list_rejects_invalid_operation_results( @@ -721,6 +866,19 @@ def test_artifact_tool_requires_absolute_server_launch_directory(): create_artifact_list_tool(launch_directory=Path("relative")) +def test_artifact_tool_requires_positive_response_size_limit( + spec_kit_project: Path, +): + with pytest.raises( + ValueError, + match="MCP response-size limit must be positive", + ): + create_artifact_list_tool( + launch_directory=spec_kit_project, + max_response_bytes=0, + ) + + def test_in_memory_client_preserves_artifact_success_and_failure_wire_shapes( spec_kit_project: Path, ): diff --git a/tests/specify_cli/artifacts/test_operation_list.py b/tests/specify_cli/artifacts/test_operation_list.py index c898a53dc2..823e61de11 100644 --- a/tests/specify_cli/artifacts/test_operation_list.py +++ b/tests/specify_cli/artifacts/test_operation_list.py @@ -12,6 +12,8 @@ from specify_cli.artifacts import ArtifactCatalog, _operation_list from specify_cli.artifacts._operation_list import ( ARTIFACT_LIST_OPERATION, + ARTIFACT_LIST_MAX_LIMIT, + ArtifactListPaginationError, ArtifactListProjectError, ArtifactListRequest, ArtifactListResolutionError, @@ -31,6 +33,7 @@ def test_artifact_list_operation_descriptor_is_stable(): assert ARTIFACT_LIST_OPERATION.error_types == ( ArtifactListProjectError, ArtifactListResolutionError, + ArtifactListPaginationError, ) assert ARTIFACT_LIST_OPERATION.capabilities == frozenset({"local-read"}) assert ARTIFACT_LIST_OPERATION.network_access == "none" @@ -48,6 +51,8 @@ def test_list_artifacts_returns_complete_typed_inventory(spec_kit_project: Path) {"id", "name", "kind", "description", "stack"} <= row.keys() for row in result.rows ) + assert result.next_cursor is None + assert result.truncated is False def test_list_artifacts_preserves_empty_inventory( @@ -76,6 +81,90 @@ def test_list_artifacts_is_stable_across_repeated_calls(spec_kit_project: Path): assert [row["id"] for row in first.rows] == [row["id"] for row in second.rows] +def test_list_artifacts_returns_deterministic_pages( + spec_kit_project: Path, + monkeypatch: pytest.MonkeyPatch, +): + rows = [ + { + "id": f"template:item-{index}", + "name": f"item-{index}", + "kind": "template", + "description": f"Item {index}", + "stack": [], + } + for index in range(5) + ] + list_with_stack = Mock(return_value=rows) + monkeypatch.setattr( + _operation_list, + "ArtifactCatalog", + lambda _project_directory: Mock(list_artifacts_with_stack=list_with_stack), + ) + + first = list_artifacts(ArtifactListRequest(spec_kit_project, limit=2)) + second = list_artifacts( + ArtifactListRequest(spec_kit_project, limit=2, cursor=first.next_cursor) + ) + final = list_artifacts( + ArtifactListRequest(spec_kit_project, limit=2, cursor=second.next_cursor) + ) + + assert [row["id"] for row in first.rows] == [ + "template:item-0", + "template:item-1", + ] + assert first.next_cursor == "2" + assert first.truncated is True + assert [row["id"] for row in second.rows] == [ + "template:item-2", + "template:item-3", + ] + assert second.next_cursor == "4" + assert second.truncated is True + assert [row["id"] for row in final.rows] == ["template:item-4"] + assert final.next_cursor is None + assert final.truncated is False + + +@pytest.mark.parametrize("limit", [0, ARTIFACT_LIST_MAX_LIMIT + 1, True]) +def test_list_artifacts_rejects_invalid_limit( + spec_kit_project: Path, + limit: int, +): + with pytest.raises(ArtifactListPaginationError) as exc_info: + list_artifacts(ArtifactListRequest(spec_kit_project, limit=limit)) + + assert exc_info.value.code == "invalid_pagination" + assert exc_info.value.message == ( + f"artifact list limit must be between 1 and {ARTIFACT_LIST_MAX_LIMIT}" + ) + assert exc_info.value.details == { + "field": "limit", + "value": limit, + "max_limit": ARTIFACT_LIST_MAX_LIMIT, + } + + +@pytest.mark.parametrize("cursor", ["", "01", "-1", "1.5", "é", "1" * 21, 1]) +def test_list_artifacts_rejects_invalid_cursor( + spec_kit_project: Path, + cursor: object, +): + with pytest.raises(ArtifactListPaginationError) as exc_info: + list_artifacts(ArtifactListRequest(spec_kit_project, cursor=cursor)) + + assert exc_info.value.code == "invalid_pagination" + assert exc_info.value.message == ( + "artifact list cursor must be a canonical decimal offset" + ) + assert exc_info.value.details == { + "field": "cursor", + "value": cursor, + "max_limit": ARTIFACT_LIST_MAX_LIMIT, + } + + def test_list_artifacts_preserves_unicode_source_paths_and_stack( spec_kit_project: Path, ): diff --git a/tests/specify_cli/mcp_server/test_server.py b/tests/specify_cli/mcp_server/test_server.py index 88808ec787..0c1640a54d 100644 --- a/tests/specify_cli/mcp_server/test_server.py +++ b/tests/specify_cli/mcp_server/test_server.py @@ -38,7 +38,11 @@ def test_tool_discovery_exposes_first_class_and_transitional_tools(tmp_path): assert schemas["specify_version"]["properties"] == {} assert schemas["specify_version"]["additionalProperties"] is False assert schemas["specify_artifact_list"]["additionalProperties"] is False - assert set(schemas["specify_artifact_list"]["properties"]) == {"project_directory"} + assert set(schemas["specify_artifact_list"]["properties"]) == { + "project_directory", + "limit", + "cursor", + } 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 606126a5e4..3f9df4292b 100644 --- a/tests/specify_cli/mcp_server/test_stdio.py +++ b/tests/specify_cli/mcp_server/test_stdio.py @@ -69,6 +69,8 @@ async def exercise() -> None: assert artifacts.is_error is False assert isinstance(artifacts.structured_content["rows"], list) assert artifacts.structured_content["rows"] + assert artifacts.structured_content["next_cursor"] is None + assert artifacts.structured_content["truncated"] is False assert artifact_failure.is_error is True assert artifact_failure.structured_content == { "error": { From c18ef28824ed1ffd3a0fb3fb72571a4c22fe0906 Mon Sep 17 00:00:00 2001 From: Manfred Riem <15701806+mnriem@users.noreply.github.com> Date: Wed, 7 Oct 2026 13:53:07 -0500 Subject: [PATCH 3/5] fix(mcp): bound artifact list wire responses Assisted-by: GitHub Copilot (model: GPT-5.6 Sol, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/specify_cli/artifacts/mcp_list.py | 60 +++++++--- tests/specify_cli/artifacts/test_mcp_list.py | 112 ++++++++++++++++--- 2 files changed, 145 insertions(+), 27 deletions(-) diff --git a/src/specify_cli/artifacts/mcp_list.py b/src/specify_cli/artifacts/mcp_list.py index 3a84de7c39..db4da36414 100644 --- a/src/specify_cli/artifacts/mcp_list.py +++ b/src/specify_cli/artifacts/mcp_list.py @@ -22,14 +22,16 @@ logger = logging.getLogger(__name__) _DEFAULT_LIMIT = 100 -_MAX_RESPONSE_BYTES = 1024 * 1024 +_MAX_WIRE_RESPONSE_BYTES = 1024 * 1024 +_WIRE_ENVELOPE_RESERVE_BYTES = 1024 _INVALID_RESULT_MESSAGE = "The artifact list operation returned an invalid result." _INTERNAL_ERROR_MESSAGE = "Unable to list Spec Kit artifacts." _RESPONSE_TOO_LARGE_MESSAGE = ( "The artifact list result exceeds the MCP response-size limit." ) _TOOL_DESCRIPTION = ( - "List every command, template, script, and hook Spec Kit exposes for a project." + "Return one paginated artifact inventory page for a project. " + "While truncated is true, pass next_cursor as cursor to retrieve the next page." ) ArtifactListLimit = Annotated[ @@ -198,24 +200,48 @@ def _convert_result(result: ArtifactListResult) -> ArtifactListToolResult: raise _InvalidOperationResult from exc +def _build_success_result(result: ArtifactListToolResult) -> CallToolResult: + """Build the duplicated structured and compatibility content sent by MCP.""" + return CallToolResult( + content=[ + TextContent( + type="text", + text=result.model_dump_json(indent=2), + ) + ], + structuredContent=result.model_dump(mode="json"), + isError=False, + ) + + +def _wire_response_size(result: CallToolResult) -> int: + """Bound the serialized result plus conservative JSON-RPC envelope space.""" + result_bytes = len( + result.model_dump_json(by_alias=True, exclude_none=True).encode("utf-8") + ) + return result_bytes + _WIRE_ENVELOPE_RESERVE_BYTES + + def create_artifact_list_tool( *, launch_directory: Path, - max_response_bytes: int = _MAX_RESPONSE_BYTES, + max_wire_response_bytes: int = _MAX_WIRE_RESPONSE_BYTES, ) -> ArtifactListTool: """Create a tool bound to the immutable MCP server launch directory.""" launch_directory = Path(launch_directory) if not launch_directory.is_absolute(): raise ValueError("MCP server launch directory must be absolute") - if max_response_bytes < 1: - raise ValueError("MCP response-size limit must be positive") + if max_wire_response_bytes <= _WIRE_ENVELOPE_RESERVE_BYTES: + raise ValueError( + "MCP response-size limit must exceed the JSON-RPC envelope reserve" + ) def specify_artifact_list( project_directory: str | None = None, limit: ArtifactListLimit = _DEFAULT_LIMIT, cursor: ArtifactListCursor | None = None, ) -> ArtifactListToolResult: - """List every artifact exposed by the selected Spec Kit project.""" + """Return one page; follow next_cursor while truncated is true.""" tool_input = ArtifactListToolInput.model_validate( { "project_directory": project_directory, @@ -239,9 +265,10 @@ def specify_artifact_list( ), ) ) - if len(result.model_dump_json().encode("utf-8")) > max_response_bytes: + success = _build_success_result(result) + if _wire_response_size(success) > max_wire_response_bytes: raise _ResponseTooLarge - return result + return success except ArtifactListError as exc: return _tool_error( exc.code, @@ -259,7 +286,7 @@ def specify_artifact_list( "response_too_large", _RESPONSE_TOO_LARGE_MESSAGE, details={ - "max_bytes": max_response_bytes, + "max_bytes": max_wire_response_bytes, "requested_limit": tool_input.limit, }, retryable=True, @@ -271,15 +298,16 @@ def specify_artifact_list( return specify_artifact_list -def _forbid_unexpected_arguments(server: MCPServer, tool_name: str) -> None: +def _configure_strict_arguments(server: MCPServer, tool_name: str) -> None: tool = server._tool_manager.get_tool(tool_name) if tool is None: # pragma: no cover - registration immediately precedes this raise RuntimeError(f"Tool registration failed: {tool_name}") - # MCP SDK argument models ignore extras by default even when discovery - # advertises a closed command-specific schema. + # MCP SDK argument models ignore extras and coerce values by default even + # when discovery advertises a closed typed schema. argument_model = tool.fn_metadata.arg_model argument_model.model_config["extra"] = "forbid" + argument_model.model_config["strict"] = True argument_model.model_rebuild(force=True) tool.parameters = argument_model.model_json_schema(by_alias=True) @@ -289,13 +317,17 @@ def register( *, launch_directory: Path, tool_name: str = "specify_artifact_list", + max_wire_response_bytes: int = _MAX_WIRE_RESPONSE_BYTES, ) -> None: """Register the first-class artifact-list MCP tool exactly once.""" if server._tool_manager.get_tool(tool_name) is not None: raise ValueError(f"MCP tool name collision: {tool_name}") server.add_tool( - create_artifact_list_tool(launch_directory=launch_directory), + create_artifact_list_tool( + launch_directory=launch_directory, + max_wire_response_bytes=max_wire_response_bytes, + ), name=tool_name, description=_TOOL_DESCRIPTION, annotations=ToolAnnotations( @@ -305,4 +337,4 @@ def register( ), structured_output=True, ) - _forbid_unexpected_arguments(server, tool_name) + _configure_strict_arguments(server, tool_name) diff --git a/tests/specify_cli/artifacts/test_mcp_list.py b/tests/specify_cli/artifacts/test_mcp_list.py index c94f40b2ee..8301961574 100644 --- a/tests/specify_cli/artifacts/test_mcp_list.py +++ b/tests/specify_cli/artifacts/test_mcp_list.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import json import logging from pathlib import Path from unittest.mock import Mock, patch @@ -14,7 +15,7 @@ from mcp.server.mcpserver.exceptions import ToolError from mcp.shared.memory import create_client_server_memory_streams -from specify_cli.artifacts import _commands, _mcp, _operation_list +from specify_cli.artifacts import _commands, _mcp, _operation_list, mcp_list from specify_cli.artifacts._operation_list import ( ARTIFACT_LIST_OPERATION, ArtifactListRequest, @@ -29,7 +30,8 @@ TOOL_NAME = "specify_artifact_list" TOOL_DESCRIPTION = ( - "List every command, template, script, and hook Spec Kit exposes for a project." + "Return one paginated artifact inventory page for a project. " + "While truncated is true, pass next_cursor as cursor to retrieve the next page." ) ARTIFACT_ROWS = ( @@ -370,14 +372,19 @@ def test_artifact_list_preserves_complete_typed_rows_unicode_and_order( ): result = tool() - assert isinstance(result, ArtifactListToolResult) - assert result.model_dump() == ARTIFACT_PAYLOAD - assert [row.id for row in result.rows] == [ + assert result.is_error is False + assert result.structured_content == ARTIFACT_PAYLOAD + assert json.loads(result.content[0].text) == ARTIFACT_PAYLOAD + typed_result = ArtifactListToolResult.model_validate( + result.structured_content, + strict=True, + ) + assert [row.id for row in typed_result.rows] == [ "template:réview", "hook:before_plan:quality", ] - assert result.rows[0].stack[0].sourcePath.endswith("réview-checklist.md") - assert result.rows[1].stack[0].sourcePath is None + assert typed_result.rows[0].stack[0].sourcePath.endswith("réview-checklist.md") + assert typed_result.rows[1].stack[0].sourcePath is None def test_artifact_list_uses_server_launch_directory_by_default( @@ -459,12 +466,12 @@ def test_artifact_list_forwards_pagination_and_returns_continuation( ) -def test_artifact_list_rejects_oversized_structured_page( +def test_artifact_list_rejects_oversized_wire_response( spec_kit_project: Path, ): tool = create_artifact_list_tool( launch_directory=spec_kit_project, - max_response_bytes=64, + max_wire_response_bytes=2048, ) with patch( "specify_cli.artifacts.mcp_list.list_artifacts", @@ -476,7 +483,7 @@ def test_artifact_list_rejects_oversized_structured_page( assert result.structured_content == _expected_error( "response_too_large", "The artifact list result exceeds the MCP response-size limit.", - details={"max_bytes": 64, "requested_limit": 100}, + details={"max_bytes": 2048, "requested_limit": 100}, retryable=True, ) assert result.content[0].text == ( @@ -609,7 +616,9 @@ def test_artifact_list_rejects_invalid_project_directory_type_before_dispatch( [ {"limit": 0}, {"limit": 1001}, + {"limit": "2"}, {"cursor": "01"}, + {"cursor": 2}, ], ) def test_artifact_list_rejects_invalid_pagination_before_dispatch( @@ -866,19 +875,96 @@ def test_artifact_tool_requires_absolute_server_launch_directory(): create_artifact_list_tool(launch_directory=Path("relative")) -def test_artifact_tool_requires_positive_response_size_limit( +def test_artifact_tool_requires_room_for_json_rpc_envelope( spec_kit_project: Path, ): with pytest.raises( ValueError, - match="MCP response-size limit must be positive", + match="MCP response-size limit must exceed the JSON-RPC envelope reserve", ): create_artifact_list_tool( launch_directory=spec_kit_project, - max_response_bytes=0, + max_wire_response_bytes=1024, ) +def test_in_memory_protocol_enforces_near_boundary_wire_size( + spec_kit_project: Path, +): + max_wire_response_bytes = 4096 + within_limit: ArtifactListResult | None = None + oversized: ArtifactListResult | None = None + + for description_length in range(max_wire_response_bytes): + candidate = ArtifactListResult( + rows=( + { + **ARTIFACT_ROWS[0], + "description": "x" * description_length, + }, + ) + ) + converted = mcp_list._convert_result(candidate) + wire_size = mcp_list._wire_response_size( + mcp_list._build_success_result(converted) + ) + if wire_size <= max_wire_response_bytes: + within_limit = candidate + continue + oversized = candidate + assert ( + len(converted.model_dump_json().encode("utf-8")) < max_wire_response_bytes + ) + break + + assert within_limit is not None + assert oversized is not None + + async def exercise(): + server = MCPServer(name="test") + mcp_list.register( + server, + launch_directory=spec_kit_project, + max_wire_response_bytes=max_wire_response_bytes, + ) + operation = Mock(side_effect=[within_limit, oversized]) + with patch("specify_cli.artifacts.mcp_list.list_artifacts", operation): + 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() + success = await session.call_tool(TOOL_NAME, {}) + failure = await session.call_tool(TOOL_NAME, {}) + task_group.cancel_scope.cancel() + return success, failure + + success, failure = _run(exercise()) + + assert success.is_error is False + assert mcp_list._wire_response_size(success) <= max_wire_response_bytes + assert failure.is_error is True + assert failure.structured_content == _expected_error( + "response_too_large", + "The artifact list result exceeds the MCP response-size limit.", + details={ + "max_bytes": max_wire_response_bytes, + "requested_limit": 100, + }, + retryable=True, + ) + + def test_in_memory_client_preserves_artifact_success_and_failure_wire_shapes( spec_kit_project: Path, ): From 1fa5ce15c395d5e2a6407fc8346bcac1db0c2584 Mon Sep 17 00:00:00 2001 From: Manfred Riem <15701806+mnriem@users.noreply.github.com> Date: Wed, 7 Oct 2026 14:26:53 -0500 Subject: [PATCH 4/5] fix(mcp): harden artifact list contracts Assisted-by: GitHub Copilot (model: GPT-5.6 Sol, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/specify_cli/artifacts/_operation_list.py | 14 ++++- src/specify_cli/artifacts/mcp_list.py | 63 +++++++++++++------ tests/specify_cli/artifacts/test_mcp_list.py | 55 +++++++++++++++- .../artifacts/test_operation_list.py | 10 +-- 4 files changed, 114 insertions(+), 28 deletions(-) diff --git a/src/specify_cli/artifacts/_operation_list.py b/src/specify_cli/artifacts/_operation_list.py index 4d1a319acf..474b7642dc 100644 --- a/src/specify_cli/artifacts/_operation_list.py +++ b/src/specify_cli/artifacts/_operation_list.py @@ -117,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.""" @@ -181,6 +192,7 @@ class ArtifactListOperationDescriptor: result_type=ArtifactListResult, warning_types=(), error_types=( + ArtifactListProjectDirectoryError, ArtifactListProjectError, ArtifactListResolutionError, ArtifactListPaginationError, @@ -219,7 +231,7 @@ def list_artifacts(request: ArtifactListRequest) -> ArtifactListResult: """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: diff --git a/src/specify_cli/artifacts/mcp_list.py b/src/specify_cli/artifacts/mcp_list.py index db4da36414..8daff1ec39 100644 --- a/src/specify_cli/artifacts/mcp_list.py +++ b/src/specify_cli/artifacts/mcp_list.py @@ -5,7 +5,7 @@ import logging from collections.abc import Callable from pathlib import Path -from typing import Annotated, Literal +from typing import Annotated, Any, Literal from mcp.server import MCPServer from mcp.types import CallToolResult, TextContent, ToolAnnotations @@ -29,6 +29,9 @@ _RESPONSE_TOO_LARGE_MESSAGE = ( "The artifact list result exceeds the MCP response-size limit." ) +_SDK_LOOKUP_ERROR = ( + "MCP SDK compatibility error: registered tool lookup is unavailable." +) _TOOL_DESCRIPTION = ( "Return one paginated artifact inventory page for a project. " "While truncated is true, pass next_cursor as cursor to retrieve the next page." @@ -36,15 +39,17 @@ ArtifactListLimit = Annotated[ int, - Field(ge=1, le=ARTIFACT_LIST_MAX_LIMIT), + Field(strict=True, ge=1, le=ARTIFACT_LIST_MAX_LIMIT), ] ArtifactListCursor = Annotated[ str, Field( + strict=True, max_length=ARTIFACT_LIST_CURSOR_MAX_LENGTH, pattern=r"^(0|[1-9][0-9]*)$", ), ] +ArtifactProjectDirectory = Annotated[str, Field(strict=True)] class ArtifactListToolInput(BaseModel): @@ -52,7 +57,7 @@ class ArtifactListToolInput(BaseModel): model_config = ConfigDict(extra="forbid") - project_directory: str | None = None + project_directory: ArtifactProjectDirectory | None = None limit: ArtifactListLimit = _DEFAULT_LIMIT cursor: ArtifactListCursor | None = None @@ -153,9 +158,10 @@ class _ResponseTooLarge(Exception): """The serialized structured result exceeds the MCP response budget.""" +ArtifactListCallResult = Annotated[CallToolResult, ArtifactListToolResult] ArtifactListTool = Callable[ - [str | None, ArtifactListLimit, ArtifactListCursor | None], - ArtifactListToolResult | CallToolResult, + [ArtifactProjectDirectory | None, ArtifactListLimit, ArtifactListCursor | None], + CallToolResult, ] @@ -237,10 +243,10 @@ def create_artifact_list_tool( ) def specify_artifact_list( - project_directory: str | None = None, + project_directory: ArtifactProjectDirectory | None = None, limit: ArtifactListLimit = _DEFAULT_LIMIT, cursor: ArtifactListCursor | None = None, - ) -> ArtifactListToolResult: + ) -> ArtifactListCallResult: """Return one page; follow next_cursor while truncated is true.""" tool_input = ArtifactListToolInput.model_validate( { @@ -298,18 +304,35 @@ def specify_artifact_list( return specify_artifact_list -def _configure_strict_arguments(server: MCPServer, tool_name: str) -> None: - tool = server._tool_manager.get_tool(tool_name) - if tool is None: # pragma: no cover - registration immediately precedes this - raise RuntimeError(f"Tool registration failed: {tool_name}") +def _lookup_registered_tool(server: MCPServer, tool_name: str) -> Any | None: + try: + manager = server._tool_manager + get_tool = manager.get_tool + except AttributeError as exc: + raise RuntimeError(_SDK_LOOKUP_ERROR) from exc + try: + return get_tool(tool_name) + except Exception as exc: + raise RuntimeError(_SDK_LOOKUP_ERROR) from exc - # MCP SDK argument models ignore extras and coerce values by default even - # when discovery advertises a closed typed schema. - argument_model = tool.fn_metadata.arg_model - argument_model.model_config["extra"] = "forbid" - argument_model.model_config["strict"] = True - argument_model.model_rebuild(force=True) - tool.parameters = argument_model.model_json_schema(by_alias=True) + +def _configure_closed_arguments(server: MCPServer, tool_name: str) -> None: + tool = _lookup_registered_tool(server, tool_name) + if tool is None: + raise RuntimeError(f"MCP tool registration was not retained: {tool_name}") + + # MCP SDK argument models ignore extras by default even when discovery + # advertises a closed typed schema. Field-level strictness is declared on + # the callable; this compatibility shim only closes the generated model. + try: + 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) + except Exception as exc: + raise RuntimeError( + f"MCP SDK compatibility error while closing arguments for {tool_name}" + ) from exc def register( @@ -320,7 +343,7 @@ def register( max_wire_response_bytes: int = _MAX_WIRE_RESPONSE_BYTES, ) -> None: """Register the first-class artifact-list MCP tool exactly once.""" - if server._tool_manager.get_tool(tool_name) is not None: + if _lookup_registered_tool(server, tool_name) is not None: raise ValueError(f"MCP tool name collision: {tool_name}") server.add_tool( @@ -337,4 +360,4 @@ def register( ), structured_output=True, ) - _configure_strict_arguments(server, tool_name) + _configure_closed_arguments(server, tool_name) diff --git a/tests/specify_cli/artifacts/test_mcp_list.py b/tests/specify_cli/artifacts/test_mcp_list.py index 8301961574..89a9d66c88 100644 --- a/tests/specify_cli/artifacts/test_mcp_list.py +++ b/tests/specify_cli/artifacts/test_mcp_list.py @@ -14,6 +14,7 @@ from mcp.server import MCPServer from mcp.server.mcpserver.exceptions import ToolError from mcp.shared.memory import create_client_server_memory_streams +from mcp.types import CallToolResult from specify_cli.artifacts import _commands, _mcp, _operation_list, mcp_list from specify_cli.artifacts._operation_list import ( @@ -387,6 +388,15 @@ def test_artifact_list_preserves_complete_typed_rows_unicode_and_order( assert typed_result.rows[1].stack[0].sourcePath is None +def test_artifact_list_callable_declares_call_tool_result_contract( + spec_kit_project: Path, +): + tool = create_artifact_list_tool(launch_directory=spec_kit_project) + + assert tool.__annotations__["return"] == "ArtifactListCallResult" + assert mcp_list.ArtifactListCallResult.__origin__ is CallToolResult + + def test_artifact_list_uses_server_launch_directory_by_default( spec_kit_project: Path, non_project: Path, @@ -652,8 +662,8 @@ def test_artifact_list_rejects_relative_project_directory( assert result.is_error is True assert result.structured_content == _expected_error( - "not_a_spec_kit_project", - "not a Spec Kit project: no .specify/ directory found", + "invalid_project_directory", + "project_directory must be an absolute path", details={"project_directory": "relative-project"}, ) @@ -867,6 +877,47 @@ def test_artifact_registration_adds_available_tool_once_and_rejects_collision( assert [tool.name for tool in _run(server.list_tools())] == available_names +def test_sdk_tool_lookup_failure_is_descriptive(): + with pytest.raises( + RuntimeError, + match="MCP SDK compatibility error: registered tool lookup is unavailable", + ): + mcp_list._lookup_registered_tool(object(), TOOL_NAME) + + +def test_sdk_tool_lookup_wraps_manager_failure(): + server = Mock() + server._tool_manager.get_tool.side_effect = RuntimeError("SDK changed") + + with pytest.raises( + RuntimeError, + match="MCP SDK compatibility error: registered tool lookup is unavailable", + ): + mcp_list._lookup_registered_tool(server, TOOL_NAME) + + +def test_closed_argument_configuration_requires_retained_registration(): + server = Mock() + server._tool_manager.get_tool.return_value = None + + with pytest.raises( + RuntimeError, + match=f"MCP tool registration was not retained: {TOOL_NAME}", + ): + mcp_list._configure_closed_arguments(server, TOOL_NAME) + + +def test_closed_argument_configuration_wraps_incompatible_metadata(): + server = Mock() + server._tool_manager.get_tool.return_value = object() + + with pytest.raises( + RuntimeError, + match=f"MCP SDK compatibility error while closing arguments for {TOOL_NAME}", + ): + mcp_list._configure_closed_arguments(server, TOOL_NAME) + + def test_artifact_tool_requires_absolute_server_launch_directory(): with pytest.raises( ValueError, diff --git a/tests/specify_cli/artifacts/test_operation_list.py b/tests/specify_cli/artifacts/test_operation_list.py index 823e61de11..05c4812e4b 100644 --- a/tests/specify_cli/artifacts/test_operation_list.py +++ b/tests/specify_cli/artifacts/test_operation_list.py @@ -14,6 +14,7 @@ ARTIFACT_LIST_OPERATION, ARTIFACT_LIST_MAX_LIMIT, ArtifactListPaginationError, + ArtifactListProjectDirectoryError, ArtifactListProjectError, ArtifactListRequest, ArtifactListResolutionError, @@ -31,6 +32,7 @@ def test_artifact_list_operation_descriptor_is_stable(): assert ARTIFACT_LIST_OPERATION.result_type is ArtifactListResult assert ARTIFACT_LIST_OPERATION.warning_types == () assert ARTIFACT_LIST_OPERATION.error_types == ( + ArtifactListProjectDirectoryError, ArtifactListProjectError, ArtifactListResolutionError, ArtifactListPaginationError, @@ -235,13 +237,11 @@ def test_list_artifacts_uses_explicit_project_without_process_cwd( def test_list_artifacts_rejects_relative_project_directory(): - with pytest.raises(ArtifactListProjectError) as exc_info: + with pytest.raises(ArtifactListProjectDirectoryError) as exc_info: list_artifacts(ArtifactListRequest(Path("relative-project"))) - assert exc_info.value.code == "not_a_spec_kit_project" - assert exc_info.value.message == ( - "not a Spec Kit project: no .specify/ directory found" - ) + assert exc_info.value.code == "invalid_project_directory" + assert exc_info.value.message == "project_directory must be an absolute path" assert exc_info.value.details == {"project_directory": "relative-project"} assert exc_info.value.retryable is False From c1c5aeb2efad24157cbe7e61233f96c0c47a46a7 Mon Sep 17 00:00:00 2001 From: Manfred Riem <15701806+mnriem@users.noreply.github.com> Date: Wed, 7 Oct 2026 15:57:48 -0500 Subject: [PATCH 5/5] fix(mcp): bound artifact list error responses Bump the artifact.list contract for the revised project-directory error semantics and enforce the response budget for every adapter result. Assisted-by: GitHub Copilot (model: GPT-5.6 Sol, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/specify_cli/artifacts/_operation_list.py | 4 +- src/specify_cli/artifacts/mcp_list.py | 90 +++++++++++++------ tests/specify_cli/artifacts/test_mcp_list.py | 57 +++++++++++- .../artifacts/test_operation_list.py | 4 +- 4 files changed, 121 insertions(+), 34 deletions(-) diff --git a/src/specify_cli/artifacts/_operation_list.py b/src/specify_cli/artifacts/_operation_list.py index 474b7642dc..d699fc365f 100644 --- a/src/specify_cli/artifacts/_operation_list.py +++ b/src/specify_cli/artifacts/_operation_list.py @@ -176,7 +176,7 @@ 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], ...] @@ -187,7 +187,7 @@ class ArtifactListOperationDescriptor: ARTIFACT_LIST_OPERATION = ArtifactListOperationDescriptor( operation_id="artifact.list", - contract_version="1", + contract_version="2", request_type=ArtifactListRequest, result_type=ArtifactListResult, warning_types=(), diff --git a/src/specify_cli/artifacts/mcp_list.py b/src/specify_cli/artifacts/mcp_list.py index 8daff1ec39..bb1dd13694 100644 --- a/src/specify_cli/artifacts/mcp_list.py +++ b/src/specify_cli/artifacts/mcp_list.py @@ -154,10 +154,6 @@ class _InvalidOperationResult(Exception): """The shared operation returned an incomplete or invalid typed result.""" -class _ResponseTooLarge(Exception): - """The serialized structured result exceeds the MCP response budget.""" - - ArtifactListCallResult = Annotated[CallToolResult, ArtifactListToolResult] ArtifactListTool = Callable[ [ArtifactProjectDirectory | None, ArtifactListLimit, ArtifactListCursor | None], @@ -228,6 +224,36 @@ def _wire_response_size(result: CallToolResult) -> int: return result_bytes + _WIRE_ENVELOPE_RESERVE_BYTES +def _response_too_large_error( + *, + max_wire_response_bytes: int, + requested_limit: int, +) -> CallToolResult: + return _tool_error( + "response_too_large", + _RESPONSE_TOO_LARGE_MESSAGE, + details={ + "max_bytes": max_wire_response_bytes, + "requested_limit": requested_limit, + }, + retryable=True, + ) + + +def _bound_tool_result( + result: CallToolResult, + *, + max_wire_response_bytes: int, + requested_limit: int, +) -> CallToolResult: + if _wire_response_size(result) <= max_wire_response_bytes: + return result + return _response_too_large_error( + max_wire_response_bytes=max_wire_response_bytes, + requested_limit=requested_limit, + ) + + def create_artifact_list_tool( *, launch_directory: Path, @@ -237,9 +263,13 @@ def create_artifact_list_tool( launch_directory = Path(launch_directory) if not launch_directory.is_absolute(): raise ValueError("MCP server launch directory must be absolute") - if max_wire_response_bytes <= _WIRE_ENVELOPE_RESERVE_BYTES: + largest_fallback = _response_too_large_error( + max_wire_response_bytes=max_wire_response_bytes, + requested_limit=ARTIFACT_LIST_MAX_LIMIT, + ) + if _wire_response_size(largest_fallback) > max_wire_response_bytes: raise ValueError( - "MCP response-size limit must exceed the JSON-RPC envelope reserve" + "MCP response-size limit cannot hold the bounded error response" ) def specify_artifact_list( @@ -272,34 +302,38 @@ def specify_artifact_list( ) ) success = _build_success_result(result) - if _wire_response_size(success) > max_wire_response_bytes: - raise _ResponseTooLarge - return success + return _bound_tool_result( + success, + max_wire_response_bytes=max_wire_response_bytes, + requested_limit=tool_input.limit, + ) except ArtifactListError as exc: - return _tool_error( - exc.code, - exc.message, - details=exc.details, - retryable=exc.retryable, + return _bound_tool_result( + _tool_error( + exc.code, + exc.message, + details=exc.details, + retryable=exc.retryable, + ), + max_wire_response_bytes=max_wire_response_bytes, + requested_limit=tool_input.limit, ) except _InvalidOperationResult: - return _tool_error( - "invalid_operation_result", - _INVALID_RESULT_MESSAGE, - ) - except _ResponseTooLarge: - return _tool_error( - "response_too_large", - _RESPONSE_TOO_LARGE_MESSAGE, - details={ - "max_bytes": max_wire_response_bytes, - "requested_limit": tool_input.limit, - }, - retryable=True, + return _bound_tool_result( + _tool_error( + "invalid_operation_result", + _INVALID_RESULT_MESSAGE, + ), + max_wire_response_bytes=max_wire_response_bytes, + requested_limit=tool_input.limit, ) except Exception: logger.exception("Unexpected failure in the artifact list MCP adapter.") - return _tool_error("internal_error", _INTERNAL_ERROR_MESSAGE) + return _bound_tool_result( + _tool_error("internal_error", _INTERNAL_ERROR_MESSAGE), + max_wire_response_bytes=max_wire_response_bytes, + requested_limit=tool_input.limit, + ) return specify_artifact_list diff --git a/tests/specify_cli/artifacts/test_mcp_list.py b/tests/specify_cli/artifacts/test_mcp_list.py index 89a9d66c88..b46a41168d 100644 --- a/tests/specify_cli/artifacts/test_mcp_list.py +++ b/tests/specify_cli/artifacts/test_mcp_list.py @@ -20,6 +20,7 @@ from specify_cli.artifacts._operation_list import ( ARTIFACT_LIST_OPERATION, ArtifactListRequest, + ArtifactListResolutionError, ArtifactListResult, ) from specify_cli.artifacts.mcp_list import ( @@ -926,12 +927,12 @@ def test_artifact_tool_requires_absolute_server_launch_directory(): create_artifact_list_tool(launch_directory=Path("relative")) -def test_artifact_tool_requires_room_for_json_rpc_envelope( +def test_artifact_tool_requires_room_for_bounded_error_response( spec_kit_project: Path, ): with pytest.raises( ValueError, - match="MCP response-size limit must exceed the JSON-RPC envelope reserve", + match="MCP response-size limit cannot hold the bounded error response", ): create_artifact_list_tool( launch_directory=spec_kit_project, @@ -1016,6 +1017,58 @@ async def exercise(): ) +def test_in_memory_protocol_bounds_oversized_expected_error( + spec_kit_project: Path, +): + max_wire_response_bytes = 4096 + oversized_path = Path("/") / ("private-" * 1024) + + async def exercise(): + server = MCPServer(name="test") + mcp_list.register( + server, + launch_directory=spec_kit_project, + max_wire_response_bytes=max_wire_response_bytes, + ) + with patch( + "specify_cli.artifacts.mcp_list.list_artifacts", + side_effect=ArtifactListResolutionError(oversized_path), + ): + 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() + failure = await session.call_tool(TOOL_NAME, {}) + task_group.cancel_scope.cancel() + return failure + + failure = _run(exercise()) + + assert failure.is_error is True + assert failure.structured_content == _expected_error( + "response_too_large", + "The artifact list result exceeds the MCP response-size limit.", + details={ + "max_bytes": max_wire_response_bytes, + "requested_limit": 100, + }, + retryable=True, + ) + assert str(oversized_path) not in failure.model_dump_json(by_alias=True) + assert mcp_list._wire_response_size(failure) <= max_wire_response_bytes + + def test_in_memory_client_preserves_artifact_success_and_failure_wire_shapes( spec_kit_project: Path, ): diff --git a/tests/specify_cli/artifacts/test_operation_list.py b/tests/specify_cli/artifacts/test_operation_list.py index 05c4812e4b..52fdb9ea20 100644 --- a/tests/specify_cli/artifacts/test_operation_list.py +++ b/tests/specify_cli/artifacts/test_operation_list.py @@ -11,8 +11,8 @@ from specify_cli import artifacts from specify_cli.artifacts import ArtifactCatalog, _operation_list from specify_cli.artifacts._operation_list import ( - ARTIFACT_LIST_OPERATION, ARTIFACT_LIST_MAX_LIMIT, + ARTIFACT_LIST_OPERATION, ArtifactListPaginationError, ArtifactListProjectDirectoryError, ArtifactListProjectError, @@ -27,7 +27,7 @@ def test_artifact_list_operation_descriptor_is_stable(): assert ARTIFACT_LIST_OPERATION.operation_id == "artifact.list" - assert ARTIFACT_LIST_OPERATION.contract_version == "1" + assert ARTIFACT_LIST_OPERATION.contract_version == "2" assert ARTIFACT_LIST_OPERATION.request_type is ArtifactListRequest assert ARTIFACT_LIST_OPERATION.result_type is ArtifactListResult assert ARTIFACT_LIST_OPERATION.warning_types == ()