gitea-admin/nocobase-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: migrate MCP runtime to TypeScript

Browse Source
This commit is contained in:
nac
2026-08-15 05:02:28 +00:00
parent 7eb416ded1
commit 27672a0940
83 changed files with 6559 additions and 12057 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/agent_install.py
-52
View File
@@ -1,52 +0,0 @@
from __future__ import annotations
from agent_installer.auth import (
connect_with_personal_key,
normalize_api_base_url,
)
from agent_installer.cli import build_parser, main
from agent_installer.common import (
DEFAULT_ENV_NAME,
ENV_NAME_PATTERN,
MIN_PYTHON,
REQUIRED_PROJECT_FILES,
SERVER_NAME,
Check,
InstallerError,
Runner,
Which,
)
from agent_installer.diagnostics import doctor_project
from agent_installer.flow import install_project
from agent_installer.project import (
inspect_project,
prepare_project,
render_client_config,
set_config_env,
)
__all__ = [
"Check",
"DEFAULT_ENV_NAME",
"ENV_NAME_PATTERN",
"InstallerError",
"MIN_PYTHON",
"REQUIRED_PROJECT_FILES",
"Runner",
"SERVER_NAME",
"Which",
"build_parser",
"connect_with_personal_key",
"doctor_project",
"inspect_project",
"install_project",
"main",
"normalize_api_base_url",
"prepare_project",
"render_client_config",
"set_config_env",
]
if __name__ == "__main__":
raise SystemExit(main())
scripts/agent_installer/__init__.py
-17
View File
@@ -1,17 +0,0 @@
from .auth import connect_with_personal_key, normalize_api_base_url
from .cli import main
from .diagnostics import doctor_project
from .flow import install_project
from .project import inspect_project, prepare_project, render_client_config, set_config_env
__all__ = [
"connect_with_personal_key",
"doctor_project",
"inspect_project",
"install_project",
"main",
"normalize_api_base_url",
"prepare_project",
"render_client_config",
"set_config_env",
]
scripts/agent_installer/auth.py
-230
View File
@@ -1,230 +0,0 @@
from __future__ import annotations
from .common import (
ANSI_ESCAPE_PATTERN,
DEFAULT_ENV_NAME,
ENV_NAME_PATTERN,
Any,
Callable,
InstallerError,
Path,
Runner,
Which,
_project_root,
json,
os,
re,
shutil,
subprocess,
sys,
urlsplit,
urlunsplit,
)
from .project import set_config_env
def normalize_api_base_url(value: str) -> str:
raw = value.strip()
if not raw:
raise InstallerError("NocoBase API base URL is required")
parsed = urlsplit(raw)
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
raise InstallerError("NocoBase URL must be an absolute http:// or https:// URL")
if parsed.username or parsed.password or parsed.query or parsed.fragment:
raise InstallerError("NocoBase URL must not contain credentials, a query string, or a fragment")
hostname = parsed.hostname.lower()
is_local = hostname == "localhost" or hostname == "::1" or hostname.startswith("127.")
if parsed.scheme != "https" and not is_local:
raise InstallerError("A remote NocoBase API Key may only be sent to an https:// endpoint")
path = parsed.path.rstrip("/")
if not path.endswith("/api"):
path = f"{path}/api" if path else "/api"
return urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))
def _parse_cli_json(output: str) -> dict[str, Any]:
decoder = json.JSONDecoder()
candidates: list[tuple[int, int, dict[str, Any]]] = []
for start, character in enumerate(output):
if character != "{":
continue
try:
value, end = decoder.raw_decode(output, start)
except json.JSONDecodeError:
continue
if isinstance(value, dict):
candidates.append((start, end, value))
outer_candidates = [
candidate
for candidate in candidates
if not any(
other_start <= candidate[0]
and other_end >= candidate[1]
and (other_start, other_end) != (candidate[0], candidate[1])
for other_start, other_end, _value in candidates
)
]
if len(outer_candidates) != 1:
raise InstallerError("nb env info did not return one JSON object")
return outer_candidates[0][2]
def _nb_command_environment(nb_bin: str) -> dict[str, str]:
environment = os.environ.copy()
environment["NB_CLI_LOG_DISABLED"] = "1"
environment["NB_SKIP_STARTUP_UPDATE"] = "1"
nb_path = Path(nb_bin)
if nb_path.is_absolute():
nb_dir = str(nb_path.parent)
existing_path = environment.get("PATH", "")
parts = [part for part in existing_path.split(os.pathsep) if part]
if nb_dir not in parts:
environment["PATH"] = nb_dir + (os.pathsep + existing_path if existing_path else "")
return environment
def _read_nb_environment(
root: Path,
env_name: str,
*,
nb_bin: str,
runner: Runner,
) -> dict[str, Any] | None:
command = [nb_bin, "env", "info", env_name, "--json"]
try:
result = runner(
command,
cwd=str(root),
env=_nb_command_environment(nb_bin),
capture_output=True,
text=True,
check=False,
)
except OSError as error:
raise InstallerError(f"Could not execute nb env info: {error}") from error
if result.returncode == 0:
return _parse_cli_json(result.stdout)
diagnostic = ANSI_ESCAPE_PATTERN.sub("", f"{result.stdout}\n{result.stderr}")
if "is not configured" in diagnostic:
return None
raise InstallerError(
f"nb env info failed with exit code {result.returncode}; inspect the nb CLI diagnostic locally"
)
def _api_details(payload: dict[str, Any]) -> tuple[str | None, str | None]:
api = payload.get("api")
auth = api.get("auth") if isinstance(api, dict) else None
api_base_url = api.get("apiBaseUrl") if isinstance(api, dict) else None
auth_type = auth.get("type") if isinstance(auth, dict) else None
normalized_url = api_base_url.strip().rstrip("/") if isinstance(api_base_url, str) else None
normalized_type = auth_type.strip() if isinstance(auth_type, str) else None
return normalized_url, normalized_type
def _suggest_env_name(api_base_url: str) -> str:
hostname = urlsplit(api_base_url).hostname or "server"
suffix = re.sub(r"[^a-z0-9]+", "-", hostname.lower()).strip("-")[:32] or "server"
return f"{DEFAULT_ENV_NAME}-{suffix}"
def connect_with_personal_key(
project_root: str | Path | None,
api_base_url: str,
*,
env_name: str = DEFAULT_ENV_NAME,
runner: Runner = subprocess.run,
which: Which = shutil.which,
stdin_isatty: Callable[[], bool] = sys.stdin.isatty,
) -> dict[str, Any]:
"""Bind a personal API Key through nb's private TTY prompt.
The key is never accepted by this function, placed in argv, inherited from
the process environment, captured, printed, or returned. The child nb CLI
reads it directly from the user-controlled terminal.
"""
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"
)
if not stdin_isatty():
raise InstallerError(
"Secure API Key entry requires a user-controlled local terminal. "
"Do not paste the Key into Agent chat or pass it with --access-token."
)
root = _project_root(project_root)
normalized_url = normalize_api_base_url(api_base_url)
discovered_nb = which("nb")
if not discovered_nb:
raise InstallerError("nb was not found on PATH")
nb_bin = os.path.abspath(discovered_nb)
existing = _read_nb_environment(root, env_name, nb_bin=nb_bin, runner=runner)
if existing is None:
interactive_command = [
nb_bin,
"env",
"add",
env_name,
"--api-base-url",
normalized_url,
"--auth-type",
"token",
]
action = "created"
else:
existing_url, _auth_type = _api_details(existing)
if existing_url != normalized_url.rstrip("/"):
suggestion = _suggest_env_name(normalized_url)
raise InstallerError(
f"Environment {env_name!r} already targets {existing_url or 'an unconfigured endpoint'!r}; "
f"use a new name such as {suggestion!r} instead of overwriting it"
)
interactive_command = [nb_bin, "env", "auth", env_name, "--auth-type", "token"]
action = "reauthenticated"
try:
result = runner(
interactive_command,
cwd=str(root),
env=_nb_command_environment(nb_bin),
text=True,
check=False,
)
except OSError as error:
raise InstallerError(f"Could not execute the interactive nb Key prompt: {error}") from error
if result.returncode != 0:
raise InstallerError(
f"Interactive API Key authentication failed with exit code {result.returncode}; "
"revoke the exposed or invalid Key if necessary and retry through the installer"
)
verified = _read_nb_environment(root, env_name, nb_bin=nb_bin, runner=runner)
if verified is None:
raise InstallerError("nb reported success but the saved environment could not be read back")
verified_url, auth_type = _api_details(verified)
if verified_url != normalized_url.rstrip("/") or auth_type != "token":
raise InstallerError("The saved nb environment did not verify as token authentication for the requested URL")
list_result = runner(
[nb_bin, "env", "list"],
cwd=str(root),
env=_nb_command_environment(nb_bin),
capture_output=True,
text=True,
check=False,
)
if list_result.returncode != 0:
raise InstallerError("API Key was saved, but nb env list readback failed")
config_result = set_config_env(root, env_name)
return {
"ok": True,
"action": action,
"env": env_name,
"apiBaseUrl": normalized_url,
"authType": "token",
"keyExposure": "private-tty-only",
"configChanged": config_result["changed"],
"configBackupPath": config_result["backupPath"],
}
scripts/agent_installer/cli.py
-107
View File
@@ -1,107 +0,0 @@
from __future__ import annotations
from .auth import connect_with_personal_key
from .common import (
DEFAULT_ENV_NAME,
Any,
InstallerError,
Path,
Sequence,
argparse,
json,
)
from .diagnostics import doctor_project
from .flow import install_project
from .project import inspect_project, prepare_project, render_client_config, set_config_env
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']}")
if "status" in payload:
print(f"Status: {payload['status']}")
if "clientConfig" in payload:
print("Client configuration:")
print(json.dumps(payload["clientConfig"], ensure_ascii=False, indent=2))
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)
install = subparsers.add_parser("install", help="Run the complete guided local installation flow.")
install.add_argument("--api-base-url", required=True)
install.add_argument("--env-name", default=DEFAULT_ENV_NAME)
install.add_argument("--enable-writes", action="store_true")
install.add_argument("--json", action="store_true", dest="as_json")
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")
connect_key = subparsers.add_parser("connect-key")
connect_key.add_argument("--api-base-url", required=True)
connect_key.add_argument("--env-name", default=DEFAULT_ENV_NAME)
connect_key.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 == "install":
payload = install_project(
args.project_root,
args.api_base_url,
env_name=args.env_name,
enable_writes=args.enable_writes,
)
_print_result(payload, as_json=args.as_json)
return 0 if payload["ok"] else 1
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 == "connect-key":
payload = connect_with_personal_key(
args.project_root,
args.api_base_url,
env_name=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}")
scripts/agent_installer/common.py
-89
View File
@@ -1,89 +0,0 @@
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
from urllib.parse import urlsplit, urlunsplit
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}$")
ANSI_ESCAPE_PATTERN = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
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 _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
__all__ = [
"ANSI_ESCAPE_PATTERN",
"Any",
"Callable",
"Check",
"DEFAULT_ENV_NAME",
"ENV_NAME_PATTERN",
"InstallerError",
"MIN_PYTHON",
"Path",
"REQUIRED_PROJECT_FILES",
"Runner",
"SERVER_NAME",
"Sequence",
"Which",
"_load_json_object",
"_project_root",
"argparse",
"asdict",
"json",
"os",
"re",
"shutil",
"subprocess",
"sys",
"tempfile",
"urlsplit",
"urlunsplit",
]
scripts/agent_installer/diagnostics.py
-83
View File
@@ -1,83 +0,0 @@
from __future__ import annotations
from .common import (
Any,
Check,
InstallerError,
Path,
Runner,
Which,
_load_json_object,
_project_root,
asdict,
shutil,
subprocess,
)
from .project import _venv_python_path, inspect_project
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"],
[
str(_venv_python_path(root)),
"-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": [
"Have the installation Agent verify the saved env through nb CLI's default redacted info output.",
"Register the generated stdio MCP configuration in the user's client.",
"Call the MCP tool crm_health and require ok=true.",
],
}
scripts/agent_installer/flow.py
-107
View File
@@ -1,107 +0,0 @@
from __future__ import annotations
from .auth import connect_with_personal_key, normalize_api_base_url
from .common import (
DEFAULT_ENV_NAME,
ENV_NAME_PATTERN,
SERVER_NAME,
Any,
Callable,
InstallerError,
Path,
Runner,
Which,
_project_root,
shutil,
subprocess,
sys,
)
from .diagnostics import doctor_project
from .project import inspect_project, prepare_project, render_client_config
def install_project(
project_root: str | Path | None,
api_base_url: str,
*,
env_name: str = DEFAULT_ENV_NAME,
enable_writes: bool = False,
runner: Runner = subprocess.run,
which: Which = shutil.which,
python_version: tuple[int, int, int] | None = None,
stdin_isatty: Callable[[], bool] = sys.stdin.isatty,
) -> dict[str, Any]:
"""Prepare, authenticate, diagnose, and describe one local installation.
The returned client configuration is intentionally not written here: the
installation Agent must merge it into its own client-specific format after
showing the destination and creating a recoverable backup.
"""
root = _project_root(project_root)
normalized_url = normalize_api_base_url(api_base_url)
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"
)
if not stdin_isatty():
raise InstallerError(
"Secure API Key entry requires a user-controlled local terminal. "
"Do not paste the Key into Agent chat or pass it with --access-token."
)
inspection = inspect_project(root, which=which, python_version=python_version)
failed_prerequisites = [check["detail"] for check in inspection["checks"] if check["required"] and not check["ok"]]
if failed_prerequisites:
raise InstallerError("Installation prerequisites are not ready: " + "; ".join(failed_prerequisites))
prepared = prepare_project(
root,
runner=runner,
which=which,
python_version=python_version,
)
authentication = connect_with_personal_key(
root,
normalized_url,
env_name=env_name,
runner=runner,
which=which,
stdin_isatty=stdin_isatty,
)
diagnosis = doctor_project(
root,
runner=runner,
which=which,
python_version=python_version,
)
client_config = render_client_config(root, enable_writes=enable_writes, which=which)
local_ready = bool(diagnosis["ok"])
return {
"ok": local_ready,
"status": "ready-for-client-registration" if local_ready else "local-checks-failed",
"projectRoot": str(root),
"mode": "project-writes" if enable_writes else "read-only",
"prepared": {
"environmentPath": prepared["environmentPath"],
"dependencyProfile": prepared["dependencies"],
"configCreated": prepared["configCreated"],
},
"authentication": authentication,
"checks": diagnosis["checks"],
"clientConfig": client_config,
"clientRegistration": {
"required": True,
"serverName": SERVER_NAME,
"backupRequired": True,
"action": "merge the generated server entry, reload the client, then call crm_health",
},
"remainingActions": [
"Merge clientConfig into the current MCP client without replacing other servers.",
"Reload or restart the MCP client.",
"Call crm_health and require ok=true before reporting installation complete.",
],
}
__all__ = ["install_project"]
scripts/agent_installer/project.py
-212
View File
@@ -1,212 +0,0 @@
from __future__ import annotations
from .common import (
ENV_NAME_PATTERN,
MIN_PYTHON,
REQUIRED_PROJECT_FILES,
SERVER_NAME,
Any,
Check,
InstallerError,
Path,
Runner,
Which,
_load_json_object,
_project_root,
asdict,
json,
os,
shutil,
subprocess,
sys,
tempfile,
)
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 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
venv_path = root / ".venv"
command = [uv_bin, "sync", "--locked", "--no-dev", "--python", f"{MIN_PYTHON[0]}.{MIN_PYTHON[1]}"]
command_environment = os.environ.copy()
command_environment["UV_PROJECT_ENVIRONMENT"] = str(venv_path)
try:
result = runner(
command,
cwd=str(root),
env=command_environment,
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-runtime-only",
"environmentPath": str(venv_path),
}
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,
os_name: str | None = None,
) -> dict[str, Any]:
root = _project_root(project_root)
nb_bin = which("nb") or "nb"
if os.path.isabs(nb_bin):
nb_bin = os.path.abspath(nb_bin)
python_bin = _venv_python_path(root, os_name=os_name)
args = [
"-m",
"nocobase_crm_mcp.server",
"--config",
str(root / "config.json"),
]
if enable_writes:
args.append("--enable-writes")
return {
"mcpServers": {
SERVER_NAME: {
"command": str(python_bin),
"args": args,
"env": {"NOCOBASE_MCP_NB_BIN": nb_bin},
}
}
}
def _venv_python_path(root: Path, *, os_name: str | None = None) -> Path:
if (os_name or os.name) == "nt":
return root / ".venv" / "Scripts" / "python.exe"
return root / ".venv" / "bin" / "python"
scripts/build.mjs
+19
View File
@@ -0,0 +1,19 @@
import { chmod, mkdir, writeFile } from "node:fs/promises";
import { build } from "esbuild";
await mkdir("dist", { recursive: true });
await build({
entryPoints: ["src/index.ts"],
outfile: "dist/index.js",
bundle: true,
platform: "node",
format: "esm",
target: "node20",
sourcemap: true,
legalComments: "none",
banner: { js: "#!/usr/bin/env node" },
});
await chmod("dist/index.js", 0o755);
// A legacy Python build may have left dist/.gitignore containing "*". npm uses
// this narrower file to include only the new bundled runtime in Release assets.
await writeFile("dist/.npmignore", "*\n!index.js\n!index.js.map\n", "utf8");
scripts/manage_mcp_service.py
-99
View File
@@ -1,99 +0,0 @@
from __future__ import annotations
from mcp_service_cli import main
from mcp_service_config import (
API_BASE_URL_ENV,
API_TOKEN_ENV,
DEFAULT_CONFIG_PATH,
DEFAULT_HOST,
DEFAULT_NB_BIN,
DEFAULT_NB_ENV,
DEFAULT_PATH,
DEFAULT_PORT,
DEFAULT_TOKEN_BYTES,
DEFAULT_UPLOAD_MAX_BYTES,
DEFAULT_UV_CACHE_DIR,
ENABLE_UPLOADS_ENV,
ENV_RELATIVE_PATH,
LEGACY_API_BASE_URL_ENV,
LEGACY_API_TOKEN_ENV,
MANAGED_ENV_NAMES,
SERVICE_NAME,
ServiceConfig,
ServicePaths,
find_uv_path,
read_service_env_file,
render_service_unit,
resolve_service_paths,
saved_service_config,
systemd_quote,
write_token_env_file,
)
from mcp_service_runtime import (
detect_remote_hosts,
find_nb_path,
generate_token,
new_token,
probe_http_service,
render_remote_config,
restart_service_if_installed,
run_systemctl,
service_state,
service_status,
show_config,
start_service,
stop_service,
uninstall_service,
wait_for_http_service,
write_service_unit,
)
__all__ = [
"SERVICE_NAME",
"ENV_RELATIVE_PATH",
"DEFAULT_HOST",
"DEFAULT_PORT",
"DEFAULT_PATH",
"DEFAULT_NB_ENV",
"DEFAULT_CONFIG_PATH",
"DEFAULT_NB_BIN",
"DEFAULT_TOKEN_BYTES",
"DEFAULT_UV_CACHE_DIR",
"DEFAULT_UPLOAD_MAX_BYTES",
"ENABLE_UPLOADS_ENV",
"API_BASE_URL_ENV",
"API_TOKEN_ENV",
"LEGACY_API_BASE_URL_ENV",
"LEGACY_API_TOKEN_ENV",
"MANAGED_ENV_NAMES",
"ServicePaths",
"ServiceConfig",
"systemd_quote",
"find_uv_path",
"resolve_service_paths",
"render_service_unit",
"write_token_env_file",
"read_service_env_file",
"saved_service_config",
"generate_token",
"find_nb_path",
"write_service_unit",
"detect_remote_hosts",
"render_remote_config",
"show_config",
"run_systemctl",
"service_state",
"probe_http_service",
"wait_for_http_service",
"restart_service_if_installed",
"new_token",
"start_service",
"stop_service",
"uninstall_service",
"service_status",
"main",
]
if __name__ == "__main__":
raise SystemExit(main())
scripts/mcp_service.sh
-19
View File
@@ -1,19 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
if [[ $# -eq 0 ]]; then
exec python3 "$PROJECT_ROOT/scripts/manage_mcp_service.py" --help
fi
COMMAND="$1"
shift
case "$COMMAND" in
new-token | start | show-config | status | uninstall)
exec python3 "$PROJECT_ROOT/scripts/manage_mcp_service.py" "$COMMAND" --project-root "$PROJECT_ROOT" "$@"
;;
*)
exec python3 "$PROJECT_ROOT/scripts/manage_mcp_service.py" "$COMMAND" "$@"
;;
esac
scripts/mcp_service_cli.py
-89
View File
@@ -1,89 +0,0 @@
from __future__ import annotations
import argparse
from collections.abc import Sequence
from pathlib import Path
from mcp_service_config import (
DEFAULT_CONFIG_PATH,
DEFAULT_HOST,
DEFAULT_NB_ENV,
DEFAULT_PATH,
DEFAULT_PORT,
DEFAULT_TOKEN_BYTES,
SERVICE_NAME,
)
from mcp_service_runtime import new_token, service_status, show_config, start_service, stop_service, uninstall_service
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Manage the HTTP NocoBase CRM MCP user service.")
subparsers = parser.add_subparsers(dest="command", required=True)
new_token_parser = subparsers.add_parser(
"new-token", help="Generate a new token and print local MCP client configuration."
)
new_token_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
new_token_parser.add_argument("--host", default=DEFAULT_HOST)
new_token_parser.add_argument("--port", default=DEFAULT_PORT, type=int)
new_token_parser.add_argument("--path", default=DEFAULT_PATH)
new_token_parser.add_argument("--nb-env", default=DEFAULT_NB_ENV)
new_token_parser.add_argument("--config-path", default=DEFAULT_CONFIG_PATH)
new_token_parser.add_argument("--token-bytes", default=DEFAULT_TOKEN_BYTES, type=int)
start_parser = subparsers.add_parser(
"start", help="Install or update the user service using the saved token, then start it."
)
start_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
subparsers.add_parser("stop", help="Stop the user service without deleting service files.")
uninstall_parser = subparsers.add_parser(
"uninstall", help="Stop and remove the managed user service and token file."
)
uninstall_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
show_config_parser = subparsers.add_parser(
"show-config", help="Print the saved token and local MCP client configuration."
)
show_config_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
status_parser = subparsers.add_parser("status", help="Show systemd status for the managed service.")
status_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
args = parser.parse_args(argv)
if args.command == "new-token":
print(
new_token(
project_root=args.project_root,
host=args.host,
port=args.port,
path=args.path,
nb_env=args.nb_env,
config_path=args.config_path,
token_bytes=args.token_bytes,
),
end="",
)
return 0
if args.command == "start":
print(start_service(project_root=args.project_root), end="")
return 0
if args.command == "stop":
stop_service()
print(f"Stopped {SERVICE_NAME}.")
return 0
if args.command == "uninstall":
uninstall_service(args.project_root)
print(f"Uninstalled {SERVICE_NAME}.")
return 0
if args.command == "show-config":
print(show_config(args.project_root), end="")
return 0
if args.command == "status":
print(service_status(project_root=args.project_root), end="")
return 0
raise AssertionError(f"unhandled command: {args.command}")
__all__ = ["main"]
scripts/mcp_service_config.py
-264
View File
@@ -1,264 +0,0 @@
from __future__ import annotations
import os
import shlex
import shutil
from dataclasses import dataclass
from pathlib import Path
SERVICE_NAME = "nocobase-crm-mcp.service"
ENV_RELATIVE_PATH = Path(".runtime") / "nocobase-crm-mcp.env"
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 10087
DEFAULT_PATH = "/mcp"
DEFAULT_NB_ENV = "nb-mcp"
DEFAULT_CONFIG_PATH = "config.json"
DEFAULT_NB_BIN = "nb"
DEFAULT_TOKEN_BYTES = 32
DEFAULT_UV_CACHE_DIR = "/tmp/uv-cache"
DEFAULT_UPLOAD_MAX_BYTES = 25 * 1024 * 1024
ENABLE_UPLOADS_ENV = "NOCOBASE_MCP_ENABLE_UPLOADS"
API_BASE_URL_ENV = "NOCOBASE_MCP_API_BASE_URL"
API_TOKEN_ENV = "NOCOBASE_MCP_API_TOKEN"
LEGACY_API_BASE_URL_ENV = "NOCOBASE_MCP_UPLOAD_API_BASE_URL"
LEGACY_API_TOKEN_ENV = "NOCOBASE_MCP_UPLOAD_JWT"
MANAGED_ENV_NAMES = frozenset(
{
"NOCOBASE_MCP_TOKEN",
"NOCOBASE_MCP_HOST",
"NOCOBASE_MCP_PORT",
"NOCOBASE_MCP_PATH",
"NOCOBASE_MCP_ENV",
"NOCOBASE_MCP_CONFIG",
"NOCOBASE_MCP_NB_BIN",
ENABLE_UPLOADS_ENV,
"NOCOBASE_MCP_UPLOAD_MAX_BYTES",
}
)
@dataclass(frozen=True)
class ServicePaths:
project_root: Path
unit_path: Path
env_path: Path
@dataclass(frozen=True)
class ServiceConfig:
token: str
host: str = DEFAULT_HOST
port: int = DEFAULT_PORT
path: str = DEFAULT_PATH
nb_env: str = DEFAULT_NB_ENV
config_path: str = DEFAULT_CONFIG_PATH
nb_bin: str = DEFAULT_NB_BIN
def systemd_quote(value: str | Path) -> str:
return shlex.quote(str(value))
def find_uv_path() -> Path:
configured = os.environ.get("UV_PATH", "").strip()
discovered = configured or shutil.which("uv")
if not discovered:
raise RuntimeError("Could not find uv on PATH. Install uv or set UV_PATH to its absolute path.")
return Path(discovered).expanduser().resolve()
def resolve_service_paths(project_root: Path, config_home: Path | None = None) -> ServicePaths:
root = project_root.expanduser().resolve()
base_config = config_home or Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
unit_path = base_config.expanduser() / "systemd" / "user" / SERVICE_NAME
return ServicePaths(project_root=root, unit_path=unit_path, env_path=root / ENV_RELATIVE_PATH)
def render_service_unit(
*,
project_root: Path,
env_file: Path,
host: str = DEFAULT_HOST,
port: int = DEFAULT_PORT,
path: str = DEFAULT_PATH,
uv_cache_dir: str = DEFAULT_UV_CACHE_DIR,
uv_path: Path | None = None,
) -> str:
resolved_uv_path = uv_path or find_uv_path()
command = " ".join(
[
systemd_quote(resolved_uv_path),
"run",
"--no-sync",
"nocobase-crm-mcp",
"--transport",
"streamable-http",
"--host",
host,
"--port",
str(port),
"--path",
path,
]
)
return (
"[Unit]\n"
"Description=NocoBase CRM HTTP MCP server\n"
"After=network-online.target\n"
"Wants=network-online.target\n"
"\n"
"[Service]\n"
"Type=simple\n"
f"WorkingDirectory={systemd_quote(project_root)}\n"
f"EnvironmentFile={systemd_quote(env_file)}\n"
f"Environment=UV_CACHE_DIR={systemd_quote(uv_cache_dir)}\n"
f"ExecStart={command}\n"
"Restart=on-failure\n"
"RestartSec=5\n"
"\n"
"[Install]\n"
"WantedBy=default.target\n"
)
def write_token_env_file(
env_path: Path,
token: str,
*,
host: str = DEFAULT_HOST,
port: int = DEFAULT_PORT,
path: str = DEFAULT_PATH,
nb_env: str = DEFAULT_NB_ENV,
config_path: str = DEFAULT_CONFIG_PATH,
nb_bin: str | Path = DEFAULT_NB_BIN,
) -> None:
uploads_enabled = _saved_uploads_setting(env_path)
preserved_lines = _operator_env_lines(env_path)
env_path.parent.mkdir(parents=True, exist_ok=True)
env_path.parent.chmod(0o700)
fd = os.open(env_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as env_file:
env_file.write(f"NOCOBASE_MCP_TOKEN={token}\n")
env_file.write(f"NOCOBASE_MCP_HOST={host}\n")
env_file.write(f"NOCOBASE_MCP_PORT={port}\n")
env_file.write(f"NOCOBASE_MCP_PATH={path}\n")
env_file.write(f"NOCOBASE_MCP_ENV={nb_env}\n")
env_file.write(f"NOCOBASE_MCP_CONFIG={config_path}\n")
env_file.write(f"NOCOBASE_MCP_NB_BIN={nb_bin}\n")
env_file.write(f"{ENABLE_UPLOADS_ENV}={uploads_enabled}\n")
env_file.write(f"NOCOBASE_MCP_UPLOAD_MAX_BYTES={DEFAULT_UPLOAD_MAX_BYTES}\n")
if preserved_lines:
env_file.write("\n")
env_file.write("\n".join(preserved_lines))
env_file.write("\n")
env_path.chmod(0o600)
def _saved_uploads_setting(env_path: Path) -> str:
if not env_path.exists():
return "0"
for raw_line in env_path.read_text(encoding="utf-8").splitlines():
key, separator, value = raw_line.strip().partition("=")
if separator and key.strip() == ENABLE_UPLOADS_ENV:
return "1" if value.strip() == "1" else "0"
return "0"
def _operator_env_lines(env_path: Path) -> list[str]:
if not env_path.exists():
return []
preserved_lines: list[str] = []
for raw_line in env_path.read_text(encoding="utf-8").splitlines():
key, separator, _value = raw_line.strip().partition("=")
if separator and key.strip() in MANAGED_ENV_NAMES:
continue
preserved_lines.append(raw_line)
return preserved_lines
def read_service_env_file(env_path: Path) -> dict[str, str]:
if not env_path.exists():
raise RuntimeError(
f"Missing MCP service env file: {env_path}. Run scripts/mcp_service.sh start to create it, "
"or scripts/mcp_service.sh new-token to rotate it."
)
values: dict[str, str] = {}
for raw_line in env_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#"):
continue
key, separator, value = line.partition("=")
if separator:
values[key.strip()] = value.strip()
return values
def saved_service_config(project_root: Path) -> ServiceConfig:
paths = resolve_service_paths(project_root)
values = read_service_env_file(paths.env_path)
token = values.get("NOCOBASE_MCP_TOKEN", "").strip()
if not token:
raise RuntimeError(f"Missing NOCOBASE_MCP_TOKEN in {paths.env_path}")
host = values.get("NOCOBASE_MCP_HOST") or DEFAULT_HOST
port = int(values.get("NOCOBASE_MCP_PORT") or DEFAULT_PORT)
path = values.get("NOCOBASE_MCP_PATH") or DEFAULT_PATH
nb_env = values.get("NOCOBASE_MCP_ENV") or DEFAULT_NB_ENV
config_path = values.get("NOCOBASE_MCP_CONFIG") or DEFAULT_CONFIG_PATH
nb_bin = values.get("NOCOBASE_MCP_NB_BIN") or DEFAULT_NB_BIN
return ServiceConfig(
token=token,
host=host,
port=port,
path=path,
nb_env=nb_env,
config_path=config_path,
nb_bin=nb_bin,
)
__all__ = [
"API_BASE_URL_ENV",
"API_TOKEN_ENV",
"DEFAULT_CONFIG_PATH",
"DEFAULT_HOST",
"DEFAULT_NB_BIN",
"DEFAULT_NB_ENV",
"DEFAULT_PATH",
"DEFAULT_PORT",
"DEFAULT_TOKEN_BYTES",
"DEFAULT_UPLOAD_MAX_BYTES",
"DEFAULT_UV_CACHE_DIR",
"ENABLE_UPLOADS_ENV",
"ENV_RELATIVE_PATH",
"LEGACY_API_BASE_URL_ENV",
"LEGACY_API_TOKEN_ENV",
"MANAGED_ENV_NAMES",
"SERVICE_NAME",
"ServiceConfig",
"ServicePaths",
"_operator_env_lines",
"_saved_uploads_setting",
"find_uv_path",
"read_service_env_file",
"render_service_unit",
"resolve_service_paths",
"saved_service_config",
"systemd_quote",
"write_token_env_file",
]
scripts/mcp_service_runtime.py
-354
View File
@@ -1,354 +0,0 @@
from __future__ import annotations
import os
import secrets
import shutil
import socket
import subprocess
import time
import urllib.error
import urllib.request
from collections.abc import Sequence
from pathlib import Path
from mcp_service_config import (
API_BASE_URL_ENV,
API_TOKEN_ENV,
DEFAULT_CONFIG_PATH,
DEFAULT_HOST,
DEFAULT_NB_ENV,
DEFAULT_PATH,
DEFAULT_PORT,
DEFAULT_TOKEN_BYTES,
ENABLE_UPLOADS_ENV,
LEGACY_API_BASE_URL_ENV,
LEGACY_API_TOKEN_ENV,
SERVICE_NAME,
ServicePaths,
find_uv_path,
read_service_env_file,
render_service_unit,
resolve_service_paths,
saved_service_config,
write_token_env_file,
)
def generate_token(num_bytes: int = DEFAULT_TOKEN_BYTES) -> str:
return secrets.token_urlsafe(num_bytes)
def find_nb_path() -> Path:
configured = os.environ.get("NOCOBASE_MCP_NB_BIN", "").strip()
candidate = configured or "nb"
candidate_path = Path(candidate).expanduser()
if candidate_path.is_absolute():
if candidate_path.is_file() and os.access(candidate_path, os.X_OK):
return Path(os.path.abspath(candidate_path))
raise RuntimeError(f"NOCOBASE_MCP_NB_BIN points to {candidate_path}, but it is not an executable file.")
discovered = shutil.which(candidate)
if not discovered:
raise RuntimeError(
"Could not find nb on PATH. Install the NocoBase CLI or set NOCOBASE_MCP_NB_BIN to its absolute path."
)
return Path(os.path.abspath(str(Path(discovered).expanduser())))
def write_service_unit(
paths: ServicePaths, *, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, path: str = DEFAULT_PATH
) -> None:
paths.unit_path.parent.mkdir(parents=True, exist_ok=True)
unit = render_service_unit(
project_root=paths.project_root,
env_file=paths.env_path,
host=host,
port=port,
path=path,
uv_path=find_uv_path(),
)
paths.unit_path.write_text(unit, encoding="utf-8")
def detect_remote_hosts() -> list[str]:
hosts: list[str] = []
try:
result = subprocess.run(["hostname", "-I"], capture_output=True, text=True, check=False)
except OSError:
result = None
if result and result.returncode == 0:
hosts.extend(result.stdout.split())
try:
for info in socket.getaddrinfo(socket.gethostname(), None):
hosts.append(info[4][0])
except OSError:
pass
cleaned: list[str] = []
for host in hosts:
normalized = host.strip()
if not normalized or normalized.startswith("127.") or normalized == "::1" or normalized.startswith("fe80:"):
continue
if normalized not in cleaned:
cleaned.append(normalized)
return cleaned
def render_remote_config(
*,
hosts: Sequence[str],
token: str,
port: int = DEFAULT_PORT,
path: str = DEFAULT_PATH,
service_status: str | None = None,
) -> str:
host = hosts[0] if hosts else "127.0.0.1"
url = f"http://{host}:{port}{path}"
alternatives = "\n".join(f"- http://{candidate}:{port}{path}" for candidate in hosts[1:])
alternatives_section = f"\nOther detected addresses:\n{alternatives}\n" if alternatives else ""
status_section = f"Service status: {service_status}\n\n" if service_status else ""
return (
"Saved MCP client config.\n"
"\n"
f"{status_section}"
"Use this loopback URL on this machine:\n"
f"{url}\n"
f"{alternatives_section}"
"\n"
"Use this header:\n"
f"Authorization: Bearer {token}\n"
"\n"
"Example MCP client config:\n"
"{\n"
' "mcpServers": {\n'
' "nocobase-crm": {\n'
' "type": "streamable-http",\n'
f' "url": "{url}",\n'
' "headers": {\n'
f' "Authorization": "Bearer {token}"\n'
" }\n"
" }\n"
" }\n"
"}\n"
)
def show_config(project_root: Path) -> str:
config = saved_service_config(project_root)
display_host = "127.0.0.1" if config.host in {"0.0.0.0", "::"} else config.host
return render_remote_config(
hosts=[display_host],
token=config.token,
port=config.port,
path=config.path,
service_status=service_state(),
)
def run_systemctl(args: Sequence[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
return subprocess.run(["systemctl", "--user", *args], text=True, check=check)
def service_state() -> str:
result = subprocess.run(
["systemctl", "--user", "is-active", SERVICE_NAME],
capture_output=True,
text=True,
check=False,
)
state = result.stdout.strip()
if state:
return state
if result.returncode == 0:
return "active"
return "unknown"
def probe_http_service(host: str, port: int, path: str, *, token: str, timeout: float = 2.0) -> dict[str, object]:
url = f"http://{host}:{port}{path}"
request = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"}, method="GET")
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
return {"reachable": True, "status": response.status, "url": url}
except urllib.error.HTTPError as error:
return {"reachable": True, "status": error.code, "url": url}
except OSError as error:
return {"reachable": False, "error": str(error), "url": url}
def wait_for_http_service(
host: str,
port: int,
path: str,
*,
token: str,
attempts: int = 20,
interval_seconds: float = 0.25,
) -> dict[str, object]:
last_probe: dict[str, object] = {"reachable": False, "error": "not probed"}
for attempt in range(max(1, attempts)):
last_probe = probe_http_service(host, port, path, token=token)
if last_probe.get("reachable"):
return last_probe
if attempt < attempts - 1:
time.sleep(interval_seconds)
return last_probe
def restart_service_if_installed(paths: ServicePaths) -> bool:
if not paths.unit_path.exists():
return False
run_systemctl(["restart", SERVICE_NAME])
return True
def new_token(
*,
project_root: Path,
host: str = DEFAULT_HOST,
port: int = DEFAULT_PORT,
path: str = DEFAULT_PATH,
nb_env: str = DEFAULT_NB_ENV,
config_path: str = DEFAULT_CONFIG_PATH,
token_bytes: int = DEFAULT_TOKEN_BYTES,
) -> str:
paths = resolve_service_paths(project_root)
token = generate_token(token_bytes)
write_token_env_file(
paths.env_path,
token,
host=host,
port=port,
path=path,
nb_env=nb_env,
config_path=config_path,
nb_bin=find_nb_path(),
)
restart_service_if_installed(paths)
display_host = "127.0.0.1" if host in {"0.0.0.0", "::"} else host
return render_remote_config(hosts=[display_host], token=token, port=port, path=path, service_status=service_state())
def start_service(*, project_root: Path) -> str:
paths = resolve_service_paths(project_root)
nb_bin = find_nb_path()
if not paths.env_path.exists():
write_token_env_file(
paths.env_path,
generate_token(),
host=DEFAULT_HOST,
port=DEFAULT_PORT,
path=DEFAULT_PATH,
nb_env=DEFAULT_NB_ENV,
config_path=DEFAULT_CONFIG_PATH,
nb_bin=nb_bin,
)
config = saved_service_config(project_root)
write_token_env_file(
paths.env_path,
config.token,
host=config.host,
port=config.port,
path=config.path,
nb_env=config.nb_env,
config_path=config.config_path,
nb_bin=nb_bin,
)
write_service_unit(paths, host=config.host, port=config.port, path=config.path)
run_systemctl(["daemon-reload"])
run_systemctl(["enable", SERVICE_NAME])
run_systemctl(["restart", SERVICE_NAME])
try:
run_systemctl(["is-active", "--quiet", SERVICE_NAME])
except subprocess.CalledProcessError as error:
raise RuntimeError(f"{SERVICE_NAME} did not become active after restart") from error
probe = wait_for_http_service("127.0.0.1", config.port, config.path, token=config.token)
if not probe.get("reachable"):
raise RuntimeError(f"{SERVICE_NAME} is active but HTTP probe failed: {probe.get('error')}")
display_host = "127.0.0.1" if config.host in {"0.0.0.0", "::"} else config.host
return render_remote_config(
hosts=[display_host],
token=config.token,
port=config.port,
path=config.path,
service_status="active",
)
def stop_service() -> None:
run_systemctl(["stop", SERVICE_NAME])
def uninstall_service(project_root: Path) -> None:
paths = resolve_service_paths(project_root)
run_systemctl(["disable", "--now", SERVICE_NAME], check=False)
if paths.unit_path.exists():
paths.unit_path.unlink()
if paths.env_path.exists():
paths.env_path.unlink()
try:
paths.env_path.parent.rmdir()
except OSError:
pass
run_systemctl(["daemon-reload"], check=False)
run_systemctl(["reset-failed", SERVICE_NAME], check=False)
def service_status(*, project_root: Path) -> str:
paths = resolve_service_paths(project_root)
state = service_state()
lines = [
f"Service status: {state}",
f"Unit file: {paths.unit_path}",
f"Env file: {paths.env_path}",
]
if paths.env_path.exists():
values = read_service_env_file(paths.env_path)
config = saved_service_config(project_root)
display_host = "127.0.0.1" if config.host in {"0.0.0.0", "::"} else config.host
lines.append(f"MCP URL: http://{display_host}:{config.port}{config.path}")
lines.append(f"nb binary: {config.nb_bin}")
lines.append(f"NocoBase CLI env: {config.nb_env} (personal API Key managed by the Agent installer)")
uploads_enabled = values.get(ENABLE_UPLOADS_ENV) == "1"
lines.append(f"Attachment uploads: {'enabled' if uploads_enabled else 'disabled (safe default)'}")
canonical_configured = bool(values.get(API_BASE_URL_ENV) and values.get(API_TOKEN_ENV))
legacy_configured = bool(values.get(LEGACY_API_BASE_URL_ENV) and values.get(LEGACY_API_TOKEN_ENV))
if canonical_configured:
credential_state = "configured (canonical)"
elif legacy_configured:
credential_state = "configured (legacy)"
elif any(
values.get(name)
for name in (API_BASE_URL_ENV, API_TOKEN_ENV, LEGACY_API_BASE_URL_ENV, LEGACY_API_TOKEN_ENV)
):
credential_state = "incomplete"
else:
credential_state = "not configured; using the local nb CLI login"
lines.append(f"Legacy NocoBase credential override: {credential_state}")
probe = probe_http_service("127.0.0.1", config.port, config.path, token=config.token)
if probe.get("reachable"):
lines.append(f"HTTP probe: reachable (status {probe.get('status')})")
else:
lines.append(f"HTTP probe: not reachable ({probe.get('error')})")
else:
lines.append("MCP URL: unavailable until first start creates the env file")
lines.append("HTTP probe: skipped")
return "\n".join(lines) + "\n"
__all__ = [
"generate_token",
"find_nb_path",
"write_service_unit",
"detect_remote_hosts",
"render_remote_config",
"show_config",
"run_systemctl",
"service_state",
"probe_http_service",
"wait_for_http_service",
"restart_service_if_installed",
"new_token",
"start_service",
"stop_service",
"uninstall_service",
"service_status",
]
scripts/verify-release-version.mjs
+12
View File
@@ -0,0 +1,12 @@
import { readFile } from "node:fs/promises";
const tag = process.argv[2];
if (!tag?.startsWith("v")) throw new Error("Release tag must have the form v<version>");
const packageJson = JSON.parse(await readFile(new URL("../package.json", import.meta.url), "utf8"));
const manifest = JSON.parse(await readFile(new URL("../install-manifest.json", import.meta.url), "utf8"));
const expected = tag.slice(1);
if (packageJson.version !== expected || manifest.version !== expected) {
throw new Error(
`Version mismatch: tag=${tag}, package=${packageJson.version}, manifest=${manifest.version}`,
);
}