diff --git a/hermes_cli/agent_import.py b/hermes_cli/agent_import.py index 3e6d61907a..5b01e0fff3 100644 --- a/hermes_cli/agent_import.py +++ b/hermes_cli/agent_import.py @@ -1,8 +1,8 @@ """hermes import-agent — import Claude Code / Codex CLI setups into Hermes. -Secrets are NEVER imported: credential files (.credentials.json, auth.json) are ignored, and MCP -server env vars with secret-looking names (KEY, TOKEN, SECRET, PASSWORD, ...) are stripped and -reported so the user can re-add them deliberately via ``hermes setup`` or config.yaml. +Secrets are NEVER imported: credential files are never read, and MCP env vars with secret-looking +names (KEY, TOKEN, SECRET, PASSWORD, ...) are stripped and reported so the user re-adds them via +``hermes setup`` or config.yaml. """ from __future__ import annotations @@ -13,42 +13,28 @@ import re import shutil import sys import time +import tomllib from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Tuple +import yaml + from utils import atomic_write_text, atomic_yaml_write logger = logging.getLogger(__name__) -# Same entry delimiter as the Hermes memory store and the openclaw migration -# script — memories/MEMORY.md entries are separated by bare "§" lines. +# Entry delimiter of the Hermes memory store (memories/MEMORY.md) and the openclaw script. ENTRY_DELIMITER = "\n§\n" - -# Character budget for merged memory files (matches the openclaw script's -# default memory limit). +# Character budget for merged memory files (openclaw script default). MEMORY_CHAR_LIMIT = 20_000 - SUPPORTED_AGENTS = ("claude-code", "codex") - -_AGENT_DEFAULT_DIRS = { - "claude-code": ".claude", - "codex": ".codex", -} - -_SKILL_CATEGORY = { - "claude-code": "claude-code-imports", - "codex": "codex-imports", -} +_AGENT_DEFAULT_DIRS = {"claude-code": ".claude", "codex": ".codex"} +_SKILL_CATEGORY = {"claude-code": "claude-code-imports", "codex": "codex-imports"} # Env var names that look like credentials — never copied into config.yaml. _SECRET_KEY_RE = re.compile( r"(?:^|_)(?:API[_-]?KEY|APIKEY|TOKEN|SECRET|PASSWORD|PASSWD|CREDENTIALS?|" - r"AUTH|PRIVATE[_-]?KEY|ACCESS[_-]?KEY)(?:_|$)|KEY$", - re.IGNORECASE, -) - -# Files inside the source tree that hold credentials — never read. -_CREDENTIAL_FILENAMES = (".credentials.json", "auth.json", "credentials.json") + r"AUTH|PRIVATE[_-]?KEY|ACCESS[_-]?KEY)(?:_|$)|KEY$", re.IGNORECASE) def is_secret_key(key: str) -> bool: @@ -65,92 +51,59 @@ def read_text(path: Path) -> str: class ConfigReadError(RuntimeError): - """An existing config file is present but cannot be read or parsed. - - Signals that a read-modify-write round trip must be abandoned: the caller has no idea what the - file holds, so writing a merged result back would replace real settings with only the keys it - merged. - """ + """Existing config file present but unreadable/unparseable: the read-modify-write round trip + must be abandoned, else the merged result would replace real settings with only merged keys.""" def load_yaml_file(path: Path) -> Dict[str, Any]: - """Load a YAML mapping, distinguishing "absent" from "unreadable". - - - Absent, or present but empty -> ``{}``; first-time creation still works. - Present but - unreadable, unparseable, or not a mapping -> raise :class:`ConfigReadError` so the caller - refuses and leaves the file byte-identical. - """ - import yaml - + """Load a YAML mapping: absent/empty -> ``{}``; unreadable/unparseable/non-mapping -> + :class:`ConfigReadError` so the caller refuses and leaves the file byte-identical.""" if not path.exists(): return {} fix_hint = "Fix it with `hermes config edit` (or move it aside), then re-run the import." + + def refusal(detail: str) -> ConfigReadError: + return ConfigReadError(f"Refusing to overwrite {path}: {detail}") + try: raw = read_text(path) except OSError as exc: - raise ConfigReadError( - f"Refusing to overwrite {path}: the existing file cannot be read " - f"({exc}). Fix the file permissions or move it aside first." - ) from exc + raise refusal(f"the existing file cannot be read ({exc}). " + "Fix the file permissions or move it aside first.") from exc try: data = yaml.safe_load(raw) except yaml.YAMLError as exc: - raise ConfigReadError( - f"Refusing to overwrite {path}: the existing file is not valid YAML ({exc}). {fix_hint}" - ) from exc - # An empty file parses to None — a legitimate state with nothing to lose. - if data is None: + raise refusal(f"the existing file is not valid YAML ({exc}). {fix_hint}") from exc + if data is None: # empty file — a legitimate state with nothing to lose return {} if not isinstance(data, dict): - raise ConfigReadError( - f"Refusing to overwrite {path}: expected the existing file to hold a " - f"YAML mapping but found {type(data).__name__}. {fix_hint}" - ) + raise refusal(f"expected the existing file to hold a YAML mapping but found " + f"{type(data).__name__}. {fix_hint}") return data def dump_yaml_file(path: Path, data: Dict[str, Any]) -> None: - """Write ``data`` as YAML atomically (temp file + fsync + rename). - - Only ever reached after :func:`load_yaml_file` has successfully read the same path, so the - mapping being written is the real file's content plus the merged section — never a silently- - empty stand-in. - """ + """Atomic YAML write; only reached after :func:`load_yaml_file` succeeded on the same path.""" atomic_yaml_write(path, data) -# --------------------------------------------------------------------------- -# Memory-entry primitives (ported from openclaw_to_hermes.py) -# --------------------------------------------------------------------------- - def extract_markdown_entries(text: str) -> List[str]: - """Split a markdown document into individual memory entries. - - Headings become context prefixes; bullets and paragraphs become entries; code blocks and - tables are skipped. - """ + """Split markdown into memory entries: headings become context prefixes; bullets and + paragraphs become entries; code blocks and tables are skipped.""" entries: List[str] = [] headings: List[str] = [] paragraph_lines: List[str] = [] - def context_prefix() -> str: - filtered = [ - h for h in headings - if h and not re.search( - r"\b(MEMORY|USER|SOUL|AGENTS|TOOLS|IDENTITY|CLAUDE)\.md\b", - h, re.I, - ) - ] - return " > ".join(filtered) - def add_entry(content: str) -> None: - prefix = context_prefix() + prefix = " > ".join( + h for h in headings + if h and not re.search(r"\b(MEMORY|USER|SOUL|AGENTS|TOOLS|IDENTITY|CLAUDE)\.md\b", + h, re.I)) entries.append(f"{prefix}: {content}" if prefix else content) def flush_paragraph() -> None: - nonlocal paragraph_lines block = " ".join(line.strip() for line in paragraph_lines).strip() - paragraph_lines = [] + paragraph_lines.clear() if block: add_entry(block) @@ -158,98 +111,54 @@ def extract_markdown_entries(text: str) -> List[str]: for raw_line in (text or "").splitlines(): line = raw_line.rstrip() stripped = line.strip() - if stripped.startswith("```"): in_code_block = not in_code_block flush_paragraph() continue if in_code_block: continue - heading_match = re.match(r"^(#{1,6})\s+(.*\S)\s*$", stripped) if heading_match: flush_paragraph() - level = len(heading_match.group(1)) - value = heading_match.group(2).strip() - while len(headings) >= level: - headings.pop() - headings.append(value) + # Drop deeper (and same-level) headings, then push this one. + headings[len(heading_match.group(1)) - 1:] = [heading_match.group(2).strip()] continue - bullet_match = re.match(r"^\s*(?:[-*]|\d+\.)\s+(.*\S)\s*$", line) if bullet_match: flush_paragraph() add_entry(bullet_match.group(1).strip()) continue - - if not stripped: - flush_paragraph() + if not stripped or (stripped.startswith("|") and stripped.endswith("|")): + flush_paragraph() # blank line or table row ends the paragraph continue - - if stripped.startswith("|") and stripped.endswith("|"): - flush_paragraph() - continue - paragraph_lines.append(stripped) flush_paragraph() - deduped: List[str] = [] seen = set() for entry in entries: normalized = normalize_text(entry) - if not normalized or normalized in seen: - continue - seen.add(normalized) - deduped.append(entry.strip()) + if normalized and normalized not in seen: + seen.add(normalized) + deduped.append(entry.strip()) return deduped def parse_existing_memory_entries(path: Path) -> List[str]: - """Parse the DESTINATION memory store into entries. - - ``memories/MEMORY.md`` is the entry-delimited store written by ``MemoryStore._write_file`` - (tools/memory_tool.py), not a markdown document, so this splits on ``ENTRY_DELIMITER`` only — - exactly what ``MemoryStore._parse_entries`` does. - - Do NOT fall back to :func:`extract_markdown_entries` here. That extractor is correct for - CLAUDE.md / AGENTS.md *sources*, but it drops fenced code blocks and table rows and splits a - block into one entry per bullet — and the merged result is written straight back over the user's - store, so the loss is permanent. - """ + """Parse the DESTINATION memory store (``ENTRY_DELIMITER``-split, as ``MemoryStore`` does). + Do NOT fall back to :func:`extract_markdown_entries`: it drops code blocks/table rows and + splits bullets, and the merged result overwrites the user's store — the loss is permanent.""" if not path.exists(): return [] - raw = read_text(path) - if not raw.strip(): - return [] - return [e.strip() for e in raw.split(ENTRY_DELIMITER) if e.strip()] + return [e.strip() for e in read_text(path).split(ENTRY_DELIMITER) if e.strip()] -def backup_memory_file(path: Path) -> Optional[Path]: - """Snapshot ``path`` before a destructive rewrite; return the backup path. - - Restores parity with the openclaw migration script this module was ported from, which calls - ``maybe_backup(destination)`` before rewriting a memory store. Uses the same - ``.bak.`` naming as ``MemoryStore._backup_drifted_file``. Returns None when there - is nothing to back up. - """ - if not path.exists(): - return None - backup = path.with_suffix(path.suffix + f".bak.{int(time.time())}") - shutil.copy2(path, backup) - return backup - - -def merge_entries( - existing: Sequence[str], - incoming: Sequence[str], - limit: int, -) -> Tuple[List[str], Dict[str, int]]: +def merge_entries(existing: Sequence[str], incoming: Sequence[str], + limit: int) -> Tuple[List[str], Dict[str, int]]: merged = list(existing) seen = {normalize_text(e) for e in existing if e.strip()} stats = {"existing": len(existing), "added": 0, "duplicates": 0, "overflowed": 0} - - current_len = len(ENTRY_DELIMITER.join(merged)) if merged else 0 + current_len = len(ENTRY_DELIMITER.join(merged)) for entry in incoming: normalized = normalize_text(entry) if not normalized: @@ -257,10 +166,8 @@ def merge_entries( if normalized in seen: stats["duplicates"] += 1 continue - candidate_len = ( - len(entry) if not merged - else current_len + len(ENTRY_DELIMITER) + len(entry) - ) + candidate_len = (len(entry) if not merged + else current_len + len(ENTRY_DELIMITER) + len(entry)) if candidate_len > limit: stats["overflowed"] += 1 continue @@ -271,43 +178,22 @@ def merge_entries( return merged, stats -# --------------------------------------------------------------------------- -# Claude Code permission rules → Hermes command patterns -# --------------------------------------------------------------------------- - _BASH_RULE_RE = re.compile(r"^Bash\((?P.*)\)$") def claude_rule_to_command_pattern(rule: str) -> Optional[str]: - """Convert a Claude Code ``Bash(...)`` permission rule into a Hermes glob. - - ``Bash(npm run test:*)`` -> ``npm run test*`` (Claude ':*' is a prefix match). Bare ``Bash`` - and non-Bash rules (``Read(...)``, ``WebFetch(...)``) return None: the former is too broad to - import, the latter gate Claude-specific tools with no command-allowlist equivalent. - """ - rule = (rule or "").strip() - m = _BASH_RULE_RE.match(rule) - if not m: - return None - inner = m.group("inner").strip() + """``Bash(npm run test:*)`` -> ``npm run test*`` (Claude ':*' is a prefix match). Bare ``Bash`` + (too broad) and non-Bash rules (Claude-only tools, no allowlist equivalent) return None.""" + m = _BASH_RULE_RE.match((rule or "").strip()) + inner = m.group("inner").strip() if m else "" if not inner: return None - if inner.endswith(":*"): - inner = inner[:-2] + "*" - return inner - - -# --------------------------------------------------------------------------- -# Detection -# --------------------------------------------------------------------------- - -def default_source_dir(agent: str) -> Path: - return Path.home() / _AGENT_DEFAULT_DIRS[agent] + return inner[:-2] + "*" if inner.endswith(":*") else inner def detect_agents() -> List[str]: """Return the list of supported agents whose default dirs exist.""" - return [a for a in SUPPORTED_AGENTS if default_source_dir(a).is_dir()] + return [a for a in SUPPORTED_AGENTS if (Path.home() / _AGENT_DEFAULT_DIRS[a]).is_dir()] def sanitize_mcp_env(env: Any) -> Tuple[Dict[str, str], List[str]]: @@ -318,13 +204,6 @@ def sanitize_mcp_env(env: Any) -> Tuple[Dict[str, str], List[str]]: return kept, [str(k) for k in env if str(k) not in kept] -def _copy_skill_dir(skill_dir: Path, destination: Path) -> None: - destination.parent.mkdir(parents=True, exist_ok=True) - if destination.exists(): - shutil.rmtree(destination) - shutil.copytree(skill_dir, destination) - - def _translate_mcp_server(name: str, srv: Dict[str, Any]) -> Tuple[Dict[str, Any], List[str]]: """Map one Claude/Codex MCP server entry to Hermes shape; returns (server, stripped secret paths).""" hermes_srv: Dict[str, Any] = {} @@ -343,35 +222,20 @@ def _translate_mcp_server(name: str, srv: Dict[str, Any]) -> Tuple[Dict[str, Any hermes_srv["url"] = srv["url"] headers = srv.get("headers") if isinstance(headers, dict): - kept_headers = { - k: v for k, v in headers.items() - if not is_secret_key(str(k)) and "authorization" not in str(k).lower() - } + kept_headers = {k: v for k, v in headers.items() + if not is_secret_key(str(k)) and "authorization" not in str(k).lower()} if kept_headers: hermes_srv["headers"] = kept_headers stripped.extend(f"mcp_servers.{name}.headers.{k}" for k in headers if k not in kept_headers) return hermes_srv, stripped -# --------------------------------------------------------------------------- -# Importer -# --------------------------------------------------------------------------- - class AgentImporter: - """Detect/parse/map/apply importer for a single agent source tree. + """Detect/parse/map/apply importer for one agent source tree. ``execute=False`` runs the full + plan without touching disk; every item is recorded as imported/skipped/conflict/error.""" - ``execute=False`` runs the full plan without touching disk (dry run). Every item is recorded - as imported/skipped/conflict/error with a reason so the CLI can print a per-item report. - """ - - def __init__( - self, - agent: str, - source_root: Path, - target_root: Path, - execute: bool = False, - overwrite: bool = False, - ) -> None: + def __init__(self, agent: str, source_root: Path, target_root: Path, + execute: bool = False, overwrite: bool = False) -> None: if agent not in SUPPORTED_AGENTS: raise ValueError(f"Unsupported agent: {agent!r}") self.agent = agent @@ -382,27 +246,16 @@ class AgentImporter: self.items: List[Dict[str, Any]] = [] self.stripped_secrets: List[str] = [] - # -- reporting --------------------------------------------------------- - def record(self, kind: str, source, destination, status: str, reason: str = "", **details) -> None: - item: Dict[str, Any] = { - "kind": kind, - "source": str(source) if source else None, - "destination": str(destination) if destination else None, - "status": status, - "reason": reason, - } - item.update(details) - self.items.append(item) + self.items.append({"kind": kind, "source": str(source) if source else None, + "destination": str(destination) if destination else None, + "status": status, "reason": reason, **details}) - def load_target_config(self, kind: str, source, destination: Path - ) -> Optional[Dict[str, Any]]: - """Read the destination config.yaml, or record a refusal and return None. - - Deliberately runs in dry-run too: ``--dry-run`` must report the refusal, not preview an - ``imported`` that would destroy the config. - """ + def load_target_config(self, kind: str, source, + destination: Path) -> Optional[Dict[str, Any]]: + """Read the destination config.yaml, or record a refusal and return None. Runs in dry-run + too: ``--dry-run`` must report the refusal, not preview an ``imported`` that destroys it.""" try: return load_yaml_file(destination) except ConfigReadError as exc: @@ -412,88 +265,74 @@ class AgentImporter: def apply(self, kind: str, source, destination, would: str, action, details: Optional[Dict[str, Any]] = None) -> None: """Record ``imported`` (reason ``would`` in dry-run); in execute mode run ``action`` first. - ``action`` may add keys to ``details`` (shared by reference) and returns an error string - to record ``error`` instead, or None on success. - """ + (recorded as ``error``) or None.""" details = details or {} - if not self.execute: - self.record(kind, source, destination, "imported", would, **details) - return - error = action() - self.record(kind, source, destination, "error" if error else "imported", error or "", **details) + error = action() if self.execute else None + self.record(kind, source, destination, "error" if error else "imported", + error or ("" if self.execute else would), **details) def build_report(self) -> Dict[str, Any]: summary = {"imported": 0, "skipped": 0, "conflict": 0, "error": 0} for item in self.items: summary[item["status"]] = summary.get(item["status"], 0) + 1 - report: Dict[str, Any] = { - "agent": self.agent, - "source": str(self.source_root), - "target": str(self.target_root), - "dry_run": not self.execute, - "items": self.items, - "summary": summary, - } + report: Dict[str, Any] = {"agent": self.agent, "source": str(self.source_root), + "target": str(self.target_root), "dry_run": not self.execute, + "items": self.items, "summary": summary} if self.stripped_secrets: report["stripped_secrets"] = sorted(set(self.stripped_secrets)) return report - # -- orchestration ----------------------------------------------------- - def run(self) -> Dict[str, Any]: if not self.source_root.is_dir(): self.record("source", self.source_root, None, "error", "Source directory does not exist") - return self.build_report() - {"claude-code": self._run_claude_code, "codex": self._run_codex}[self.agent]() + else: + {"claude-code": self._run_claude_code, "codex": self._run_codex}[self.agent]() return self.build_report() def _run_claude_code(self) -> None: - settings = self._load_claude_settings() - self.import_context_file(self.source_root / "CLAUDE.md", kind="claude-md") - self.import_permission_allowlist(settings) - self.import_permission_denylist(settings) - self.import_mcp_servers(self._claude_mcp_servers(settings), kind="mcp-servers") + settings = self._load_source_mapping( + "settings", self.source_root / "settings.json", json.loads, self._JSON_ERRORS, + record_missing=True, non_mapping_error="settings.json is not a JSON object") + self.import_context_file(self.source_root / "CLAUDE.md", "claude-md") + self._import_permission_rules(settings, "allow") + self._import_permission_rules(settings, "deny") + # mcpServers: ~/.claude.json (preferred; lives NEXT TO ~/.claude/) then settings.json + claude_json = self._load_source_mapping( + "mcp-servers", self.source_root.parent / ".claude.json", json.loads, self._JSON_ERRORS) + servers: Dict[str, Any] = {} + for source in (claude_json.get("mcpServers"), settings.get("mcpServers")): + if isinstance(source, dict): + for name, srv in source.items(): + servers.setdefault(name, srv) + self.import_mcp_servers(servers, kind="mcp-servers") self.import_skills(self.source_root / "skills") commands_dir = self.source_root / "commands" if commands_dir.is_dir() and any(commands_dir.glob("*.md")): - self.record( - "slash-commands", commands_dir, None, "skipped", - "Claude slash commands have no direct Hermes equivalent — " - "consider converting them into skills", - ) + self.record("slash-commands", commands_dir, None, "skipped", + "Claude slash commands have no direct Hermes equivalent — " + "consider converting them into skills") def _run_codex(self) -> None: - config = self._load_codex_config() - self.import_context_file(self.source_root / "AGENTS.md", kind="agents-md") + config = self._load_source_mapping("config", self.source_root / "config.toml", + tomllib.loads, Exception, record_missing=True) + self.import_context_file(self.source_root / "AGENTS.md", "agents-md") mcp = config.get("mcp_servers") self.import_mcp_servers(mcp if isinstance(mcp, dict) else {}, kind="mcp-servers") - self.import_memories_dir(self.source_root / "memories") + memories_dir = self.source_root / "memories" + self._import_markdown_files( + "memories", memories_dir, + sorted(memories_dir.glob("*.md")) if memories_dir.is_dir() else None, + "No memories directory found") self.import_skills(self.source_root / "skills") - # -- parsers (fail soft: bad files become per-item error records) ------- - _JSON_ERRORS = (json.JSONDecodeError, OSError) - def _load_claude_settings(self) -> Dict[str, Any]: - return self._load_source_mapping( - "settings", self.source_root / "settings.json", json.loads, self._JSON_ERRORS, - record_missing=True, non_mapping_error="settings.json is not a JSON object", - ) - - def _load_codex_config(self) -> Dict[str, Any]: - import tomllib - - return self._load_source_mapping( - "config", self.source_root / "config.toml", tomllib.loads, Exception, - record_missing=True, - ) - def _load_source_mapping(self, kind: str, path: Path, parse, errors, *, record_missing: bool = False, non_mapping_error: str = "") -> Dict[str, Any]: - """Parse ``path`` into a mapping; problems become per-item error records, result ``{}``.""" + """Parse ``path`` into a mapping; problems become per-item error records and ``{}``.""" if not path.exists(): if record_missing: self.record(kind, None, None, "skipped", f"No {path.name} found") @@ -509,46 +348,20 @@ class AgentImporter: self.record(kind, path, None, "error", non_mapping_error) return {} - def _claude_mcp_servers(self, settings: Dict[str, Any]) -> Dict[str, Any]: - """Collect mcpServers from ~/.claude.json (preferred) and settings.json.""" - # ~/.claude.json lives NEXT TO ~/.claude/, not inside it - data = self._load_source_mapping( - "mcp-servers", self.source_root.parent / ".claude.json", json.loads, self._JSON_ERRORS) - servers: Dict[str, Any] = {} - if isinstance(data.get("mcpServers"), dict): - servers.update(data["mcpServers"]) - from_settings = settings.get("mcpServers") - if isinstance(from_settings, dict): - for name, srv in from_settings.items(): - servers.setdefault(name, srv) - return servers - - # -- mappers ------------------------------------------------------------- - def import_context_file(self, source: Path, kind: str) -> None: """CLAUDE.md / AGENTS.md → memory entries in memories/MEMORY.md.""" - if not source.exists(): - self.record(kind, None, self.target_root / "memories" / "MEMORY.md", - "skipped", f"No {source.name} found") - return - self._import_markdown_files(kind, source, [source], single_file=True) - - def import_memories_dir(self, memories_dir: Path) -> None: - """codex memories/*.md → memory entries in memories/MEMORY.md.""" - if not memories_dir.is_dir(): - self.record("memories", None, self.target_root / "memories" / "MEMORY.md", - "skipped", "No memories directory found") - return - self._import_markdown_files("memories", memories_dir, sorted(memories_dir.glob("*.md"))) - - def _import_markdown_files(self, kind: str, source: Path, files: List[Path], - single_file: bool = False) -> None: - """Extract entries from ``files`` and merge them into memories/MEMORY.md. + self._import_markdown_files(kind, source, [source] if source.exists() else None, + f"No {source.name} found", single_file=True) + def _import_markdown_files(self, kind: str, source: Path, files: Optional[List[Path]], + missing_reason: str, single_file: bool = False) -> None: + """Extract entries from ``files`` (None = source missing) and merge into memories/MEMORY.md. An unreadable file records an error; a directory import then still reports "no entries" - when nothing was extracted, while a single-file import stops at the error. - """ + when nothing was extracted, while a single-file import stops at the error.""" destination = self.target_root / "memories" / "MEMORY.md" + if files is None: + self.record(kind, None, destination, "skipped", missing_reason) + return incoming: List[str] = [] failed = False for md_file in files: @@ -561,35 +374,29 @@ class AgentImporter: if not (failed and single_file): self.record(kind, source, destination, "skipped", "No importable entries found") return - self._merge_memory_entries(kind, source, destination, incoming) - - def _merge_memory_entries(self, kind: str, source: Path, - destination: Path, incoming: List[str]) -> None: existing = parse_existing_memory_entries(destination) merged, stats = merge_entries(existing, incoming, MEMORY_CHAR_LIMIT) - details = { - "existing_entries": stats["existing"], - "added_entries": stats["added"], - "duplicate_entries": stats["duplicates"], - "overflowed_entries": stats["overflowed"], - } + details = {"existing_entries": stats["existing"], "added_entries": stats["added"], + "duplicate_entries": stats["duplicates"], + "overflowed_entries": stats["overflowed"]} if stats["added"] == 0: self.record(kind, source, destination, "skipped", "No new entries to import", **details) return def write() -> Optional[str]: destination.parent.mkdir(parents=True, exist_ok=True) + # Snapshot first (``.bak.``, as MemoryStore does); never rewrite the + # store when the safety net failed. + step = "back up existing" try: - backup = backup_memory_file(destination) - except OSError as exc: - # Never rewrite the store when the safety net failed. - return f"Could not back up existing memory file: {exc}" - if backup is not None: - details["backup"] = str(backup) - try: + if destination.exists(): + backup = destination.with_suffix(f"{destination.suffix}.bak.{int(time.time())}") + shutil.copy2(destination, backup) + details["backup"] = str(backup) + step = "write merged" atomic_write_text(destination, ENTRY_DELIMITER.join(merged) + ("\n" if merged else "")) except OSError as exc: - return f"Could not write merged memory file: {exc}" + return f"Could not {step} memory file: {exc}" return None self.apply(kind, source, destination, "Would merge entries", write, details) @@ -597,18 +404,10 @@ class AgentImporter: # (settings key, item kind, config path, dry-run tracks unmapped rules) _PERMISSION_RULES = { "allow": ("command-allowlist", ("command_allowlist",), True), - "deny": ("command-denylist", ("approvals", "deny"), False), - } - - def import_permission_allowlist(self, settings: Dict[str, Any]) -> None: - """settings.json permissions.allow → config.yaml command_allowlist.""" - self._import_permission_rules(settings, "allow") - - def import_permission_denylist(self, settings: Dict[str, Any]) -> None: - """settings.json permissions.deny → config.yaml approvals.deny.""" - self._import_permission_rules(settings, "deny") + "deny": ("command-denylist", ("approvals", "deny"), False)} def _import_permission_rules(self, settings: Dict[str, Any], key: str) -> None: + """settings.json permissions.allow/deny → config.yaml command_allowlist / approvals.deny.""" kind, config_path, track_unmapped = self._PERMISSION_RULES[key] label = f"settings.json permissions.{key}" destination = self.target_root / "config.yaml" @@ -617,18 +416,9 @@ class AgentImporter: if not isinstance(rules, list) or not rules: self.record(kind, None, destination, "skipped", f"No permissions.{key} rules found") return - - patterns: List[str] = [] - skipped_rules: List[str] = [] - for rule in rules: - if not isinstance(rule, str): - continue - pattern = claude_rule_to_command_pattern(rule) - if pattern: - patterns.append(pattern) - else: - skipped_rules.append(rule) - patterns = sorted(dict.fromkeys(patterns)) + mapped = [(r, claude_rule_to_command_pattern(r)) for r in rules if isinstance(r, str)] + patterns = sorted(dict.fromkeys(p for _, p in mapped if p)) + skipped_rules = [r for r, p in mapped if not p] unmapped: Dict[str, Any] = {"unmapped_rules": skipped_rules} if track_unmapped else {} if not patterns: self.record(kind, None, destination, "skipped", @@ -636,7 +426,6 @@ class AgentImporter: return if not skipped_rules: unmapped = {} - config = self.load_target_config(kind, label, destination) if config is None: return @@ -644,18 +433,15 @@ class AgentImporter: parent: Dict[str, Any] = config for part in config_path[:-1]: child = parent.get(part) - if not isinstance(child, dict): - child = {} - parent[part] = child - parent = child + parent[part] = parent = child if isinstance(child, dict) else {} current = parent.get(config_path[-1], []) - if not isinstance(current, list): - current = [] + current = current if isinstance(current, list) else [] merged = sorted(dict.fromkeys(list(current) + patterns)) added = [p for p in merged if p not in current] if not added: self.record(kind, label, destination, "skipped", "All patterns already present") return + def write() -> None: parent[config_path[-1]] = merged dump_yaml_file(destination, config) @@ -669,13 +455,11 @@ class AgentImporter: if not servers: self.record(kind, None, destination, "skipped", "No MCP servers found") return - config = self.load_target_config(kind, None, destination) if config is None: return existing = config.get("mcp_servers") - if not isinstance(existing, dict): - existing = {} + existing = existing if isinstance(existing, dict) else {} added = 0 for name, srv in servers.items(): if not isinstance(srv, dict): @@ -685,32 +469,26 @@ class AgentImporter: self.record(kind, name, f"mcp_servers.{name}", "conflict", "MCP server already exists in Hermes config") continue - hermes_srv, stripped = _translate_mcp_server(name, srv) self.stripped_secrets.extend(stripped) if not hermes_srv: self.record(kind, name, None, "skipped", "Server has neither a command nor a url") continue - existing[name] = hermes_srv added += 1 self.record(kind, name, f"config.yaml mcp_servers.{name}", "imported") - if added > 0 and self.execute: config["mcp_servers"] = existing dump_yaml_file(destination, config) def import_skills(self, source_root: Path) -> None: """skills//SKILL.md dirs → HERMES_HOME/skills//.""" - category = _SKILL_CATEGORY[self.agent] - destination_root = self.target_root / "skills" / category + destination_root = self.target_root / "skills" / _SKILL_CATEGORY[self.agent] if not source_root.is_dir(): self.record("skills", None, destination_root, "skipped", "No skills directory found") return - skill_dirs = [ - p for p in sorted(source_root.iterdir()) - if p.is_dir() and (p / "SKILL.md").exists() - ] + skill_dirs = [p for p in sorted(source_root.iterdir()) + if p.is_dir() and (p / "SKILL.md").exists()] if not skill_dirs: self.record("skills", source_root, destination_root, "skipped", "No skills with SKILL.md found") @@ -721,26 +499,25 @@ class AgentImporter: self.record("skill", skill_dir, destination, "conflict", "Destination skill already exists") continue - self.apply("skill", skill_dir, destination, "Would copy skill directory", - lambda: _copy_skill_dir(skill_dir, destination)) + def copy(skill_dir=skill_dir, destination=destination) -> None: + destination.parent.mkdir(parents=True, exist_ok=True) + if destination.exists(): + shutil.rmtree(destination) + shutil.copytree(skill_dir, destination) + + self.apply("skill", skill_dir, destination, "Would copy skill directory", copy) -# --------------------------------------------------------------------------- -# CLI entry point -# --------------------------------------------------------------------------- def import_agent_command(args) -> None: """Handle ``hermes import-agent`` (invoked from hermes_cli.main).""" from hermes_cli.config import get_config_path, load_config, save_config from hermes_constants import get_hermes_home - from hermes_cli.setup import ( - Colors, color, print_header, print_info, print_success, print_error, prompt_yes_no, - ) + from hermes_cli.setup import (Colors, color, print_header, print_info, print_success, + print_error, prompt_yes_no) - agent, explicit_source = args.agent, args.source - dry_run, overwrite, auto_yes = args.dry_run, args.overwrite, args.yes + agent, explicit_source, overwrite = args.agent, args.source, args.overwrite - # -- detect -------------------------------------------------------------- if agent is None: detected = detect_agents() if not detected: @@ -754,20 +531,17 @@ def import_agent_command(args) -> None: print_info("Pick one: hermes import-agent claude-code or hermes import-agent codex") return agent = detected[0] - - source_dir = Path(explicit_source) if explicit_source else default_source_dir(agent) + source_dir = Path(explicit_source or Path.home() / _AGENT_DEFAULT_DIRS[agent]) print() print(color("┌─────────────────────────────────────────────────────────┐", Colors.MAGENTA)) print(color("│ ⚕ Hermes — Import From Another Agent │", Colors.MAGENTA)) print(color("└─────────────────────────────────────────────────────────┘", Colors.MAGENTA)) - if not source_dir.is_dir(): print() print_error(f"Agent directory not found: {source_dir}") print_info(f"Specify a custom path: hermes import-agent {agent} --source /path/to/{_AGENT_DEFAULT_DIRS[agent]}") return - hermes_home = get_hermes_home() print() print_header("Import Settings") @@ -776,48 +550,41 @@ def import_agent_command(args) -> None: print_info(f"Target: {hermes_home}") print_info(f"Overwrite: {'yes' if overwrite else 'no (skip conflicts)'}") print_info("Secrets: never imported — run 'hermes setup' for credentials") - # Ensure config.yaml exists before the import tries to merge into it - config_path = get_config_path() - if not config_path.exists(): + if not get_config_path().exists(): save_config(load_config()) - # -- Phase 1: preview (always) -------------------------------------------- - def run_import(execute: bool) -> Dict[str, Any]: - return AgentImporter( - agent=agent, - source_root=source_dir.resolve(), - target_root=hermes_home.resolve(), - execute=execute, - overwrite=overwrite, - ).run() + def run_import(execute: bool, phase: str) -> Optional[Dict[str, Any]]: + """Run the importer; on failure print the error and return None.""" + try: + return AgentImporter(agent, source_dir.resolve(), hermes_home.resolve(), + execute=execute, overwrite=overwrite).run() + except Exception as e: + print() + print_error(f"Import{phase} failed: {e}") + logger.debug(f"import-agent{phase} error", exc_info=True) + return None - try: - preview = run_import(execute=False) - except Exception as e: - print() - print_error(f"Import preview failed: {e}") - logger.debug("import-agent preview error", exc_info=True) + # Phase 1: preview (always) + preview = run_import(False, " preview") + if preview is None: return - summary = preview.get("summary", {}) if summary.get("imported", 0) == 0 and summary.get("conflict", 0) == 0: print() print_info(f"Nothing to import from {agent}.") print_import_report(preview, dry_run=True) return - print() print_header(f"Import Preview — {summary.get('imported', 0)} item(s) would be imported") print_info("No changes have been made yet. Review the list below:") print_import_report(preview, dry_run=True) - - if dry_run: + if args.dry_run: return - # -- Phase 2: confirm and execute ----------------------------------------- + # Phase 2: confirm and execute print() - if not auto_yes: + if not args.yes: if not sys.stdin.isatty(): print_info("Non-interactive session — preview only.") print_info(f"To execute, re-run with: hermes import-agent {agent} --yes") @@ -825,15 +592,9 @@ def import_agent_command(args) -> None: if not prompt_yes_no("Proceed with import?", default=True): print_info("Import cancelled.") return - - try: - report = run_import(execute=True) - except Exception as e: - print() - print_error(f"Import failed: {e}") - logger.debug("import-agent error", exc_info=True) + report = run_import(True, "") + if report is None: return - print_import_report(report, dry_run=False) print() print_success("Import complete.") @@ -845,47 +606,36 @@ def print_import_report(report: Dict[str, Any], dry_run: bool) -> None: """Print a formatted per-item import report (claw-migrate style).""" from hermes_cli.setup import Colors, color, print_header, print_info - summary = report.get("summary", {}) print() + print_header("Dry Run Results" if dry_run else "Import Results") if dry_run: - print_header("Dry Run Results") print_info("No files were modified. This is a preview of what would happen.") - else: - print_header("Import Results") print() - - items = report.get("items", []) + # (status, colour, group heading, summary label) groups = ( - ("imported", Colors.GREEN, "✓ Would import" if dry_run else "✓ Imported"), - ("conflict", Colors.YELLOW, "⚠ Conflicts (skipped — use --overwrite to force)"), - ("skipped", Colors.DIM, "─ Skipped"), - ("error", Colors.RED, "✗ Errors"), - ) - for status, col, label in groups: - group_items = [i for i in items if i.get("status") == status] + ("imported", Colors.GREEN, "✓ Would import" if dry_run else "✓ Imported", + "would import" if dry_run else "imported"), + ("conflict", Colors.YELLOW, "⚠ Conflicts (skipped — use --overwrite to force)", + "conflict(s)"), + ("skipped", Colors.DIM, "─ Skipped", "skipped"), + ("error", Colors.RED, "✗ Errors", "error(s)")) + for status, col, label, _ in groups: + group_items = [i for i in report.get("items", []) if i.get("status") == status] if not group_items: continue print(color(f" {label}:", col)) for item in group_items: - tail = ( - "→ " + str(item.get("destination") or "").replace(str(Path.home()), "~") - if status == "imported" else f" {item.get('reason', '')}" - ) + tail = ("→ " + str(item.get("destination") or "").replace(str(Path.home()), "~") + if status == "imported" else f" {item.get('reason', '')}") print(f" {item.get('kind', 'unknown'):<22s} {tail}") print() - - stripped = report.get("stripped_secrets") or [] - if stripped: + if stripped := report.get("stripped_secrets"): print(color(" ⚷ Secrets stripped (never imported):", Colors.YELLOW)) for name in stripped: print(f" {name}") print_info("Re-add credentials deliberately via 'hermes setup' or ~/.hermes/.env.") print() - - labels = ( - ("imported", "would import" if dry_run else "imported"), - ("conflict", "conflict(s)"), ("skipped", "skipped"), ("error", "error(s)"), - ) - parts = [f"{summary[k]} {label}" for k, label in labels if summary.get(k)] + summary = report.get("summary", {}) + parts = [f"{summary[k]} {label}" for k, _, _, label in groups if summary.get(k)] if parts: print_info(f"Summary: {', '.join(parts)}") diff --git a/hermes_cli/agent_plugins.py b/hermes_cli/agent_plugins.py index 6428f80315..01366b6011 100644 --- a/hermes_cli/agent_plugins.py +++ b/hermes_cli/agent_plugins.py @@ -2,23 +2,22 @@ from __future__ import annotations +import ipaddress import json import os import re from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, Mapping, Tuple +from urllib.parse import urlsplit from agent.skill_utils import yaml_load - PLUGIN_SCHEMA_V1 = "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json" MCP_SCHEMA_V1 = "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json" -_PLUGIN_FIELDS = { - "$schema", "name", "version", "description", "author", "homepage", "repository", "license", - "keywords", "extensions", -} +_PLUGIN_FIELDS = {"$schema", "name", "version", "description", "author", "homepage", "repository", + "license", "keywords", "extensions"} _AUTHOR_FIELDS = {"name", "email", "url"} _STDIO_FIELDS = {"type", "command", "args", "env", "cwd"} _REMOTE_FIELDS = {"type", "url", "headers"} @@ -62,8 +61,7 @@ class AgentPluginPackage: def _inside(path: Path, root: Path) -> bool: try: - path.resolve(strict=False).relative_to(root.resolve(strict=True)) - return True + return path.resolve(strict=False).is_relative_to(root.resolve(strict=True)) except (OSError, RuntimeError, ValueError): return False @@ -92,52 +90,38 @@ def _validate_manifest(root: Path) -> tuple[dict, list[AgentPluginDiagnostic]]: raise AgentPluginError("plugin.json must be a regular file within the plugin root") manifest = _read_json_object(manifest_path, label="plugin.json") diagnostics: list[AgentPluginDiagnostic] = [] - for field in sorted(set(manifest) - _PLUGIN_FIELDS): diagnostics.append(AgentPluginDiagnostic("manifest", f"ignored unknown top-level field: {field}")) manifest.pop(field) - if manifest.get("$schema") != PLUGIN_SCHEMA_V1: raise AgentPluginError("plugin.json declares an unsupported or missing Agent Plugins schema") name = manifest.get("name") if not isinstance(name, str) or not 1 <= len(name) <= 64 or _PLUGIN_NAME_RE.fullmatch(name) is None: raise AgentPluginError("plugin.json name does not satisfy v1 constraints") - for field in ("version", "description", "homepage", "repository", "license"): if field in manifest and not isinstance(manifest[field], str): raise AgentPluginError(f"plugin.json {field} must be a string") - - if "keywords" in manifest: - keywords = manifest["keywords"] - if not isinstance(keywords, list) or not _all_str(keywords): - raise AgentPluginError("plugin.json keywords must be an array of strings") - - if "author" in manifest: - author = manifest["author"] - if not isinstance(author, dict): - raise AgentPluginError("plugin.json author must be an object") - if set(author) - _AUTHOR_FIELDS or not _all_str(author.values()): - raise AgentPluginError("plugin.json author may contain only string name, email, and url fields") - - if "extensions" in manifest: - extensions = manifest["extensions"] - if not isinstance(extensions, dict): - diagnostics.append(AgentPluginDiagnostic("manifest", "ignored non-object extensions field")) - manifest.pop("extensions") - elif any(not isinstance(value, dict) for value in extensions.values()): - raise AgentPluginError("plugin.json extension namespace values must be objects") - + keywords = manifest.get("keywords", []) + if not isinstance(keywords, list) or not _all_str(keywords): + raise AgentPluginError("plugin.json keywords must be an array of strings") + author = manifest.get("author", {}) + if not isinstance(author, dict): + raise AgentPluginError("plugin.json author must be an object") + if set(author) - _AUTHOR_FIELDS or not _all_str(author.values()): + raise AgentPluginError("plugin.json author may contain only string name, email, and url fields") + extensions = manifest.get("extensions", {}) + if not isinstance(extensions, dict): + diagnostics.append(AgentPluginDiagnostic("manifest", "ignored non-object extensions field")) + manifest.pop("extensions") + elif any(not isinstance(value, dict) for value in extensions.values()): + raise AgentPluginError("plugin.json extension namespace values must be objects") return manifest, diagnostics def _valid_skill_frontmatter(frontmatter: Mapping[str, Any], directory_name: str) -> str | None: name = frontmatter.get("name") - if ( - not isinstance(name, str) - or name != directory_name - or not 1 <= len(name) <= 64 - or _SKILL_NAME_RE.fullmatch(name) is None - ): + if (not isinstance(name, str) or name != directory_name or not 1 <= len(name) <= 64 + or _SKILL_NAME_RE.fullmatch(name) is None): return "name must match the directory and satisfy Agent Skills constraints" description = frontmatter.get("description") if not isinstance(description, str) or not 1 <= len(description) <= 1024: @@ -184,15 +168,11 @@ def _discover_skills(root: Path, diagnostics: list[AgentPluginDiagnostic]) -> tu except OSError as exc: diagnostics.append(AgentPluginDiagnostic("skills", f"cannot list skills: {exc}")) return () - skills: list[AgentPluginSkill] = [] for child in children: skill_md = child / "SKILL.md" if not child.is_dir() or not skill_md.exists(): continue - scope = f"skill:{child.name}" - error: str | None = None - frontmatter: dict = {} if not _inside(skill_md, root) or not skill_md.is_file(): error = "SKILL.md must be a regular in-root file" else: @@ -203,17 +183,12 @@ def _discover_skills(root: Path, diagnostics: list[AgentPluginDiagnostic]) -> tu else: error = _valid_skill_frontmatter(frontmatter, child.name) if error: - diagnostics.append(AgentPluginDiagnostic(scope, error)) + diagnostics.append(AgentPluginDiagnostic(f"skill:{child.name}", error)) continue - skills.append( - AgentPluginSkill( - name=child.name, - description=frontmatter["description"], - root=child.resolve(strict=True), - skill_md=skill_md.resolve(strict=True), - frontmatter=dict(frontmatter), - ) - ) + skills.append(AgentPluginSkill( + name=child.name, description=frontmatter["description"], + root=child.resolve(strict=True), skill_md=skill_md.resolve(strict=True), + frontmatter=dict(frontmatter))) return tuple(skills) @@ -222,9 +197,8 @@ def _expand(value: str, plugin_root: Path, data_root: Path) -> str: return _PLACEHOLDER_RE.sub(lambda match: replacements[match.group(1)], value) -def _resolve_scoped_path( - value: str, plugin_root: Path, data_root: Path, *, expand_placeholders: bool = True -) -> Path: +def _resolve_scoped_path(value: str, plugin_root: Path, data_root: Path, *, + expand_placeholders: bool = True) -> Path: expanded = _expand(value, plugin_root, data_root) if expand_placeholders else value if value.startswith("./"): base, candidate = plugin_root, plugin_root / expanded[2:] @@ -243,28 +217,17 @@ def _resolve_scoped_path( def _validate_headers(headers: object) -> bool: - if headers is None: - return True if not isinstance(headers, dict): - return False + return headers is None names = [name for name in headers if isinstance(name, str) and _HEADER_NAME_RE.fullmatch(name)] - return ( - len(names) == len(headers) - and len({name.lower() for name in names}) == len(names) - and all(isinstance(v, str) and "\r" not in v and "\n" not in v for v in headers.values()) - ) + values = headers.values() + return (len(names) == len(headers) and len({name.lower() for name in names}) == len(names) + and all(isinstance(v, str) and "\r" not in v and "\n" not in v for v in values)) def _validate_remote_url(url: object) -> str: - """Validate a portable remote MCP URL per the v1 spec and return it. - - Rules (Agent Plugins v1 §7.2.1): absolute http(s) URL, no user information, no fragment; non- - loopback endpoints must use HTTPS. HTTP is allowed only when the host is exactly ``localhost`` - or an IP literal in a loopback range. No placeholder or environment expansion is performed. - """ - - from urllib.parse import urlsplit - + """Validate a portable remote MCP URL per Agent Plugins v1 §7.2.1 and return it: absolute + http(s), no user info, no fragment; HTTP only for ``localhost``/loopback IP. No expansion.""" if not isinstance(url, str) or not url: raise ValueError("url must be a non-empty string") try: @@ -282,8 +245,6 @@ def _validate_remote_url(url: object) -> str: if not host: raise ValueError("url must have a host") if scheme == "http" and host != "localhost": - import ipaddress - try: loopback = ipaddress.ip_address(host).is_loopback except ValueError: @@ -294,13 +255,8 @@ def _validate_remote_url(url: object) -> str: def _translate_remote(config: Mapping[str, Any]) -> Dict[str, Any]: - """Translate a portable ``streamable-http`` entry into native MCP config. - - ``strict_redirect_headers`` tells the runtime to drop configured headers on any cross-origin - redirect, as the v1 spec requires for portable packages (no forwarding to another origin - without explicit user authorization). - """ - + """Translate a portable ``streamable-http`` entry into native MCP config. The v1 spec requires + ``strict_redirect_headers``: drop configured headers on any cross-origin redirect.""" if set(config) - _REMOTE_FIELDS: raise ValueError("unknown remote field") url = _validate_remote_url(config.get("url")) @@ -312,122 +268,83 @@ def _translate_remote(config: Mapping[str, Any]) -> Dict[str, Any]: return translated -def _translate_stdio(config: Mapping[str, Any], plugin_root: Path, data_root: Path) -> Dict[str, Any]: +def _translate_stdio(config: Mapping[str, Any], plugin_root: Path, data_root: Path, + create_data: bool = False) -> Dict[str, Any]: if set(config) - _STDIO_FIELDS: raise ValueError("unknown stdio field") command = config.get("command") if not isinstance(command, str) or not command or "\x00" in command: raise ValueError("command must be a non-empty executable token") if command.startswith("./"): - command_value = str(_resolve_scoped_path(command, plugin_root, data_root, expand_placeholders=False)) + command = str(_resolve_scoped_path(command, plugin_root, data_root, + expand_placeholders=False)) elif any(character.isspace() for character in command): raise ValueError("command must contain one executable token") elif "/" in command or "\\" in command or command in {".", ".."}: raise ValueError("command must be a bare executable or begin with ./") - else: - command_value = command - args = config.get("args", []) if not isinstance(args, list) or not _all_str(args): raise ValueError("args must be an array of strings") env = config.get("env", {}) if not _str_map(env): raise ValueError("env must map string keys to string values") - env_keys = {key.upper() if os.name == "nt" else key for key in env} - if "PLUGIN_ROOT" in env_keys or "PLUGIN_DATA" in env_keys: + if {"PLUGIN_ROOT", "PLUGIN_DATA"} & {key.upper() if os.name == "nt" else key for key in env}: raise ValueError("PLUGIN_ROOT and PLUGIN_DATA are reserved") - cwd = config.get("cwd") - if cwd is None: - cwd_value = plugin_root - elif not isinstance(cwd, str): + if cwd is not None and not isinstance(cwd, str): raise ValueError("cwd must be a string") - else: - cwd_value = _resolve_scoped_path(cwd, plugin_root, data_root) - - translated_env = {key: _expand(value, plugin_root, data_root) for key, value in env.items()} - translated_env["PLUGIN_ROOT"] = str(plugin_root) - translated_env["PLUGIN_DATA"] = str(data_root) - return { - "command": command_value, - "args": [_expand(value, plugin_root, data_root) for value in args], - "env": translated_env, - "cwd": str(cwd_value), - } - - -def _translate_stdio_server(server: Mapping[str, Any], root: Path, data_root: Path, create_data: bool) -> Dict[str, Any]: - translated_server = _translate_stdio(server, root, data_root) + cwd_path = plugin_root if cwd is None else _resolve_scoped_path(cwd, plugin_root, data_root) if create_data: data_root.mkdir(parents=True, exist_ok=True) - cwd_path = Path(translated_server["cwd"]) - # The MCP client starts stdio servers with this cwd. Create only data-root - # descendants; plugin-root paths remain package-owned and are never made - # writable as a side effect of discovery. + # The MCP client starts stdio servers with this cwd. Create only data-root descendants; + # plugin-root paths remain package-owned and are never made writable by discovery. if cwd_path.is_relative_to(data_root): cwd_path.mkdir(parents=True, exist_ok=True) - return translated_server + translated_env = {**{key: _expand(value, plugin_root, data_root) for key, value in env.items()}, + "PLUGIN_ROOT": str(plugin_root), "PLUGIN_DATA": str(data_root)} + return {"command": command, "args": [_expand(value, plugin_root, data_root) for value in args], + "env": translated_env, "cwd": str(cwd_path)} def _reject_sse(server: Mapping[str, Any]) -> None: - if ( - set(server) - _REMOTE_FIELDS - or not isinstance(server.get("url"), str) - or not server.get("url") - or not _validate_headers(server.get("headers")) - ): + if (set(server) - _REMOTE_FIELDS or not isinstance(server.get("url"), str) + or not server.get("url") or not _validate_headers(server.get("headers"))): raise ValueError("invalid remote entry") raise ValueError("portable sse transport is not supported") -def _discover_mcp( - root: Path, - data_root: Path, - diagnostics: list[AgentPluginDiagnostic], - *, - create_data: bool = True, -) -> Dict[str, Dict[str, Any]]: +def _discover_mcp(root: Path, data_root: Path, diagnostics: list[AgentPluginDiagnostic], *, + create_data: bool = True) -> Dict[str, Dict[str, Any]]: mcp_path = root / "mcp.json" if not mcp_path.exists() and not mcp_path.is_symlink(): return {} - - def _fail(message: str) -> Dict[str, Dict[str, Any]]: - diagnostics.append(AgentPluginDiagnostic("mcp", message)) - return {} - - if not _inside(mcp_path, root) or not mcp_path.is_file(): - return _fail("mcp.json must be a regular in-root file") - try: + try: # file-level problems (incl. AgentPluginError from the JSON read) -> one "mcp" diagnostic + if not _inside(mcp_path, root) or not mcp_path.is_file(): + raise ValueError("mcp.json must be a regular in-root file") config = _read_json_object(mcp_path, label="mcp.json") - except AgentPluginError as exc: - return _fail(str(exc)) - if set(config) != {"$schema", "mcpServers"}: - return _fail("mcp.json has an invalid top-level shape") - if config.get("$schema") != MCP_SCHEMA_V1: - return _fail("mcp.json declares an unsupported schema") - servers = config.get("mcpServers") - if not isinstance(servers, dict): - return _fail("mcpServers must be an object") - - translators = { - "stdio": lambda server: _translate_stdio_server(server, root, data_root, create_data), - "streamable-http": _translate_remote, - "sse": _reject_sse, - } + if set(config) != {"$schema", "mcpServers"}: + raise ValueError("mcp.json has an invalid top-level shape") + if config.get("$schema") != MCP_SCHEMA_V1: + raise ValueError("mcp.json declares an unsupported schema") + servers = config.get("mcpServers") + if not isinstance(servers, dict): + raise ValueError("mcpServers must be an object") + except ValueError as exc: + diagnostics.append(AgentPluginDiagnostic("mcp", str(exc))) + return {} + translators = {"stdio": lambda server: _translate_stdio(server, root, data_root, create_data), + "streamable-http": _translate_remote, "sse": _reject_sse} translated: Dict[str, Dict[str, Any]] = {} for name, server in servers.items(): - scope = f"mcp:{name}" - if not isinstance(name, str) or not name or not isinstance(server, dict): - diagnostics.append(AgentPluginDiagnostic(scope, "invalid server entry")) - continue - translate = translators.get(server.get("type")) - if translate is None: - diagnostics.append(AgentPluginDiagnostic(scope, "unknown MCP server type")) - continue try: + if not isinstance(name, str) or not name or not isinstance(server, dict): + raise ValueError("invalid server entry") + translate = translators.get(server.get("type")) + if translate is None: + raise ValueError("unknown MCP server type") translated[name] = translate(server) except (OSError, ValueError) as exc: - diagnostics.append(AgentPluginDiagnostic(scope, str(exc))) + diagnostics.append(AgentPluginDiagnostic(f"mcp:{name}", str(exc))) return translated @@ -440,39 +357,23 @@ def _validate_root(plugin_root: Path) -> tuple[Path, dict, list[AgentPluginDiagn def load_agent_plugin(plugin_root: Path, data_root: Path) -> AgentPluginPackage: """Validate and translate one installed Agent Plugins v1 package.""" - root, manifest, diagnostics = _validate_root(plugin_root) resolved_data = Path(data_root).resolve(strict=False) - skills = _discover_skills(root, diagnostics) - mcp_servers = _discover_mcp(root, resolved_data, diagnostics) - return AgentPluginPackage( - name=manifest["name"], - version=manifest.get("version", ""), - description=manifest.get("description", ""), - root=root, - data_root=resolved_data, - manifest=dict(manifest), - skills=skills, - mcp_servers=mcp_servers, - diagnostics=tuple(diagnostics), - ) + return AgentPluginPackage( # skills are discovered before MCP: diagnostics keep that order + name=manifest["name"], version=manifest.get("version", ""), + description=manifest.get("description", ""), root=root, data_root=resolved_data, + manifest=dict(manifest), skills=_discover_skills(root, diagnostics), + mcp_servers=_discover_mcp(root, resolved_data, diagnostics), diagnostics=tuple(diagnostics)) def read_agent_plugin_manifest(plugin_root: Path) -> tuple[dict, tuple[AgentPluginDiagnostic, ...]]: """Validate only root ``plugin.json`` without discovering components.""" - _root, manifest, diagnostics = _validate_root(plugin_root) return manifest, tuple(diagnostics) def has_enabled_agent_plugin_mcp(raw_config: Mapping[str, Any]) -> bool: - """Compatibility wrapper for the shared PluginManager MCP probe. - - Directory scanning belongs to :mod:`hermes_cli.plugins` so startup gating and full plugin - discovery cannot drift apart. Keep this import-compatible entry point for callers that used the - original helper. - """ - + """Import-compatible wrapper for the shared PluginManager MCP probe; directory scanning lives + in :mod:`hermes_cli.plugins` so startup gating and full plugin discovery cannot drift apart.""" from hermes_cli.plugins import has_enabled_agent_plugin_mcp as _probe - return _probe(raw_config)