552 lines
21 KiB
Python
552 lines
21 KiB
Python
"""Check an existing checkout and synchronize installed local MCP clients."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import tomllib
|
|
import urllib.request
|
|
from urllib.parse import urlsplit
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def launch_text(english: str, chinese: str) -> str:
|
|
return chinese if os.environ.get("ARTEMIS_LAUNCH_LANG") == "zh-CN" else english
|
|
|
|
|
|
def git(root: Path, *args: str, timeout: int = 15) -> str:
|
|
env = {**os.environ, "GIT_TERMINAL_PROMPT": "0", "GCM_INTERACTIVE": "Never"}
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), *args],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
env=env,
|
|
timeout=timeout,
|
|
)
|
|
if result.returncode:
|
|
# Remote diagnostics can contain URLs/userinfo. Keep them out of launcher logs.
|
|
raise RuntimeError(
|
|
f"git {args[0]} failed (exit {result.returncode}); check network, credentials or repository state in Git."
|
|
)
|
|
return result.stdout.strip()
|
|
|
|
|
|
def update_source(root: Path, port: int = 8000) -> int:
|
|
"""Return 10 after a fast-forward (caller reloads launcher); never overwrite local work."""
|
|
try:
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
|
with opener.open(f"http://127.0.0.1:{port}/api/status", timeout=2) as response:
|
|
status = json.load(response)
|
|
if (
|
|
status.get("active_tasks")
|
|
or status.get("queue")
|
|
or (status.get("status") == "running" and status.get("pid"))
|
|
):
|
|
print(
|
|
launch_text(
|
|
"[WAIT] Tasks are active. Finish/stop them before updating dependencies or restarting.",
|
|
"[等待] 有任务正在执行或排队。请先处理任务，再更新依赖或重启。",
|
|
)
|
|
)
|
|
return 20
|
|
except (OSError, ValueError):
|
|
pass
|
|
if not shutil.which("git"):
|
|
print(
|
|
launch_text(
|
|
"[SKIP] Git is not installed. Continuing with local source; install Git for automatic source updates.",
|
|
"[跳过] 未安装 Git，继续使用当前源码；安装 Git 后可自动检查更新。",
|
|
)
|
|
)
|
|
return 0
|
|
if not (root / ".git").exists():
|
|
print(
|
|
launch_text(
|
|
"[SKIP] This older source ZIP has no Git metadata. Use a new release ZIP or Git clone for automatic updates; local files are preserved.",
|
|
"[跳过] 旧版源码包不含 Git 信息。请使用新版源码包或 Git 克隆；当前文件保持不变。",
|
|
)
|
|
)
|
|
return 0
|
|
try:
|
|
branch = git(root, "symbolic-ref", "--quiet", "--short", "HEAD")
|
|
remote = git(root, "config", "--get", f"branch.{branch}.remote")
|
|
merge_ref = git(root, "config", "--get", f"branch.{branch}.merge")
|
|
if remote == "." or remote.startswith("-") or not merge_ref.startswith("refs/heads/"):
|
|
raise RuntimeError(
|
|
"No remote tracking branch; configure an upstream before enabling source updates."
|
|
)
|
|
url = git(root, "remote", "get-url", remote)
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme in {"http", "https"} and (parsed.username or parsed.password):
|
|
raise RuntimeError(
|
|
"Use a clean Git remote URL and a credential manager, not embedded credentials."
|
|
)
|
|
print(
|
|
launch_text(
|
|
"[CHECK] Checking the current branch's configured upstream...",
|
|
"[检查] 正在检查当前分支的远程更新...",
|
|
)
|
|
)
|
|
git(root, "fetch", "--no-tags", remote, merge_ref, timeout=45)
|
|
target = git(root, "rev-parse", "FETCH_HEAD")
|
|
current = git(root, "rev-parse", "HEAD")
|
|
if current == target:
|
|
print(launch_text("[OK] Source is up to date.", "[完成] 源码已是最新版。"))
|
|
return 0
|
|
if git(root, "status", "--porcelain"):
|
|
print(
|
|
launch_text(
|
|
"[SKIP] Upstream differs, but local changes exist. Commit/back up your work before updating; continuing locally.",
|
|
"[跳过] 发现本地修改；请先提交或备份。此次继续使用本地源码。",
|
|
)
|
|
)
|
|
return 0
|
|
ancestor = git(root, "merge-base", "HEAD", target)
|
|
if ancestor != current:
|
|
print(
|
|
launch_text(
|
|
"[SKIP] Local branch is ahead or diverged. No automatic merge/rebase; continuing locally.",
|
|
"[跳过] 本地分支领先或分叉，不自动合并；继续使用本地源码。",
|
|
)
|
|
)
|
|
return 0
|
|
git(root, "merge", "--ff-only", target, timeout=30)
|
|
print(
|
|
launch_text(
|
|
f"[UPDATED] Source {current[:12]} -> {target[:12]}. Reloading launcher and checking dependencies.",
|
|
f"[已更新] 源码 {current[:12]} -> {target[:12]}，重新载入启动脚本并检查依赖。",
|
|
)
|
|
)
|
|
return 10
|
|
except (RuntimeError, subprocess.TimeoutExpired) as exc:
|
|
print(
|
|
launch_text(
|
|
f"[WARN] {exc if isinstance(exc, RuntimeError) else 'Git update timed out.'} Continuing with local source.",
|
|
"[提示] Git 更新失败或超时；请检查网络、凭据和仓库状态。继续使用当前源码。",
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
def _atomic_write(path: Path, text: str) -> None:
|
|
if path.is_file() and path.read_text(encoding="utf-8") == text:
|
|
return
|
|
backup = path.with_name(path.name + ".artemis-update.bak")
|
|
if path.is_file() and not backup.exists():
|
|
shutil.copy2(path, backup)
|
|
fd, name = tempfile.mkstemp(prefix=".artemis-update-", dir=path.parent)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as stream:
|
|
stream.write(text)
|
|
os.replace(name, path)
|
|
finally:
|
|
Path(name).unlink(missing_ok=True)
|
|
|
|
|
|
def _is_standard_local_artemis(config: dict) -> bool:
|
|
if config.get("url") or config.get("type") not in (None, "stdio"):
|
|
return False
|
|
args = config.get("args", [])
|
|
# Running this launcher selects this checkout, including after a user moves
|
|
# a release ZIP to a new directory. Do not take over custom wrappers/SSE.
|
|
return args in (["-m", "mcp_server"], ["-m", "mcp_server.server"])
|
|
|
|
|
|
def _updated_entry(config: dict, template: dict, root: Path, names: tuple[str, ...]) -> dict:
|
|
updated = {**config, "command": template["command"], "args": template["args"]}
|
|
if "cwd" in template or "cwd" in config:
|
|
updated["cwd"] = str(root)
|
|
environment = config.get("env", {})
|
|
if not isinstance(environment, dict):
|
|
raise ValueError("The existing Artemis MCP environment is not an object")
|
|
updated["env"] = {**environment, **template["env"]}
|
|
if "enabled_tools" in template or "enabled_tools" in config:
|
|
updated["enabled_tools"] = list(names)
|
|
if "tools" in template or "tools" in config:
|
|
current_tools = config.get("tools", {})
|
|
if not isinstance(current_tools, dict):
|
|
raise ValueError("The existing Artemis MCP tools setting is not an object")
|
|
updated["tools"] = {name: {"eager": True, **current_tools.get(name, {})} for name in names}
|
|
for key in ("type", "required", "startup_timeout_sec", "connectionTimeoutMs"):
|
|
if key in template and key not in updated:
|
|
updated[key] = template[key]
|
|
return updated
|
|
|
|
|
|
def refresh_json(
|
|
path: Path,
|
|
keys: tuple[str, ...],
|
|
root: Path,
|
|
python: str,
|
|
names: tuple[str, ...],
|
|
template: dict | None = None,
|
|
install_missing: bool = False,
|
|
) -> bool | None:
|
|
if not path.is_file() and not install_missing:
|
|
return False
|
|
raw = path.read_text(encoding="utf-8") if path.is_file() else "{}"
|
|
data = json.loads(raw)
|
|
if not isinstance(data, dict):
|
|
raise ValueError("MCP configuration must be a JSON object")
|
|
parent = data
|
|
for key in keys:
|
|
child = parent.get(key)
|
|
if child is None and install_missing:
|
|
child = parent[key] = {}
|
|
if not isinstance(child, dict):
|
|
raise ValueError(f"Invalid MCP configuration section: {key}")
|
|
parent = child
|
|
config = parent.get("artemis")
|
|
if config is None and install_missing:
|
|
if template is None:
|
|
raise ValueError("A standard MCP template is required for installation")
|
|
updated = template
|
|
elif isinstance(config, dict) and _is_standard_local_artemis(config):
|
|
if template is None:
|
|
template = {
|
|
"command": python,
|
|
"args": ["-m", "mcp_server"],
|
|
"env": {"PYTHONPATH": str(root)},
|
|
}
|
|
updated = _updated_entry(config, template, root, names)
|
|
elif config is not None and not isinstance(config, dict):
|
|
raise ValueError("The Artemis MCP configuration is not an object")
|
|
else:
|
|
return None
|
|
if updated == config:
|
|
return False
|
|
parent["artemis"] = updated
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
_atomic_write(path, json.dumps(data, ensure_ascii=False, indent=2) + "\n")
|
|
return True
|
|
|
|
|
|
def _toml_value(value: object) -> str:
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
return json.dumps(value, ensure_ascii=False)
|
|
|
|
|
|
def _toml_inline_table(values: dict) -> str:
|
|
return "{ " + ", ".join(f"{key} = {_toml_value(value)}" for key, value in values.items()) + " }"
|
|
|
|
|
|
def _replace_toml_tables(raw: str, assignments: dict[str, dict]) -> str:
|
|
"""Update only Artemis TOML tables; preserve every unrelated section and setting."""
|
|
lines = raw.splitlines(keepends=True)
|
|
output: list[str] = []
|
|
section = ""
|
|
found: dict[str, set[str]] = {name: set() for name in assignments}
|
|
seen: set[str] = set()
|
|
|
|
def append_missing() -> None:
|
|
if section in assignments:
|
|
for key, value in assignments[section].items():
|
|
if key not in found[section]:
|
|
rendered = _toml_inline_table(value) if key == "env" else _toml_value(value)
|
|
output.append(f"{key} = {rendered}\n")
|
|
found[section].add(key)
|
|
|
|
index = 0
|
|
while index < len(lines):
|
|
line = lines[index]
|
|
if line.lstrip().startswith("["):
|
|
append_missing()
|
|
header = re.sub(r"[\s\"']", "", line.split("#", 1)[0]).strip()
|
|
section = header[1:-1] if header.startswith("[") and header.endswith("]") else ""
|
|
seen.add(section)
|
|
match = re.match(r"\s*(\w+)\s*=", line) if section in assignments else None
|
|
if match and match[1] in assignments[section]:
|
|
key = match[1]
|
|
original = line
|
|
while True:
|
|
try:
|
|
tomllib.loads(original)
|
|
break
|
|
except tomllib.TOMLDecodeError:
|
|
index += 1
|
|
if index >= len(lines):
|
|
raise ValueError("Cannot safely parse existing MCP setting")
|
|
original += lines[index]
|
|
value = assignments[section][key]
|
|
rendered = _toml_inline_table(value) if key == "env" else _toml_value(value)
|
|
output.append(f"{key} = {rendered}\n")
|
|
found[section].add(key)
|
|
else:
|
|
output.append(line)
|
|
index += 1
|
|
append_missing()
|
|
for name, values in assignments.items():
|
|
if name not in seen:
|
|
if output and not output[-1].endswith("\n"):
|
|
output.append("\n")
|
|
output.append(f"\n[{name}]\n")
|
|
for key, value in values.items():
|
|
rendered = _toml_inline_table(value) if key == "env" else _toml_value(value)
|
|
output.append(f"{key} = {rendered}\n")
|
|
return "".join(output)
|
|
|
|
|
|
def refresh_codex(
|
|
path: Path,
|
|
root: Path,
|
|
python: str,
|
|
names: tuple[str, ...],
|
|
template: dict | None = None,
|
|
install_missing: bool = False,
|
|
) -> bool | None:
|
|
if not path.is_file() and not install_missing:
|
|
return False
|
|
raw = path.read_text(encoding="utf-8") if path.is_file() else ""
|
|
parsed = tomllib.loads(raw)
|
|
config = parsed.get("mcp_servers", {}).get("artemis")
|
|
if config is None and install_missing:
|
|
if template is None:
|
|
raise ValueError("A standard MCP template is required for installation")
|
|
updated = template
|
|
elif isinstance(config, dict) and _is_standard_local_artemis(config):
|
|
if template is None:
|
|
template = {
|
|
"command": python,
|
|
"args": ["-m", "mcp_server"],
|
|
"cwd": str(root),
|
|
"env": {"PYTHONPATH": str(root)},
|
|
"enabled_tools": list(names),
|
|
}
|
|
updated = _updated_entry(config, template, root, names)
|
|
else:
|
|
return None
|
|
top = {
|
|
key: value
|
|
for key, value in updated.items()
|
|
if key != "env" and (not config or config.get(key) != value)
|
|
}
|
|
environment = updated["env"]
|
|
if config and config.get("env") == environment:
|
|
env_updates = {}
|
|
else:
|
|
env_updates = {
|
|
key: value
|
|
for key, value in environment.items()
|
|
if not config or config.get("env", {}).get(key) != value
|
|
}
|
|
env_header = "mcp_servers.artemis.env"
|
|
has_env_table = re.search(r"(?m)^\s*\[mcp_servers\.artemis\.env\]\s*(?:#.*)?$", raw) is not None
|
|
if config and "env" in config and not has_env_table and env_updates:
|
|
top["env"] = environment
|
|
env_updates = {}
|
|
assignments = {"mcp_servers.artemis": top}
|
|
if env_updates:
|
|
assignments[env_header] = env_updates
|
|
if not top and not env_updates:
|
|
return False
|
|
result = _replace_toml_tables(raw, assignments)
|
|
actual = tomllib.loads(result)["mcp_servers"]["artemis"]
|
|
if any(actual.get(key) != value for key, value in top.items()) or any(
|
|
actual.get("env", {}).get(key) != value for key, value in env_updates.items()
|
|
):
|
|
raise ValueError("Could not safely refresh the Artemis MCP table")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
_atomic_write(path, result)
|
|
return result != raw
|
|
|
|
|
|
def refresh_mcp(root: Path) -> int:
|
|
sys.path.insert(0, str(root))
|
|
from artemis.interfaces.cli.commands.mcp import (
|
|
ARTEMIS_MCP_TOOLS,
|
|
_get_config_snippet,
|
|
_get_vscode_user_dir,
|
|
)
|
|
|
|
home = Path.home()
|
|
vscode = _get_vscode_user_dir()
|
|
appdata = Path(os.environ.get("APPDATA", home / "AppData/Roaming"))
|
|
cline_store = vscode / "globalStorage/saoudrizwan.claude-dev"
|
|
roo_store = vscode / "globalStorage/rooveterinaryinc.roo-cline"
|
|
paths = [
|
|
(home / ".gemini/config/mcp_config.json", ("mcpServers",), "antigravity", home / ".gemini"),
|
|
(home / ".gemini/jetski/mcp_config.json", ("mcpServers",), "jetski", home / ".gemini"),
|
|
(
|
|
home / ".gemini/antigravity/mcp_config.json",
|
|
("mcpServers",),
|
|
"antigravity",
|
|
home / ".gemini",
|
|
),
|
|
(home / ".cursor/mcp.json", ("mcpServers",), "cursor", home / ".cursor"),
|
|
(
|
|
home / ".codeium/windsurf/mcp_config.json",
|
|
("mcpServers",),
|
|
"windsurf",
|
|
home / ".codeium/windsurf",
|
|
),
|
|
(home / ".claude.json", ("mcpServers",), "claude", home / ".claude"),
|
|
(
|
|
appdata / "Claude/claude_desktop_config.json",
|
|
("mcpServers",),
|
|
"claude",
|
|
appdata / "Claude",
|
|
),
|
|
(vscode / "mcp.json", ("servers",), "vscode", vscode),
|
|
(home / ".copilot/mcp-config.json", ("servers",), "vscode", home / ".copilot"),
|
|
(
|
|
home / ".cline/data/settings/cline_mcp_settings.json",
|
|
("mcpServers",),
|
|
"cline",
|
|
home / ".cline",
|
|
),
|
|
(home / ".cline/mcp.json", ("mcpServers",), "cline", home / ".cline"),
|
|
(
|
|
cline_store / "settings/cline_mcp_settings.json",
|
|
("mcpServers",),
|
|
"cline",
|
|
cline_store,
|
|
),
|
|
(
|
|
roo_store / "settings/mcp_settings.json",
|
|
("mcpServers",),
|
|
"roo",
|
|
roo_store,
|
|
),
|
|
(
|
|
roo_store / "settings/cline_mcp_settings.json",
|
|
("mcpServers",),
|
|
"roo",
|
|
roo_store,
|
|
),
|
|
(root / ".roo/mcp.json", ("mcpServers",), "roo", root / ".roo"),
|
|
(home / ".openclaw/openclaw.json", ("mcp", "servers"), "openclaw", home / ".openclaw"),
|
|
]
|
|
codex_home = Path(os.environ.get("CODEX_HOME", home / ".codex"))
|
|
paths.append((codex_home / "config.toml", (), "codex", codex_home))
|
|
installed = updated_count = ready = skipped = errors = 0
|
|
for path, keys, client, marker in paths:
|
|
if not path.is_file() and not marker.is_dir():
|
|
continue
|
|
existed = path.is_file()
|
|
template = _get_config_snippet(client, sys.executable, str(root))
|
|
for key in keys:
|
|
template = template[key]
|
|
template = template["artemis"] if keys else template["mcp_servers"]["artemis"]
|
|
try:
|
|
changed = (
|
|
refresh_json(
|
|
path,
|
|
keys,
|
|
root,
|
|
sys.executable,
|
|
ARTEMIS_MCP_TOOLS,
|
|
template=template,
|
|
install_missing=True,
|
|
)
|
|
if keys
|
|
else refresh_codex(
|
|
path,
|
|
root,
|
|
sys.executable,
|
|
ARTEMIS_MCP_TOOLS,
|
|
template=template,
|
|
install_missing=True,
|
|
)
|
|
)
|
|
if changed:
|
|
print(
|
|
launch_text(
|
|
f"[{'UPDATED' if existed else 'INSTALLED'}] MCP configuration: {path}",
|
|
f"[{'已更新' if existed else '已安装'}] MCP 配置：{path}",
|
|
)
|
|
)
|
|
if existed:
|
|
updated_count += 1
|
|
else:
|
|
installed += 1
|
|
elif changed is None:
|
|
print(
|
|
launch_text(
|
|
f"[SKIP] Custom Artemis MCP entry in {path}; use the manual installer for this client.",
|
|
f"[跳过] {path} 使用自定义 ARTEMIS MCP 启动方式；请手动检查此客户端。",
|
|
)
|
|
)
|
|
skipped += 1
|
|
else:
|
|
ready += 1
|
|
except (OSError, ValueError, TypeError, AttributeError):
|
|
print(
|
|
launch_text(
|
|
f"[ERROR] Could not safely synchronize {path}; original configuration preserved.",
|
|
f"[错误] 无法安全同步 {path}；原配置已保留。",
|
|
)
|
|
)
|
|
errors += 1
|
|
print(
|
|
launch_text(
|
|
f"[CHECK] MCP clients: installed {installed}, updated {updated_count}, already current {ready}, custom skipped {skipped}, errors {errors}.",
|
|
f"[检查] MCP 客户端：新装 {installed} 个、更新 {updated_count} 个、已是最新 {ready} 个、自定义跳过 {skipped} 个、失败 {errors} 个。",
|
|
)
|
|
)
|
|
print(
|
|
launch_text(
|
|
"[ACTION] Reload or restart your AI client to load the current MCP process and tool schemas.",
|
|
"[操作] 请在 AI 客户端重新加载 MCP 或重启客户端，使新版进程和工具签名生效。",
|
|
)
|
|
)
|
|
if not (installed or updated_count or ready or skipped or errors):
|
|
print(
|
|
launch_text(
|
|
"[TIP] No supported client detected; install one with: uv run artemis mcp --install <client>",
|
|
"[提示] 未检测到受支持的客户端；可执行 uv run artemis mcp --install <client>。",
|
|
)
|
|
)
|
|
return 1 if errors else 0
|
|
|
|
|
|
def frontend_fingerprint(root: Path) -> str:
|
|
ui = root / "apps/showcase_ui"
|
|
files = list((ui / "src").rglob("*")) + list((ui / "public").rglob("*"))
|
|
files += [
|
|
ui / name
|
|
for name in (
|
|
"package.json",
|
|
"package-lock.json",
|
|
"angular.json",
|
|
"tsconfig.json",
|
|
"tsconfig.app.json",
|
|
)
|
|
]
|
|
digest = hashlib.sha256()
|
|
for path in sorted(p for p in files if p.is_file()):
|
|
digest.update(path.relative_to(ui).as_posix().encode())
|
|
digest.update(path.read_bytes())
|
|
return digest.hexdigest()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("action", choices=["source", "mcp", "frontend-check", "frontend-mark"])
|
|
parser.add_argument("--root", type=Path, default=ROOT)
|
|
parser.add_argument("--port", type=int, default=8000)
|
|
args = parser.parse_args()
|
|
if args.action == "source":
|
|
sys.exit(update_source(args.root, args.port))
|
|
elif args.action == "mcp":
|
|
sys.exit(refresh_mcp(args.root))
|
|
else:
|
|
marker = args.root / "apps/showcase_ui/node_modules/.artemis-source-hash"
|
|
current = frontend_fingerprint(args.root)
|
|
if args.action == "frontend-mark":
|
|
marker.parent.mkdir(parents=True, exist_ok=True)
|
|
marker.write_text(current, encoding="ascii")
|
|
elif not marker.is_file() or marker.read_text(encoding="ascii").strip() != current:
|
|
sys.exit(10)
|