nocobase-mcp/scripts/agent_install.py

#!/usr/bin/env python3
"""Deterministic helpers for the agent-guided local installation flow.

This script deliberately does not clone the repository, install global tools,
authenticate to NocoBase, or edit an AI client's configuration. Those actions
need environment-specific judgment or direct user participation and are
governed by INSTALL.md.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from collections.abc import Callable, Sequence
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any

MIN_PYTHON = (3, 12)
DEFAULT_ENV_NAME = "nb-mcp"
SERVER_NAME = "nocobase-crm"
REQUIRED_PROJECT_FILES = ("pyproject.toml", "uv.lock", "config.example.json")
ENV_NAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")

Which = Callable[[str], str | None]
Runner = Callable[..., subprocess.CompletedProcess[str]]


class InstallerError(RuntimeError):
    """Raised when a deterministic installation step cannot finish safely."""


@dataclass(frozen=True)
class Check:
    name: str
    ok: bool
    required: bool
    detail: str


def _project_root(path: str | Path | None) -> Path:
    candidate = Path(path) if path is not None else Path(__file__).resolve().parent.parent
    return candidate.expanduser().resolve()


def inspect_project(
    project_root: str | Path | None = None,
    *,
    which: Which = shutil.which,
    python_version: tuple[int, int, int] | None = None,
) -> dict[str, Any]:
    root = _project_root(project_root)
    version = python_version or (sys.version_info.major, sys.version_info.minor, sys.version_info.micro)
    commands = {name: which(name) for name in ("git", "uv", "nb")}
    missing_files = [name for name in REQUIRED_PROJECT_FILES if not (root / name).is_file()]
    checks = [
        Check(
            "python",
            version[:2] >= MIN_PYTHON,
            True,
            f"Python {version[0]}.{version[1]}.{version[2]} (requires >= {MIN_PYTHON[0]}.{MIN_PYTHON[1]})",
        ),
        Check("git", bool(commands["git"]), True, commands["git"] or "git was not found on PATH"),
        Check("uv", bool(commands["uv"]), True, commands["uv"] or "uv was not found on PATH"),
        Check("nb", bool(commands["nb"]), True, commands["nb"] or "nb was not found on PATH"),
        Check(
            "project_files",
            not missing_files,
            True,
            "required project files are present" if not missing_files else f"missing: {', '.join(missing_files)}",
        ),
        Check(
            "config",
            (root / "config.json").is_file(),
            False,
            "config.json is present" if (root / "config.json").is_file() else "config.json will be created by prepare",
        ),
    ]
    return {
        "ok": all(check.ok for check in checks if check.required),
        "projectRoot": str(root),
        "checks": [asdict(check) for check in checks],
    }


def _require_prepare_prerequisites(
    root: Path,
    *,
    which: Which,
    python_version: tuple[int, int, int],
) -> str:
    inspection = inspect_project(root, which=which, python_version=python_version)
    required = {check["name"]: check for check in inspection["checks"] if check["required"]}
    blocking = [required[name]["detail"] for name in ("python", "uv", "project_files") if not required[name]["ok"]]
    if blocking:
        raise InstallerError("; ".join(blocking))
    uv_bin = which("uv")
    if not uv_bin:
        raise InstallerError("uv was not found on PATH")
    return os.path.abspath(uv_bin)


def _load_json_object(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise InstallerError(f"Could not read valid JSON from {path}: {error}") from error
    if not isinstance(value, dict):
        raise InstallerError(f"Expected a JSON object in {path}")
    return value


def prepare_project(
    project_root: str | Path | None = None,
    *,
    runner: Runner = subprocess.run,
    which: Which = shutil.which,
    python_version: tuple[int, int, int] | None = None,
) -> dict[str, Any]:
    root = _project_root(project_root)
    version = python_version or (sys.version_info.major, sys.version_info.minor, sys.version_info.micro)
    uv_bin = _require_prepare_prerequisites(root, which=which, python_version=version)
    example_path = root / "config.example.json"
    config_path = root / "config.json"
    _load_json_object(example_path)

    created_config = False
    if config_path.exists():
        _load_json_object(config_path)
    else:
        shutil.copy2(example_path, config_path)
        created_config = True

    command = [uv_bin, "sync", "--locked", "--python", f"{MIN_PYTHON[0]}.{MIN_PYTHON[1]}"]
    try:
        result = runner(command, cwd=str(root), capture_output=True, text=True, check=False)
    except OSError as error:
        raise InstallerError(f"Could not execute uv sync: {error}") from error
    if result.returncode != 0:
        raise InstallerError(
            f"uv sync --locked failed with exit code {result.returncode}; rerun it directly to inspect package-manager output"
        )

    return {
        "ok": True,
        "projectRoot": str(root),
        "configPath": str(config_path),
        "configCreated": created_config,
        "dependencies": "locked",
    }


def set_config_env(project_root: str | Path | None, env_name: str) -> dict[str, Any]:
    if not ENV_NAME_PATTERN.fullmatch(env_name):
        raise InstallerError(
            "Environment name must start with a letter or digit and contain only letters, digits, dot, underscore, or dash"
        )
    root = _project_root(project_root)
    config_path = root / "config.json"
    config = _load_json_object(config_path)
    previous = config.get("env")
    if previous == env_name:
        return {
            "ok": True,
            "changed": False,
            "configPath": str(config_path),
            "env": env_name,
            "backupPath": None,
        }

    backup_path = root / "config.json.pre-agent-install.bak"
    if not backup_path.exists():
        shutil.copy2(config_path, backup_path)
    config["env"] = env_name

    fd, temporary_name = tempfile.mkstemp(prefix=".config.json.", dir=root)
    temporary_path = Path(temporary_name)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(config, handle, ensure_ascii=False, indent=2)
            handle.write("\n")
        os.replace(temporary_path, config_path)
    finally:
        if temporary_path.exists():
            temporary_path.unlink()

    return {
        "ok": True,
        "changed": True,
        "configPath": str(config_path),
        "env": env_name,
        "previousEnv": previous,
        "backupPath": str(backup_path),
    }


def render_client_config(
    project_root: str | Path | None = None,
    *,
    enable_writes: bool = False,
    which: Which = shutil.which,
) -> dict[str, Any]:
    root = _project_root(project_root)
    uv_bin = which("uv") or "uv"
    nb_bin = which("nb") or "nb"
    if os.path.isabs(uv_bin):
        uv_bin = os.path.abspath(uv_bin)
    if os.path.isabs(nb_bin):
        nb_bin = os.path.abspath(nb_bin)
    args = [
        "--directory",
        str(root),
        "run",
        "--no-sync",
        "nocobase-crm-mcp",
        "--config",
        str(root / "config.json"),
    ]
    if enable_writes:
        args.append("--enable-writes")
    return {
        "mcpServers": {
            SERVER_NAME: {
                "command": uv_bin,
                "args": args,
                "env": {"NOCOBASE_MCP_NB_BIN": nb_bin},
            }
        }
    }


def doctor_project(
    project_root: str | Path | None = None,
    *,
    runner: Runner = subprocess.run,
    which: Which = shutil.which,
    python_version: tuple[int, int, int] | None = None,
) -> dict[str, Any]:
    root = _project_root(project_root)
    inspection = inspect_project(root, which=which, python_version=python_version)
    checks = [Check(**item) for item in inspection["checks"]]

    config_path = root / "config.json"
    if config_path.is_file():
        try:
            config = _load_json_object(config_path)
            env_name = config.get("env")
            config_ok = isinstance(env_name, str) and bool(env_name.strip())
            detail = f"config env is {env_name!r}" if config_ok else "config.json has no non-empty env"
        except InstallerError as error:
            config_ok = False
            detail = str(error)
    else:
        config_ok = False
        detail = "config.json is missing; run prepare"
    checks.append(Check("configured_env", config_ok, True, detail))

    uv_bin = which("uv")
    if uv_bin:
        commands = (
            [uv_bin, "lock", "--check"],
            [
                uv_bin,
                "run",
                "--no-sync",
                "python",
                "-c",
                "import mcp; import nocobase_crm_mcp",
            ],
        )
        labels = ("lockfile", "python_environment")
        for label, command in zip(labels, commands, strict=True):
            try:
                result = runner(command, cwd=str(root), capture_output=True, text=True, check=False)
                ok = result.returncode == 0
                detail = "passed" if ok else f"failed with exit code {result.returncode}"
            except OSError as error:
                ok = False
                detail = str(error)
            checks.append(Check(label, ok, True, detail))
    else:
        checks.extend(
            [
                Check("lockfile", False, True, "uv was not found on PATH"),
                Check("python_environment", False, True, "uv was not found on PATH"),
            ]
        )

    return {
        "ok": all(check.ok for check in checks if check.required),
        "scope": "local-installation",
        "projectRoot": str(root),
        "checks": [asdict(check) for check in checks],
        "remainingVerification": [
            "Run nb env info <configured-env> in its default redacted mode.",
            "Register the generated stdio MCP configuration in the user's client.",
            "Call the MCP tool crm_health and require ok=true.",
        ],
    }


def _print_result(payload: dict[str, Any], *, as_json: bool) -> None:
    if as_json:
        print(json.dumps(payload, ensure_ascii=False, indent=2))
        return
    print(f"Overall: {'ready' if payload.get('ok') else 'action required'}")
    if "projectRoot" in payload:
        print(f"Project root: {payload['projectRoot']}")
    for check in payload.get("checks", []):
        marker = "PASS" if check["ok"] else ("FAIL" if check["required"] else "INFO")
        print(f"[{marker}] {check['name']}: {check['detail']}")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Prepare and inspect an agent-guided NocoBase CRM MCP installation.")
    parser.add_argument("--project-root", type=Path, default=None)
    subparsers = parser.add_subparsers(dest="command", required=True)

    for name in ("inspect", "prepare", "doctor"):
        command = subparsers.add_parser(name)
        command.add_argument("--json", action="store_true", dest="as_json")

    set_env = subparsers.add_parser("set-env")
    set_env.add_argument("env_name")
    set_env.add_argument("--json", action="store_true", dest="as_json")

    client_config = subparsers.add_parser("client-config")
    client_config.add_argument("--enable-writes", action="store_true")
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        if args.command == "inspect":
            payload = inspect_project(args.project_root)
            _print_result(payload, as_json=args.as_json)
            return 0 if payload["ok"] else 1
        if args.command == "prepare":
            payload = prepare_project(args.project_root)
            _print_result(payload, as_json=args.as_json)
            return 0
        if args.command == "set-env":
            payload = set_config_env(args.project_root, args.env_name)
            _print_result(payload, as_json=args.as_json)
            return 0
        if args.command == "client-config":
            print(json.dumps(render_client_config(args.project_root, enable_writes=args.enable_writes), indent=2))
            return 0
        if args.command == "doctor":
            payload = doctor_project(args.project_root)
            _print_result(payload, as_json=args.as_json)
            return 0 if payload["ok"] else 1
    except InstallerError as error:
        print(json.dumps({"ok": False, "error": {"code": "install_step_failed", "message": str(error)}}))
        return 1
    raise AssertionError(f"Unhandled command: {args.command}")


if __name__ == "__main__":
    raise SystemExit(main())