Codex
937d9a002f
Add OBSIDIAN_MCP_ENABLED_TOOLS as the service config source for MCP tool groups while keeping OBSIDIAN_MCP_ENABLE_INGEST as a legacy fallback. Add mcp_service.sh tools list/enable/disable commands, preserve enabled tools across normal start and token rotation, and let start --enable-ingest/--disable-ingest update the same config file. Teach the MCP runtime to derive file-assets and ingest registration from enabled tool groups so file asset tools can be disabled separately from ingest. Validation: python3 -m unittest discover -v; python3 -m py_compile scripts/manage_mcp_service.py obsidian_wiki/mcp_server.py tests/test_mcp_service.py tests/test_mcp_server_remote.py; CLI tools enable/disable dry run; git diff --check.
635 lines
22 KiB
Python
Executable File
635 lines
22 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Manage the user-level HTTP MCP service for obsidian-wiki source files."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import secrets
|
|
import shlex
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Sequence
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from obsidian_wiki.config import resolve_vault_path
|
|
|
|
|
|
SERVICE_NAME = "obsidian-wiki-mcp.service"
|
|
ENV_RELATIVE_PATH = Path(".runtime") / "obsidian-wiki-mcp.env"
|
|
MCP_CLIENT_NAME = "wiki-file-assets"
|
|
DEFAULT_HOST = "0.0.0.0"
|
|
DEFAULT_PORT = 8765
|
|
DEFAULT_PATH = "/mcp"
|
|
DEFAULT_TOKEN_BYTES = 32
|
|
DEFAULT_UV_CACHE_DIR = "/tmp/uv-cache"
|
|
TOOL_FILE_ASSETS = "file-assets"
|
|
TOOL_INGEST = "ingest"
|
|
AVAILABLE_TOOL_GROUPS = (TOOL_FILE_ASSETS, TOOL_INGEST)
|
|
DEFAULT_ENABLED_TOOLS = (TOOL_FILE_ASSETS,)
|
|
|
|
|
|
def normalize_tool_name(name: str) -> str:
|
|
normalized = name.strip().lower().replace("_", "-")
|
|
aliases = {
|
|
"file": TOOL_FILE_ASSETS,
|
|
"files": TOOL_FILE_ASSETS,
|
|
"file-assets": TOOL_FILE_ASSETS,
|
|
"source-file": TOOL_FILE_ASSETS,
|
|
"source-files": TOOL_FILE_ASSETS,
|
|
"ingest": TOOL_INGEST,
|
|
"upload": TOOL_INGEST,
|
|
"uploads": TOOL_INGEST,
|
|
"opencode-ingest": TOOL_INGEST,
|
|
}
|
|
if normalized not in aliases:
|
|
available = ", ".join(AVAILABLE_TOOL_GROUPS)
|
|
raise ValueError(f"unknown MCP tool group: {name}. Available: {available}")
|
|
return aliases[normalized]
|
|
|
|
|
|
def parse_bool(value: str | None) -> bool:
|
|
return str(value or "").strip().lower() in {"1", "true", "yes", "on", "enabled"}
|
|
|
|
|
|
def normalize_enabled_tools(tools: Sequence[str]) -> tuple[str, ...]:
|
|
enabled = {normalize_tool_name(tool) for tool in tools if str(tool).strip()}
|
|
return tuple(tool for tool in AVAILABLE_TOOL_GROUPS if tool in enabled)
|
|
|
|
|
|
def enabled_tools_from_config(values: dict[str, str]) -> tuple[str, ...]:
|
|
raw_tools = values.get("OBSIDIAN_MCP_ENABLED_TOOLS", "").strip()
|
|
if raw_tools:
|
|
if raw_tools.lower() in {"none", "off", "disabled"}:
|
|
return ()
|
|
return normalize_enabled_tools(raw_tools.split(","))
|
|
|
|
enabled = set(DEFAULT_ENABLED_TOOLS)
|
|
if parse_bool(values.get("OBSIDIAN_MCP_ENABLE_INGEST")):
|
|
enabled.add(TOOL_INGEST)
|
|
return tuple(tool for tool in AVAILABLE_TOOL_GROUPS if tool in enabled)
|
|
|
|
|
|
def format_enabled_tools(enabled_tools: Sequence[str]) -> str:
|
|
return ",".join(normalize_enabled_tools(enabled_tools))
|
|
|
|
|
|
def update_ingest_tool(enabled_tools: Sequence[str], enable_ingest: bool | None) -> tuple[str, ...]:
|
|
current = set(normalize_enabled_tools(enabled_tools))
|
|
if enable_ingest is True:
|
|
current.add(TOOL_INGEST)
|
|
elif enable_ingest is False:
|
|
current.discard(TOOL_INGEST)
|
|
return tuple(tool for tool in AVAILABLE_TOOL_GROUPS if tool in current)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ServicePaths:
|
|
project_root: Path
|
|
unit_path: Path
|
|
env_path: Path
|
|
|
|
|
|
def generate_token(num_bytes: int = DEFAULT_TOKEN_BYTES) -> str:
|
|
return secrets.token_urlsafe(num_bytes)
|
|
|
|
|
|
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 resolve_vault_for_service(project_root: Path, vault_path: Path | str | None = None) -> Path:
|
|
if vault_path is not None:
|
|
return Path(vault_path).expanduser().resolve()
|
|
configured = resolve_vault_path(None, start=project_root)
|
|
if configured:
|
|
return Path(configured).expanduser().resolve()
|
|
return project_root.expanduser().resolve()
|
|
|
|
|
|
def render_service_unit(
|
|
*,
|
|
project_root: Path,
|
|
vault_path: 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",
|
|
"python",
|
|
"-m",
|
|
"obsidian_wiki.mcp_server",
|
|
"--vault",
|
|
systemd_quote(vault_path),
|
|
"--transport",
|
|
"streamable-http",
|
|
"--host",
|
|
host,
|
|
"--port",
|
|
str(port),
|
|
"--path",
|
|
path,
|
|
]
|
|
)
|
|
return (
|
|
"[Unit]\n"
|
|
"Description=obsidian-wiki readonly 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 prepare_runtime(
|
|
project_root: Path,
|
|
*,
|
|
uv_path: Path | None = None,
|
|
uv_cache_dir: str = DEFAULT_UV_CACHE_DIR,
|
|
) -> None:
|
|
resolved_uv_path = uv_path or find_uv_path()
|
|
env = {**os.environ, "UV_CACHE_DIR": uv_cache_dir}
|
|
subprocess.run(
|
|
[
|
|
str(resolved_uv_path),
|
|
"run",
|
|
"--extra",
|
|
"mcp",
|
|
"python",
|
|
"-c",
|
|
"import mcp.server.fastmcp",
|
|
],
|
|
cwd=project_root,
|
|
env=env,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
|
|
|
|
def write_token_env_file(
|
|
env_path: Path,
|
|
token: str,
|
|
*,
|
|
host: str = DEFAULT_HOST,
|
|
port: int = DEFAULT_PORT,
|
|
path: str = DEFAULT_PATH,
|
|
enabled_tools: Sequence[str] | None = None,
|
|
enable_ingest: bool | None = None,
|
|
) -> None:
|
|
tools = normalize_enabled_tools(DEFAULT_ENABLED_TOOLS if enabled_tools is None else enabled_tools)
|
|
tools = update_ingest_tool(tools, enable_ingest)
|
|
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"OBSIDIAN_MCP_TOKEN={token}\n")
|
|
env_file.write(f"OBSIDIAN_MCP_HOST={host}\n")
|
|
env_file.write(f"OBSIDIAN_MCP_PORT={port}\n")
|
|
env_file.write(f"OBSIDIAN_MCP_PATH={path}\n")
|
|
env_file.write(f"OBSIDIAN_MCP_ENABLED_TOOLS={format_enabled_tools(tools)}\n")
|
|
env_file.write(f"OBSIDIAN_MCP_ENABLE_INGEST={1 if TOOL_INGEST in tools else 0}\n")
|
|
env_path.chmod(0o600)
|
|
|
|
|
|
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.")
|
|
|
|
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) -> tuple[str, str, int, str, tuple[str, ...]]:
|
|
paths = resolve_service_paths(project_root)
|
|
values = read_service_env_file(paths.env_path)
|
|
token = values.get("OBSIDIAN_MCP_TOKEN", "").strip()
|
|
if not token:
|
|
raise RuntimeError(f"Missing OBSIDIAN_MCP_TOKEN in {paths.env_path}")
|
|
host = values.get("OBSIDIAN_MCP_HOST") or DEFAULT_HOST
|
|
port = int(values.get("OBSIDIAN_MCP_PORT") or DEFAULT_PORT)
|
|
path = values.get("OBSIDIAN_MCP_PATH") or DEFAULT_PATH
|
|
enabled_tools = enabled_tools_from_config(values)
|
|
return token, host, port, path, enabled_tools
|
|
|
|
|
|
def write_service_unit(
|
|
paths: ServicePaths,
|
|
*,
|
|
vault_path: Path,
|
|
host: str = DEFAULT_HOST,
|
|
port: int = DEFAULT_PORT,
|
|
path: str = DEFAULT_PATH,
|
|
uv_path: Path | None = None,
|
|
) -> None:
|
|
paths.unit_path.parent.mkdir(parents=True, exist_ok=True)
|
|
unit = render_service_unit(
|
|
project_root=paths.project_root,
|
|
vault_path=vault_path,
|
|
env_file=paths.env_path,
|
|
host=host,
|
|
port=port,
|
|
path=path,
|
|
uv_path=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,
|
|
config_path: Path | None = None,
|
|
enabled_tools: Sequence[str] | None = None,
|
|
) -> str:
|
|
host = hosts[0] if hosts else "<server-ip>"
|
|
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 ""
|
|
config_section = f"\nConfig file:\n{config_path}\n" if config_path is not None else ""
|
|
enabled_section = ""
|
|
if enabled_tools is not None:
|
|
enabled_section = f"\nEnabled tool groups:\n{format_enabled_tools(enabled_tools) or 'none'}\n"
|
|
return (
|
|
"MCP service is running.\n"
|
|
"\n"
|
|
"Use this URL from another machine:\n"
|
|
f"{url}\n"
|
|
f"{alternatives_section}"
|
|
f"{config_section}"
|
|
f"{enabled_section}"
|
|
"\n"
|
|
"Use this header:\n"
|
|
f"Authorization: Bearer {token}\n"
|
|
"\n"
|
|
"Example MCP client config:\n"
|
|
"{\n"
|
|
' "mcpServers": {\n'
|
|
f' "{MCP_CLIENT_NAME}": {{\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:
|
|
paths = resolve_service_paths(project_root)
|
|
token, _host, port, path, enabled_tools = saved_service_config(project_root)
|
|
return render_remote_config(
|
|
hosts=detect_remote_hosts(),
|
|
token=token,
|
|
port=port,
|
|
path=path,
|
|
config_path=paths.env_path,
|
|
enabled_tools=enabled_tools,
|
|
)
|
|
|
|
|
|
def run_systemctl(args: Sequence[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(["systemctl", "--user", *args], text=True, check=check)
|
|
|
|
|
|
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,
|
|
token_bytes: int = DEFAULT_TOKEN_BYTES,
|
|
) -> str:
|
|
paths = resolve_service_paths(project_root)
|
|
enabled_tools = DEFAULT_ENABLED_TOOLS
|
|
try:
|
|
_old_token, _old_host, _old_port, _old_path, enabled_tools = saved_service_config(project_root)
|
|
except RuntimeError:
|
|
pass
|
|
token = generate_token(token_bytes)
|
|
write_token_env_file(paths.env_path, token, host=host, port=port, path=path, enabled_tools=enabled_tools)
|
|
restart_service_if_installed(paths)
|
|
return render_remote_config(
|
|
hosts=detect_remote_hosts(),
|
|
token=token,
|
|
port=port,
|
|
path=path,
|
|
config_path=paths.env_path,
|
|
enabled_tools=enabled_tools,
|
|
)
|
|
|
|
|
|
def start_service(
|
|
*,
|
|
project_root: Path,
|
|
vault_path: Path | str | None = None,
|
|
host: str = DEFAULT_HOST,
|
|
port: int = DEFAULT_PORT,
|
|
path: str = DEFAULT_PATH,
|
|
token_bytes: int = DEFAULT_TOKEN_BYTES,
|
|
enable_ingest: bool | None = None,
|
|
) -> str:
|
|
paths = resolve_service_paths(project_root)
|
|
try:
|
|
token, host, port, path, enabled_tools = saved_service_config(project_root)
|
|
except RuntimeError:
|
|
if paths.env_path.exists():
|
|
raise
|
|
token = generate_token(token_bytes)
|
|
enabled_tools = DEFAULT_ENABLED_TOOLS
|
|
enabled_tools = update_ingest_tool(enabled_tools, enable_ingest)
|
|
write_token_env_file(paths.env_path, token, host=host, port=port, path=path, enabled_tools=enabled_tools)
|
|
uv_path = find_uv_path()
|
|
prepare_runtime(paths.project_root, uv_path=uv_path)
|
|
resolved_vault = resolve_vault_for_service(paths.project_root, vault_path=vault_path)
|
|
write_service_unit(paths, vault_path=resolved_vault, host=host, port=port, path=path, uv_path=uv_path)
|
|
|
|
run_systemctl(["daemon-reload"])
|
|
run_systemctl(["enable", SERVICE_NAME])
|
|
run_systemctl(["restart", SERVICE_NAME])
|
|
return render_remote_config(
|
|
hosts=detect_remote_hosts(),
|
|
token=token,
|
|
port=port,
|
|
path=path,
|
|
config_path=paths.env_path,
|
|
enabled_tools=enabled_tools,
|
|
)
|
|
|
|
|
|
def render_tool_status(
|
|
*,
|
|
config_path: Path,
|
|
enabled_tools: Sequence[str],
|
|
restarted: bool = False,
|
|
) -> str:
|
|
enabled = set(normalize_enabled_tools(enabled_tools))
|
|
lines = [
|
|
"MCP tool configuration.",
|
|
"",
|
|
f"Config file: {config_path}",
|
|
f"Enabled tool groups: {format_enabled_tools(enabled_tools) or 'none'}",
|
|
"",
|
|
"Available tool groups:",
|
|
]
|
|
for tool in AVAILABLE_TOOL_GROUPS:
|
|
state = "enabled" if tool in enabled else "disabled"
|
|
lines.append(f"- {tool}: {state}")
|
|
lines.append("")
|
|
lines.append(f"Service restarted: {'yes' if restarted else 'no'}")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def ensure_service_config(project_root: Path) -> tuple[ServicePaths, str, str, int, str, tuple[str, ...]]:
|
|
paths = resolve_service_paths(project_root)
|
|
try:
|
|
token, host, port, path, enabled_tools = saved_service_config(project_root)
|
|
except RuntimeError:
|
|
token = generate_token()
|
|
host = DEFAULT_HOST
|
|
port = DEFAULT_PORT
|
|
path = DEFAULT_PATH
|
|
enabled_tools = DEFAULT_ENABLED_TOOLS
|
|
write_token_env_file(paths.env_path, token, host=host, port=port, path=path, enabled_tools=enabled_tools)
|
|
return paths, token, host, port, path, enabled_tools
|
|
|
|
|
|
def list_tools(project_root: Path) -> str:
|
|
paths, _token, _host, _port, _path, enabled_tools = ensure_service_config(project_root)
|
|
return render_tool_status(config_path=paths.env_path, enabled_tools=enabled_tools)
|
|
|
|
|
|
def set_tool_enabled(project_root: Path, tool: str, enabled: bool) -> str:
|
|
tool_name = normalize_tool_name(tool)
|
|
paths, token, host, port, path, enabled_tools = ensure_service_config(project_root)
|
|
current = set(enabled_tools)
|
|
if enabled:
|
|
current.add(tool_name)
|
|
else:
|
|
current.discard(tool_name)
|
|
updated_tools = tuple(item for item in AVAILABLE_TOOL_GROUPS if item in current)
|
|
write_token_env_file(paths.env_path, token, host=host, port=port, path=path, enabled_tools=updated_tools)
|
|
restarted = restart_service_if_installed(paths)
|
|
return render_tool_status(config_path=paths.env_path, enabled_tools=updated_tools, restarted=restarted)
|
|
|
|
|
|
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() -> None:
|
|
run_systemctl(["status", SERVICE_NAME], check=False)
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Manage the readonly HTTP obsidian-wiki MCP user service.")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
new_token_parser = subparsers.add_parser("new-token", help="Generate a token and print remote 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("--token-bytes", default=DEFAULT_TOKEN_BYTES, type=int)
|
|
|
|
start_parser = subparsers.add_parser("start", help="Install or update the user service and print remote MCP config.")
|
|
start_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
|
|
start_parser.add_argument("--vault", dest="vault_path", type=Path, help="Vault path to serve. Defaults to OBSIDIAN_VAULT_PATH config.")
|
|
start_parser.add_argument("--host", default=DEFAULT_HOST)
|
|
start_parser.add_argument("--port", default=DEFAULT_PORT, type=int)
|
|
start_parser.add_argument("--path", default=DEFAULT_PATH)
|
|
start_parser.add_argument("--token-bytes", default=DEFAULT_TOKEN_BYTES, type=int)
|
|
ingest_group = start_parser.add_mutually_exclusive_group()
|
|
ingest_group.add_argument(
|
|
"--enable-ingest",
|
|
dest="enable_ingest",
|
|
action="store_true",
|
|
default=None,
|
|
help="Expose write-capable upload and local OpenCode ingest MCP tools.",
|
|
)
|
|
ingest_group.add_argument(
|
|
"--disable-ingest",
|
|
dest="enable_ingest",
|
|
action="store_false",
|
|
help="Disable write-capable upload and local OpenCode ingest MCP tools.",
|
|
)
|
|
|
|
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 remote MCP client configuration.")
|
|
show_config_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
|
|
|
|
subparsers.add_parser("status", help="Show systemd status for the managed service.")
|
|
|
|
tools_parser = subparsers.add_parser("tools", help="List or update enabled MCP tool groups in the service config.")
|
|
tools_subparsers = tools_parser.add_subparsers(dest="tools_command", required=True)
|
|
|
|
tools_list_parser = tools_subparsers.add_parser("list", help="List configured MCP tool groups.")
|
|
tools_list_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
|
|
|
|
tools_enable_parser = tools_subparsers.add_parser("enable", help="Enable an MCP tool group.")
|
|
tools_enable_parser.add_argument("tool", choices=AVAILABLE_TOOL_GROUPS)
|
|
tools_enable_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
|
|
|
|
tools_disable_parser = tools_subparsers.add_parser("disable", help="Disable an MCP tool group.")
|
|
tools_disable_parser.add_argument("tool", choices=AVAILABLE_TOOL_GROUPS)
|
|
tools_disable_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,
|
|
token_bytes=args.token_bytes,
|
|
),
|
|
end="",
|
|
)
|
|
return 0
|
|
if args.command == "start":
|
|
print(
|
|
start_service(
|
|
project_root=args.project_root,
|
|
vault_path=args.vault_path,
|
|
host=args.host,
|
|
port=args.port,
|
|
path=args.path,
|
|
token_bytes=args.token_bytes,
|
|
enable_ingest=args.enable_ingest,
|
|
),
|
|
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":
|
|
service_status()
|
|
return 0
|
|
if args.command == "tools":
|
|
if args.tools_command == "list":
|
|
print(list_tools(args.project_root), end="")
|
|
return 0
|
|
if args.tools_command == "enable":
|
|
print(set_tool_enabled(args.project_root, args.tool, True), end="")
|
|
return 0
|
|
if args.tools_command == "disable":
|
|
print(set_tool_enabled(args.project_root, args.tool, False), end="")
|
|
return 0
|
|
raise AssertionError(f"unhandled command: {args.command}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|