mcp: register release_* tools mapping 1:1 to cli subcommands

This commit is contained in:
Jan Nedbal
2026-04-14 19:33:25 +02:00
parent 860face1d1
commit 7117d25a9a

View File

@@ -0,0 +1,232 @@
"""Declarative tool specs for the Phase 3 MCP wrapper.
Each tool corresponds 1:1 to a ``metin-release release …`` subcommand. A
single :class:`ToolSpec` list drives three things:
1. The MCP ``list_tools`` response (via :func:`json_schema`)
2. Input validation for ``call_tool``
3. The CLI flag list emitted by :mod:`cli_runner`
Keeping all three in one place is what lets us guarantee the schema
matches the real argparse signature without duplicating it.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Literal
FieldType = Literal["string", "integer", "number", "boolean", "path", "array"]
@dataclass(frozen=True)
class FieldSpec:
"""One tool input field.
``name`` is the MCP-facing key (snake_case). ``flag`` is the argparse
flag name; if omitted it's derived from ``name`` by replacing ``_``
with ``-`` and prefixing ``--``. ``kind`` maps to JSON Schema types:
``path`` is a string that the CLI will read as a filesystem path,
``boolean`` is a store_true flag (passed with no value when truthy
and omitted when falsy).
"""
name: str
kind: FieldType
required: bool = False
description: str = ""
flag: str | None = None
default: Any = None
@property
def cli_flag(self) -> str:
return self.flag or f"--{self.name.replace('_', '-')}"
@property
def json_type(self) -> str:
if self.kind == "path":
return "string"
if self.kind == "array":
return "array"
return self.kind
@dataclass(frozen=True)
class ToolSpec:
name: str
subcommand: tuple[str, ...]
description: str
fields: tuple[FieldSpec, ...] = field(default_factory=tuple)
# ---------------------------------------------------------------------------
# Tool catalogue — must stay 1:1 with metin_release.commands.*.add_parser
# ---------------------------------------------------------------------------
_RELEASE = ("release",)
TOOL_SPECS: tuple[ToolSpec, ...] = (
ToolSpec(
name="release_inspect",
subcommand=_RELEASE + ("inspect",),
description="Scan a client source root and report file/byte counts plus launcher/main-exe presence.",
fields=(
FieldSpec("source", "path", required=True, description="Client root directory."),
),
),
ToolSpec(
name="release_build_manifest",
subcommand=_RELEASE + ("build-manifest",),
description="Build a signed-ready manifest.json for a source tree.",
fields=(
FieldSpec("source", "path", required=True, description="Client source root."),
FieldSpec("version", "string", required=True, description="Release version, e.g. 2026.04.14-1."),
FieldSpec("out", "path", required=True, description="Output manifest.json path."),
FieldSpec("previous", "string", description="Previous release version, if any."),
FieldSpec("notes", "path", description="Path to a release-notes file."),
FieldSpec("launcher", "string", description="Launcher filename (default Metin2Launcher.exe)."),
FieldSpec("created_at", "string", description="Override created_at for reproducible test runs."),
),
),
ToolSpec(
name="release_sign",
subcommand=_RELEASE + ("sign",),
description="Sign a manifest.json with an Ed25519 private key (mode-600 enforced).",
fields=(
FieldSpec("manifest", "path", required=True, description="Path to manifest.json."),
FieldSpec("key", "path", required=True, description="Absolute path to raw 32-byte private key."),
FieldSpec("out", "path", description="Signature output path (default: <manifest>.sig)."),
),
),
ToolSpec(
name="release_diff_remote",
subcommand=_RELEASE + ("diff-remote",),
description="HEAD every manifest blob hash against a base URL and report the missing set.",
fields=(
FieldSpec("manifest", "path", required=True, description="Path to manifest.json."),
FieldSpec("base_url", "string", required=True, description="Remote base URL."),
FieldSpec("timeout", "number", description="Per-request timeout in seconds (default 10)."),
),
),
ToolSpec(
name="release_upload_blobs",
subcommand=_RELEASE + ("upload-blobs",),
description="Rsync a release directory (excluding manifest) to the target.",
fields=(
FieldSpec("release_dir", "path", required=True, description="Local release output directory."),
FieldSpec("rsync_target", "string", required=True, description="rsync destination."),
FieldSpec("dry_run", "boolean", description="Run rsync --dry-run."),
FieldSpec("yes", "boolean", description="Skip interactive confirmation."),
),
),
ToolSpec(
name="release_promote",
subcommand=_RELEASE + ("promote",),
description="Promote a staged release by pushing manifest.json + manifest.json.sig.",
fields=(
FieldSpec("release_dir", "path", required=True, description="Local release directory."),
FieldSpec("rsync_target", "string", required=True, description="rsync destination top-level."),
FieldSpec("yes", "boolean", description="Skip interactive confirmation."),
FieldSpec("dry_run", "boolean", description="Run rsync --dry-run."),
),
),
ToolSpec(
name="release_verify_public",
subcommand=_RELEASE + ("verify-public",),
description="Fetch manifest.json + signature from a base URL and Ed25519-verify.",
fields=(
FieldSpec("base_url", "string", required=True, description="Remote base URL."),
FieldSpec("public_key", "string", required=True, description="Ed25519 public key: hex or path."),
FieldSpec("sample_blobs", "integer", description="GET and hash-check N random blobs."),
FieldSpec("timeout", "number", description="Per-request timeout in seconds (default 15)."),
),
),
ToolSpec(
name="release_publish",
subcommand=_RELEASE + ("publish",),
description="End-to-end release: build-manifest -> sign -> upload-blobs -> promote -> verify-public.",
fields=(
FieldSpec("source", "path", required=True, description="Client source root."),
FieldSpec("version", "string", required=True, description="Release version."),
FieldSpec("out", "path", required=True, description="Release output directory."),
FieldSpec("key", "path", required=True, description="Signing key path (mode 600)."),
FieldSpec("rsync_target", "string", required=True, description="rsync target for blobs + manifest."),
FieldSpec("base_url", "string", required=True, description="Public base URL for verification."),
FieldSpec("public_key", "string", required=True, description="Public key (hex or file)."),
FieldSpec("previous", "string", description="Previous release version."),
FieldSpec("notes", "path", description="Release notes file."),
FieldSpec("launcher", "string", description="Launcher filename."),
FieldSpec("created_at", "string", description="Override manifest created_at."),
FieldSpec("sample_blobs", "integer", description="verify-public sample count."),
FieldSpec("yes", "boolean", description="Skip interactive confirmation."),
FieldSpec("force", "boolean", description="Allow non-empty output directory."),
FieldSpec("dry_run_upload", "boolean", description="rsync --dry-run for upload + promote."),
),
),
)
TOOLS_BY_NAME: dict[str, ToolSpec] = {t.name: t for t in TOOL_SPECS}
def json_schema(spec: ToolSpec) -> dict[str, Any]:
"""Build a JSON Schema for one tool's input object."""
props: dict[str, Any] = {}
required: list[str] = []
for f in spec.fields:
prop: dict[str, Any] = {"type": f.json_type}
if f.description:
prop["description"] = f.description
props[f.name] = prop
if f.required:
required.append(f.name)
schema: dict[str, Any] = {
"type": "object",
"properties": props,
"additionalProperties": False,
}
if required:
schema["required"] = required
return schema
def build_cli_args(spec: ToolSpec, payload: dict[str, Any]) -> list[str]:
"""Translate a tool input dict into a ``metin-release`` argv tail.
Output is ``[*subcommand, "--json", *flags]``. Unknown keys raise
:class:`~metin_release_mcp.errors.InvalidToolInputError`. Required
fields that are missing do the same.
"""
from .errors import InvalidToolInputError
known = {f.name for f in spec.fields}
unknown = set(payload) - known
if unknown:
raise InvalidToolInputError(
f"tool {spec.name} got unknown fields: {sorted(unknown)}"
)
argv: list[str] = [*spec.subcommand, "--json"]
for f in spec.fields:
if f.name not in payload or payload[f.name] is None:
if f.required:
raise InvalidToolInputError(
f"tool {spec.name} missing required field: {f.name}"
)
continue
value = payload[f.name]
if f.kind == "boolean":
if bool(value):
argv.append(f.cli_flag)
continue
if f.kind == "array":
if not isinstance(value, list):
raise InvalidToolInputError(
f"field {f.name} must be an array"
)
for item in value:
argv.extend([f.cli_flag, str(item)])
continue
argv.extend([f.cli_flag, str(value)])
return argv