GodsBoy
557 lines
20 KiB
Python
557 lines
20 KiB
Python
"""Compatibility helpers for Agent Plugins v1 portable directory packages.
|
|
|
|
This module validates the versioned portable format locally and translates its
|
|
supported components into records consumed by Hermes' existing skill and MCP
|
|
runtimes. It deliberately performs no schema fetching and imports no plugin
|
|
Python code.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Mapping, Tuple
|
|
|
|
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",
|
|
}
|
|
_AUTHOR_FIELDS = {"name", "email", "url"}
|
|
_STDIO_FIELDS = {"type", "command", "args", "env", "cwd"}
|
|
_REMOTE_FIELDS = {"type", "url", "headers"}
|
|
_PLUGIN_NAME_RE = re.compile(
|
|
r"^(?!.*(?:--|\.\.))[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$"
|
|
)
|
|
_SKILL_NAME_RE = re.compile(r"^(?!.*--)[a-z0-9]+(?:-[a-z0-9]+)*$")
|
|
_PLACEHOLDER_RE = re.compile(r"\$\{(PLUGIN_ROOT|PLUGIN_DATA)\}")
|
|
_HEADER_NAME_RE = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$")
|
|
|
|
|
|
class AgentPluginError(ValueError):
|
|
"""Fatal portable manifest validation failure."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AgentPluginDiagnostic:
|
|
scope: str
|
|
message: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AgentPluginSkill:
|
|
name: str
|
|
description: str
|
|
root: Path
|
|
skill_md: Path
|
|
frontmatter: Mapping[str, Any]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AgentPluginPackage:
|
|
name: str
|
|
version: str
|
|
description: str
|
|
root: Path
|
|
data_root: Path
|
|
manifest: Mapping[str, Any]
|
|
skills: Tuple[AgentPluginSkill, ...]
|
|
mcp_servers: Mapping[str, Dict[str, Any]]
|
|
diagnostics: Tuple[AgentPluginDiagnostic, ...]
|
|
|
|
|
|
def _inside(path: Path, root: Path) -> bool:
|
|
try:
|
|
path.resolve(strict=False).relative_to(root.resolve(strict=True))
|
|
return True
|
|
except (OSError, RuntimeError, ValueError):
|
|
return False
|
|
|
|
|
|
def _read_json_object(path: Path, *, label: str) -> dict:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
raise AgentPluginError(f"{label} is not valid readable JSON: {exc}") from exc
|
|
if not isinstance(value, dict):
|
|
raise AgentPluginError(f"{label} must contain a JSON object")
|
|
return value
|
|
|
|
|
|
def _validate_manifest(root: Path) -> tuple[dict, list[AgentPluginDiagnostic]]:
|
|
manifest_path = root / "plugin.json"
|
|
if not _inside(manifest_path, root) or not manifest_path.is_file():
|
|
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 any(
|
|
not isinstance(value, str) for value in 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")
|
|
unknown = set(author) - _AUTHOR_FIELDS
|
|
if unknown or any(not isinstance(value, str) for value in 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")
|
|
|
|
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
|
|
):
|
|
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:
|
|
return "description must be a non-empty string of at most 1024 characters"
|
|
if "license" in frontmatter and not isinstance(frontmatter["license"], str):
|
|
return "license must be a string"
|
|
if "compatibility" in frontmatter:
|
|
compatibility = frontmatter["compatibility"]
|
|
if not isinstance(compatibility, str) or not 1 <= len(compatibility) <= 500:
|
|
return "compatibility must be a string of 1 to 500 characters"
|
|
if "metadata" in frontmatter:
|
|
metadata = frontmatter["metadata"]
|
|
if not isinstance(metadata, dict) or any(
|
|
not isinstance(key, str) or not isinstance(value, str)
|
|
for key, value in metadata.items()
|
|
):
|
|
return "metadata must map string keys to string values"
|
|
if "allowed-tools" in frontmatter and not isinstance(
|
|
frontmatter["allowed-tools"], str
|
|
):
|
|
return "allowed-tools must be a string"
|
|
return None
|
|
|
|
|
|
def _discover_skills(
|
|
root: Path, diagnostics: list[AgentPluginDiagnostic]
|
|
) -> tuple[AgentPluginSkill, ...]:
|
|
skills_root = root / "skills"
|
|
if not skills_root.exists() and not skills_root.is_symlink():
|
|
return ()
|
|
if not _inside(skills_root, root) or not skills_root.is_dir():
|
|
diagnostics.append(
|
|
AgentPluginDiagnostic("skills", "skills must be an in-root directory")
|
|
)
|
|
return ()
|
|
|
|
skills: list[AgentPluginSkill] = []
|
|
try:
|
|
children = sorted(skills_root.iterdir(), key=lambda path: path.name)
|
|
except OSError as exc:
|
|
diagnostics.append(AgentPluginDiagnostic("skills", f"cannot list skills: {exc}"))
|
|
return ()
|
|
|
|
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}"
|
|
if not _inside(skill_md, root) or not skill_md.is_file():
|
|
diagnostics.append(
|
|
AgentPluginDiagnostic(scope, "SKILL.md must be a regular in-root file")
|
|
)
|
|
continue
|
|
try:
|
|
content = skill_md.read_text(encoding="utf-8")
|
|
content = content.lstrip("\ufeff")
|
|
if not content.startswith("---"):
|
|
raise ValueError("missing YAML frontmatter")
|
|
end_match = re.search(r"\n---\s*\n", content[3:])
|
|
if end_match is None:
|
|
raise ValueError("unterminated YAML frontmatter")
|
|
try:
|
|
parsed = yaml_load(content[3 : end_match.start() + 3])
|
|
except Exception as exc:
|
|
raise ValueError(f"invalid YAML frontmatter: {exc}") from exc
|
|
if not isinstance(parsed, dict):
|
|
raise ValueError("YAML frontmatter must be an object")
|
|
frontmatter = parsed
|
|
except (OSError, UnicodeError, ValueError) as exc:
|
|
diagnostics.append(AgentPluginDiagnostic(scope, f"invalid SKILL.md: {exc}"))
|
|
continue
|
|
error = _valid_skill_frontmatter(frontmatter, child.name)
|
|
if error:
|
|
diagnostics.append(AgentPluginDiagnostic(scope, 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),
|
|
)
|
|
)
|
|
return tuple(skills)
|
|
|
|
|
|
def _expand(value: str, plugin_root: Path, data_root: Path) -> str:
|
|
replacements = {
|
|
"PLUGIN_ROOT": str(plugin_root),
|
|
"PLUGIN_DATA": str(data_root),
|
|
}
|
|
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:
|
|
expanded = _expand(value, plugin_root, data_root) if expand_placeholders else value
|
|
if value.startswith("./"):
|
|
base = plugin_root
|
|
candidate = base / expanded[2:]
|
|
elif value == "${PLUGIN_ROOT}" or value.startswith("${PLUGIN_ROOT}/"):
|
|
base = plugin_root
|
|
candidate = Path(expanded)
|
|
elif value == "${PLUGIN_DATA}" or value.startswith("${PLUGIN_DATA}/"):
|
|
base = data_root
|
|
candidate = Path(expanded)
|
|
else:
|
|
raise ValueError("path must start with ./, ${PLUGIN_ROOT}, or ${PLUGIN_DATA}")
|
|
resolved = candidate.resolve(strict=False)
|
|
try:
|
|
resolved.relative_to(base.resolve(strict=False))
|
|
except (OSError, RuntimeError, ValueError) as exc:
|
|
raise ValueError("path escapes its resolved root") from exc
|
|
return resolved
|
|
|
|
|
|
def _validate_headers(headers: object) -> bool:
|
|
if headers is None:
|
|
return True
|
|
if not isinstance(headers, dict):
|
|
return False
|
|
seen: set[str] = set()
|
|
for name, value in headers.items():
|
|
if (
|
|
not isinstance(name, str)
|
|
or _HEADER_NAME_RE.fullmatch(name) is None
|
|
or not isinstance(value, str)
|
|
or "\r" in value
|
|
or "\n" in value
|
|
or name.lower() in seen
|
|
):
|
|
return False
|
|
seen.add(name.lower())
|
|
return True
|
|
|
|
|
|
def _translate_stdio(
|
|
config: Mapping[str, Any], plugin_root: Path, data_root: Path
|
|
) -> 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,
|
|
)
|
|
)
|
|
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 any(not isinstance(value, str) for value in args):
|
|
raise ValueError("args must be an array of strings")
|
|
env = config.get("env", {})
|
|
if not isinstance(env, dict) or any(
|
|
not isinstance(key, str) or not isinstance(value, str)
|
|
for key, value in env.items()
|
|
):
|
|
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:
|
|
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):
|
|
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),
|
|
"agent_plugin": True,
|
|
}
|
|
|
|
|
|
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 {}
|
|
if not _inside(mcp_path, root) or not mcp_path.is_file():
|
|
diagnostics.append(
|
|
AgentPluginDiagnostic("mcp", "mcp.json must be a regular in-root file")
|
|
)
|
|
return {}
|
|
try:
|
|
config = _read_json_object(mcp_path, label="mcp.json")
|
|
except AgentPluginError as exc:
|
|
diagnostics.append(AgentPluginDiagnostic("mcp", str(exc)))
|
|
return {}
|
|
if set(config) != {"$schema", "mcpServers"}:
|
|
diagnostics.append(
|
|
AgentPluginDiagnostic("mcp", "mcp.json has an invalid top-level shape")
|
|
)
|
|
return {}
|
|
if config.get("$schema") != MCP_SCHEMA_V1:
|
|
diagnostics.append(
|
|
AgentPluginDiagnostic("mcp", "mcp.json declares an unsupported schema")
|
|
)
|
|
return {}
|
|
servers = config.get("mcpServers")
|
|
if not isinstance(servers, dict):
|
|
diagnostics.append(
|
|
AgentPluginDiagnostic("mcp", "mcpServers must be an object")
|
|
)
|
|
return {}
|
|
|
|
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
|
|
server_type = server.get("type")
|
|
if server_type == "stdio":
|
|
try:
|
|
translated_server = _translate_stdio(server, root, data_root)
|
|
if create_data:
|
|
data_root.mkdir(parents=True, exist_ok=True)
|
|
translated[name] = translated_server
|
|
except (OSError, ValueError) as exc:
|
|
diagnostics.append(AgentPluginDiagnostic(scope, str(exc)))
|
|
elif server_type in {"streamable-http", "sse"}:
|
|
if (
|
|
set(server) - _REMOTE_FIELDS
|
|
or not isinstance(server.get("url"), str)
|
|
or not server.get("url")
|
|
or not _validate_headers(server.get("headers"))
|
|
):
|
|
diagnostics.append(AgentPluginDiagnostic(scope, "invalid remote entry"))
|
|
else:
|
|
diagnostics.append(
|
|
AgentPluginDiagnostic(
|
|
scope,
|
|
f"portable {server_type} transport is not supported",
|
|
)
|
|
)
|
|
else:
|
|
diagnostics.append(AgentPluginDiagnostic(scope, "unknown MCP server type"))
|
|
return translated
|
|
|
|
|
|
def load_agent_plugin(plugin_root: Path, data_root: Path) -> AgentPluginPackage:
|
|
"""Validate and translate one installed Agent Plugins v1 package.
|
|
|
|
Fatal manifest errors raise :class:`AgentPluginError`. Component and entry
|
|
failures are returned as diagnostics and isolated to their owning scope.
|
|
"""
|
|
|
|
root = Path(plugin_root).resolve(strict=True)
|
|
if not root.is_dir():
|
|
raise AgentPluginError("plugin root must be a directory")
|
|
manifest, diagnostics = _validate_manifest(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),
|
|
)
|
|
|
|
|
|
def read_agent_plugin_manifest(plugin_root: Path) -> tuple[dict, tuple[AgentPluginDiagnostic, ...]]:
|
|
"""Validate only root ``plugin.json`` without discovering components."""
|
|
|
|
root = Path(plugin_root).resolve(strict=True)
|
|
if not root.is_dir():
|
|
raise AgentPluginError("plugin root must be a directory")
|
|
manifest, diagnostics = _validate_manifest(root)
|
|
return manifest, tuple(diagnostics)
|
|
|
|
|
|
def has_enabled_agent_plugin_mcp(raw_config: Mapping[str, Any]) -> bool:
|
|
"""Cheaply detect an enabled portable package with a root ``mcp.json``.
|
|
|
|
This probe intentionally does not import or register native plugins. Full
|
|
validation remains in the background MCP discovery path.
|
|
"""
|
|
|
|
plugins_config = raw_config.get("plugins")
|
|
if not isinstance(plugins_config, dict):
|
|
return False
|
|
enabled_value = plugins_config.get("enabled")
|
|
if not isinstance(enabled_value, list):
|
|
return False
|
|
enabled = {value for value in enabled_value if isinstance(value, str)}
|
|
disabled_value = plugins_config.get("disabled", [])
|
|
disabled = (
|
|
{value for value in disabled_value if isinstance(value, str)}
|
|
if isinstance(disabled_value, list)
|
|
else set()
|
|
)
|
|
if not enabled:
|
|
return False
|
|
|
|
from hermes_constants import get_hermes_home
|
|
from utils import env_var_enabled
|
|
|
|
if env_var_enabled("HERMES_SAFE_MODE"):
|
|
return False
|
|
|
|
bundled = Path(
|
|
os.getenv("HERMES_BUNDLED_PLUGINS", Path(__file__).resolve().parent.parent / "plugins")
|
|
)
|
|
search_roots: list[tuple[Path, set[str]]] = [
|
|
(
|
|
bundled,
|
|
{"memory", "context_engine", "platforms", "model-providers"},
|
|
),
|
|
(bundled / "platforms", set()),
|
|
(get_hermes_home() / "plugins", set()),
|
|
]
|
|
if env_var_enabled("HERMES_ENABLE_PROJECT_PLUGINS"):
|
|
search_roots.append((Path.cwd() / ".hermes" / "plugins", set()))
|
|
|
|
winners: dict[str, tuple[str, Path]] = {}
|
|
|
|
def scan(directory: Path, *, prefix: str, depth: int, skip: set[str]) -> None:
|
|
try:
|
|
children = sorted(directory.iterdir(), key=lambda path: path.name)
|
|
except OSError:
|
|
return
|
|
for child in children:
|
|
if not child.is_dir() or (depth == 0 and child.name in skip):
|
|
continue
|
|
if (child / "plugin.yaml").exists() or (child / "plugin.yml").exists():
|
|
continue
|
|
portable_file = child / "plugin.json"
|
|
if portable_file.exists() or portable_file.is_symlink():
|
|
try:
|
|
manifest, _ = read_agent_plugin_manifest(child)
|
|
except (AgentPluginError, OSError, RuntimeError):
|
|
continue
|
|
key = f"{prefix}/{child.name}" if prefix else manifest["name"]
|
|
winners[key] = (manifest["name"], child)
|
|
continue
|
|
if depth == 0:
|
|
nested_prefix = f"{prefix}/{child.name}" if prefix else child.name
|
|
scan(child, prefix=nested_prefix, depth=1, skip=set())
|
|
|
|
for search_root, skip_names in search_roots:
|
|
scan(search_root, prefix="", depth=0, skip=skip_names)
|
|
|
|
for key, (name, root) in winners.items():
|
|
if key in disabled or name in disabled:
|
|
continue
|
|
if key not in enabled and name not in enabled:
|
|
continue
|
|
if _discover_mcp(root, get_hermes_home() / "plugin-data" / name, [], create_data=False):
|
|
return True
|
|
return False
|