feat: add agent-guided installer
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
# Secrets / local runtime config
|
||||
.env
|
||||
config.json
|
||||
config.json*.bak
|
||||
.runtime/
|
||||
|
||||
# Python / build artifacts
|
||||
|
||||
+149
@@ -0,0 +1,149 @@
|
||||
# Agent 安装入口:NocoBase CRM MCP
|
||||
|
||||
这是一份供任意终端型 Agent 执行、也供用户审阅的安装协议。它不依赖 Codex、Claude Code、Cursor、OpenCode 或其他单一产品的插件格式。MCP 是可移植的功能层;Agent 只需把最后生成的 stdio 启动配置转换成自己客户端支持的格式。
|
||||
|
||||
如果你是收到本链接的 Agent:请完整阅读本文后再执行命令,按顺序持续工作到“完成条件”全部满足。不要把本文当成需要原样转发给用户的普通说明;你应根据操作系统、Shell、已有软件和当前 Agent 客户端自适应执行,并向用户报告关键检查点。
|
||||
|
||||
## 不可变安全规则
|
||||
|
||||
1. 只安装 [`install-manifest.json`](install-manifest.json) 指定的仓库和 `releaseRef`,不得默认安装 `main`、开发分支或其他未固定版本。
|
||||
2. 不使用 `curl | sh`、`wget | sh` 或同类远程脚本管道。缺少全局依赖时,先说明来源、版本、影响范围并取得用户同意,再采用该工具的官方安装方式。
|
||||
3. 不使用 `sudo`,不删除或覆盖未知目录,不丢弃已有 Git 改动。目标目录已存在时先识别它;不是本项目或工作区不干净时停止覆盖并向用户说明。
|
||||
4. 用户必须亲自在浏览器完成 NocoBase OAuth 登录。不得索取、复制、打印或保存用户的 Access Token、Refresh Token、Cookie、密码或 `~/.nocobase` 目录;不得执行带 `--show-secrets` 的命令。
|
||||
5. 默认只读。除非用户明确要求且管理员已确认 NocoBase ACL,否则不得启用 `--enable-writes`、附件上传或原始数据输出。
|
||||
6. 写入 Agent/MCP 客户端配置前,显示目标文件和将新增的 server 名称,创建可恢复备份,并只合并 `nocobase-crm` 条目;不得覆盖其他 MCP 配置。
|
||||
7. 不把本机 MCP 端口暴露到局域网或公网。通用安装默认使用 stdio,不创建后台服务,也不生成共享业务凭证。
|
||||
|
||||
## 安装输入
|
||||
|
||||
从用户或当前上下文确定以下内容;已有明确答案时不要重复提问:
|
||||
|
||||
- NocoBase 应用 URL,例如 `https://nocobase.example`。执行 `nb env add` 前规范化为以 `/api` 结尾。
|
||||
- 安装目录。使用该操作系统的普通用户应用/数据目录作为默认值,并在写入前告知用户;不要默认放在系统目录。
|
||||
- 当前 Agent/MCP 客户端的名称及其本机 MCP 配置位置。
|
||||
- 功能模式。未明确选择时必须使用“只读”。
|
||||
|
||||
不要要求用户提供 NocoBase Token。OAuth 登录之外没有必填秘密输入。
|
||||
|
||||
## 执行协议
|
||||
|
||||
### 1. 读取并锁定发布版本
|
||||
|
||||
读取同版本的 `install-manifest.json`。本版本的预期来源是:
|
||||
|
||||
```text
|
||||
repository: https://github.com/Donaldttt/nocobase-mcp.git
|
||||
releaseRef: v0.2.0
|
||||
```
|
||||
|
||||
若 `v0.2.0` 尚不存在,或 Agent 无权读取私有仓库,停止安装并清楚报告“发布物/访问权限尚未就绪”,不得退回开发分支。面向普通外部用户发布时,安装方必须能匿名读取本页和发布物;私有分发则需要用户已连接有权访问该仓库的 GitHub 账号。
|
||||
|
||||
首次安装使用等效于以下语义的 Git 命令,并根据当前 Shell 正确转义路径:
|
||||
|
||||
```bash
|
||||
git clone --branch v0.2.0 --depth 1 https://github.com/Donaldttt/nocobase-mcp.git <install-dir>
|
||||
git -C <install-dir> describe --tags --exact-match HEAD
|
||||
```
|
||||
|
||||
第二条必须精确输出 `v0.2.0`。已有安装的更新不得直接 `git pull` 开发分支;先保存当前版本和 `config.json`,确认工作区干净,再获取并切换到用户指定的新 Release Tag。
|
||||
|
||||
### 2. 环境检查与确定性准备
|
||||
|
||||
需要:Git、`uv`、Python 3.12+、NocoBase `nb` CLI 2.1+。先运行只读检查:
|
||||
|
||||
```bash
|
||||
uv run --no-project --python 3.12 python <install-dir>/scripts/agent_install.py --project-root <install-dir> inspect --json
|
||||
```
|
||||
|
||||
若 `uv` 或 `nb` 不存在,先取得用户同意再安装。NocoBase CLI 当前官方包入口是 `npm install -g @nocobase/cli`;安装后必须重新运行检查,不要通过修改 PATH 输出伪造成功。
|
||||
|
||||
然后运行幂等准备步骤:
|
||||
|
||||
```bash
|
||||
uv run --no-project --python 3.12 python <install-dir>/scripts/agent_install.py --project-root <install-dir> prepare --json
|
||||
```
|
||||
|
||||
该脚本只会校验项目文件、在缺失时从示例创建 `config.json`,并执行锁定依赖同步。已有 `config.json` 会被保留,依赖必须来自 `uv.lock`。
|
||||
|
||||
### 3. 连接用户自己的 NocoBase 账号
|
||||
|
||||
先直接执行:
|
||||
|
||||
```bash
|
||||
nb env list
|
||||
```
|
||||
|
||||
如果 `nb-mcp` 不存在,为规范化后的目标 URL 创建 OAuth 环境,并在写操作后读回列表:
|
||||
|
||||
```bash
|
||||
nb env add nb-mcp --api-base-url <normalized-url-ending-in-/api> --auth-type oauth
|
||||
nb env list
|
||||
```
|
||||
|
||||
如果 `nb-mcp` 已存在,先用不显示秘密的 `nb env info nb-mcp` 核对 URL。URL 相同则复用;URL 不同则不要覆盖,选择清楚且未占用的新名称(例如 `nb-mcp-example`),创建后同步更新项目配置:
|
||||
|
||||
```bash
|
||||
uv run --no-project --python 3.12 python <install-dir>/scripts/agent_install.py --project-root <install-dir> set-env <new-env-name> --json
|
||||
```
|
||||
|
||||
此命令修改前会保留 `config.json.pre-agent-install.bak`。然后启动登录:
|
||||
|
||||
```bash
|
||||
nb env auth <configured-env>
|
||||
```
|
||||
|
||||
让用户亲自在浏览器完成登录;Agent 只负责保持命令运行、提示用户返回以及继续读取 CLI 的成功或错误信息。最后执行 `nb env info <configured-env>` 验证,不得附加 `--show-secrets`。如果登录失败或过期,重新运行同一个 `nb env auth`,不得换用共享 Token。
|
||||
|
||||
### 4. 注册到当前 Agent/MCP 客户端
|
||||
|
||||
生成不含 NocoBase Token 的通用配置:
|
||||
|
||||
```bash
|
||||
uv run --no-project --python 3.12 python <install-dir>/scripts/agent_install.py --project-root <install-dir> client-config
|
||||
```
|
||||
|
||||
输出采用常见的 `mcpServers` JSON 形状,包含绝对的项目、`uv` 和 `nb` 路径。Agent 应查明自己客户端当前版本支持的 MCP 配置格式,将其中的 `command`、`args`、`env` 等价映射过去。
|
||||
|
||||
写入前:
|
||||
|
||||
1. 向用户显示目标客户端配置文件路径和新增的 `nocobase-crm` 条目。
|
||||
2. 备份原文件;解析后做结构化合并,不能用字符串替换或整文件覆盖。
|
||||
3. 保留所有其他 server、设置和注释(客户端格式允许时)。
|
||||
4. 按客户端要求重载 MCP 配置或提醒用户重启客户端。
|
||||
|
||||
只有用户明确要求项目创建/更新,且其管理员确认对应角色 ACL 后,才可重新生成含写工具的配置:
|
||||
|
||||
```bash
|
||||
uv run --no-project --python 3.12 python <install-dir>/scripts/agent_install.py --project-root <install-dir> client-config --enable-writes
|
||||
```
|
||||
|
||||
这仍不启用上传、删除或原始数据。附件上传属于单独的 Linux 本机 HTTP 高级部署,不是通用安装流程的一部分。
|
||||
|
||||
### 5. 验证
|
||||
|
||||
先检查本机安装,不读取 NocoBase 秘密:
|
||||
|
||||
```bash
|
||||
uv run --no-project --python 3.12 python <install-dir>/scripts/agent_install.py --project-root <install-dir> doctor --json
|
||||
```
|
||||
|
||||
然后通过已注册的 MCP 调用 `crm_health`。只有同时满足以下条件才可向用户报告安装完成:
|
||||
|
||||
- `doctor` 返回 `ok: true`;
|
||||
- `nb env info <configured-env>` 成功且目标 URL 正确;
|
||||
- 客户端已发现 `nocobase-crm`;
|
||||
- `crm_health` 返回 `ok: true`,不是 `nb_auth_failed`、`NO_PERMISSIONS` 或其他结构化错误;
|
||||
- 默认状态仍是只读、uploads disabled、raw data disabled。
|
||||
|
||||
最后报告安装目录、Release Tag、NocoBase 环境名、客户端配置备份位置、启用的能力以及一条最小使用示例。不要报告或回显任何 Token。
|
||||
|
||||
## 给用户的最短用法
|
||||
|
||||
普通情况下,用户只需把当前**不可变 Release 版本**的 `INSTALL.md` 链接发给具备本机终端和文件权限的 Agent,并说:
|
||||
|
||||
```text
|
||||
请完整阅读这个安装协议,根据我的系统和当前 Agent 客户端自适应安装。默认只读;不要索取或显示任何 Token;需要全局安装依赖、写客户端配置或启用写功能前先告诉我。安装后必须通过 doctor 和 crm_health:
|
||||
<version-pinned INSTALL.md URL>
|
||||
```
|
||||
|
||||
如果 Agent 只能聊天、不能执行本机命令或写 MCP 配置,它应说明能力限制并按本文生成逐步人工安装指令,而不能声称已经安装。
|
||||
@@ -4,6 +4,8 @@ Local MCP server for end users who need customer, contact, follow-up, CRM overvi
|
||||
|
||||
Each user runs this MCP on their own computer and signs in to the remote NocoBase environment with `nb` CLI. Every query and write therefore uses that user's NocoBase identity and ACL. The project does not require a shared remote MCP service or a shared NocoBase service account.
|
||||
|
||||
For an agent-guided installation that works across MCP clients, give the installing Agent the immutable Release version of [`INSTALL.md`](INSTALL.md). The guide combines adaptive prompts with an idempotent helper script; it defaults to local stdio and read-only access. Do not distribute a `main`-branch installation link as a release.
|
||||
|
||||
The default stdio server is read-only. Add `--enable-writes` for fixed-schema project create/update tools. Optional local HTTP mode also provides authenticated attachment uploads. Neither mode exposes arbitrary collections, fields, commands, or delete operations.
|
||||
|
||||
## Configure
|
||||
|
||||
@@ -18,6 +18,14 @@ MCP 不保存公共业务账号,也不需要部署一个多人共用的远程
|
||||
- 可通过浏览器访问目标 NocoBase
|
||||
- 管理员已为用户配置所需的 NocoBase 角色与数据权限
|
||||
|
||||
## 推荐:链接驱动的 Agent 安装
|
||||
|
||||
正式发布后,优先把对应 Release Tag 下的 [`INSTALL.md`](../INSTALL.md) 不可变链接交给用户。该入口不是 Codex 专用插件:任何能够读取链接、执行本机终端并结构化修改 MCP 客户端配置的 Agent,都可以按协议完成安装。确定性工作由 `scripts/agent_install.py` 处理;Agent 负责识别操作系统与客户端、解释变更、备份并合并客户端配置;用户本人完成 OAuth 登录。
|
||||
|
||||
发布方必须保证安装 Agent 能读取链接和相同版本的代码。当前仓库为 private 时,用户需要已经连接有仓库权限的 GitHub 账号;面向普通外部用户的一条链接安装,则必须公开该版本的 `INSTALL.md` 与发布物,或放到用户 Agent 可认证访问的分发渠道。不要把开发分支链接当作交付链接。
|
||||
|
||||
这个通用流程默认使用 stdio,因此不需要后台服务或本机 MCP Bearer Token。Linux systemd/HTTP 方式只用于确有附件上传需求的高级场景。
|
||||
|
||||
## 首次安装
|
||||
|
||||
```bash
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
{
|
||||
"schemaVersion": 1,
|
||||
"id": "com.nocobase.crm-mcp",
|
||||
"name": "NocoBase CRM MCP",
|
||||
"version": "0.2.0",
|
||||
"distribution": {
|
||||
"repository": "https://github.com/Donaldttt/nocobase-mcp.git",
|
||||
"releaseRef": "v0.2.0",
|
||||
"guide": "https://raw.githubusercontent.com/Donaldttt/nocobase-mcp/v0.2.0/INSTALL.md",
|
||||
"accessRequirement": "The installing agent must be able to read the guide and clone the exact release ref. A private repository requires an authenticated GitHub connection."
|
||||
},
|
||||
"runtime": {
|
||||
"location": "user-device",
|
||||
"defaultTransport": "stdio",
|
||||
"python": ">=3.12",
|
||||
"packageManager": "uv",
|
||||
"nocobaseCli": ">=2.1"
|
||||
},
|
||||
"entrypoints": {
|
||||
"guide": "INSTALL.md",
|
||||
"helper": "scripts/agent_install.py",
|
||||
"configTemplate": "config.example.json"
|
||||
},
|
||||
"securityDefaults": {
|
||||
"readOnly": true,
|
||||
"writes": false,
|
||||
"uploads": false,
|
||||
"rawData": false,
|
||||
"sharedNocobaseCredential": false,
|
||||
"networkBind": "none-stdio"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,368 @@
|
||||
#!/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())
|
||||
@@ -0,0 +1,134 @@
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import tomllib
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts"
|
||||
if str(SCRIPTS_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(SCRIPTS_DIR))
|
||||
|
||||
import agent_install # noqa: E402
|
||||
|
||||
|
||||
def make_project(root: Path) -> None:
|
||||
(root / "pyproject.toml").write_text('[project]\nname = "test"\n', encoding="utf-8")
|
||||
(root / "uv.lock").write_text("version = 1\n", encoding="utf-8")
|
||||
(root / "config.example.json").write_text(json.dumps({"env": "nb-mcp", "collections": {}}), encoding="utf-8")
|
||||
|
||||
|
||||
def fake_which(name: str) -> str | None:
|
||||
return {
|
||||
"git": "/usr/bin/git",
|
||||
"uv": "/opt/bin/uv",
|
||||
"nb": "/opt/bin/nb",
|
||||
}.get(name)
|
||||
|
||||
|
||||
class AgentInstallTests(unittest.TestCase):
|
||||
def test_inspect_reports_required_tools_and_config_as_optional_before_prepare(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
make_project(root)
|
||||
|
||||
result = agent_install.inspect_project(root, which=fake_which, python_version=(3, 12, 4))
|
||||
|
||||
self.assertTrue(result["ok"])
|
||||
checks = {check["name"]: check for check in result["checks"]}
|
||||
self.assertTrue(checks["uv"]["ok"])
|
||||
self.assertFalse(checks["config"]["ok"])
|
||||
self.assertFalse(checks["config"]["required"])
|
||||
|
||||
def test_prepare_is_idempotent_and_never_overwrites_existing_config(self) -> None:
|
||||
calls = []
|
||||
|
||||
def runner(command, **kwargs):
|
||||
calls.append((command, kwargs))
|
||||
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
make_project(root)
|
||||
|
||||
first = agent_install.prepare_project(root, runner=runner, which=fake_which, python_version=(3, 12, 4))
|
||||
config_path = root / "config.json"
|
||||
config_path.write_text(json.dumps({"env": "custom-env"}), encoding="utf-8")
|
||||
second = agent_install.prepare_project(root, runner=runner, which=fake_which, python_version=(3, 12, 4))
|
||||
|
||||
preserved = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
|
||||
self.assertTrue(first["configCreated"])
|
||||
self.assertFalse(second["configCreated"])
|
||||
self.assertEqual(preserved["env"], "custom-env")
|
||||
self.assertEqual(calls[0][0], ["/opt/bin/uv", "sync", "--locked", "--python", "3.12"])
|
||||
self.assertEqual(len(calls), 2)
|
||||
|
||||
def test_set_env_backs_up_config_before_changing_it(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
make_project(root)
|
||||
original = {"env": "nb-mcp", "collections": {"customers": {"resource": "customers"}}}
|
||||
(root / "config.json").write_text(json.dumps(original), encoding="utf-8")
|
||||
|
||||
result = agent_install.set_config_env(root, "nb-mcp-acme")
|
||||
|
||||
changed = json.loads((root / "config.json").read_text(encoding="utf-8"))
|
||||
backup = json.loads((root / "config.json.pre-agent-install.bak").read_text(encoding="utf-8"))
|
||||
|
||||
self.assertTrue(result["changed"])
|
||||
self.assertEqual(changed["env"], "nb-mcp-acme")
|
||||
self.assertEqual(backup, original)
|
||||
|
||||
def test_client_config_uses_absolute_paths_and_is_read_only_by_default(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
config = agent_install.render_client_config(root, which=fake_which)
|
||||
writable = agent_install.render_client_config(root, enable_writes=True, which=fake_which)
|
||||
|
||||
server = config["mcpServers"]["nocobase-crm"]
|
||||
self.assertEqual(server["command"], "/opt/bin/uv")
|
||||
self.assertEqual(server["env"]["NOCOBASE_MCP_NB_BIN"], "/opt/bin/nb")
|
||||
self.assertIn(str(root.resolve()), server["args"])
|
||||
self.assertNotIn("--enable-writes", server["args"])
|
||||
self.assertIn("--enable-writes", writable["mcpServers"]["nocobase-crm"]["args"])
|
||||
|
||||
def test_doctor_checks_local_install_without_reading_nb_secrets(self) -> None:
|
||||
commands = []
|
||||
|
||||
def runner(command, **kwargs):
|
||||
commands.append(command)
|
||||
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
make_project(root)
|
||||
(root / "config.json").write_text(json.dumps({"env": "nb-mcp"}), encoding="utf-8")
|
||||
|
||||
result = agent_install.doctor_project(
|
||||
root,
|
||||
runner=runner,
|
||||
which=fake_which,
|
||||
python_version=(3, 12, 4),
|
||||
)
|
||||
|
||||
self.assertTrue(result["ok"])
|
||||
self.assertEqual(commands[0], ["/opt/bin/uv", "lock", "--check"])
|
||||
self.assertNotIn("--show-secrets", json.dumps(result))
|
||||
self.assertTrue(any("crm_health" in step for step in result["remainingVerification"]))
|
||||
|
||||
def test_release_manifest_version_matches_package_version(self) -> None:
|
||||
root = Path(__file__).resolve().parent.parent
|
||||
manifest = json.loads((root / "install-manifest.json").read_text(encoding="utf-8"))
|
||||
package = tomllib.loads((root / "pyproject.toml").read_text(encoding="utf-8"))
|
||||
|
||||
version = package["project"]["version"]
|
||||
self.assertEqual(manifest["version"], version)
|
||||
self.assertEqual(manifest["distribution"]["releaseRef"], f"v{version}")
|
||||
self.assertTrue(manifest["securityDefaults"]["readOnly"])
|
||||
self.assertFalse(manifest["securityDefaults"]["uploads"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user