mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-08 13:28:43 +03:00
feat(skills): load Agent Plugins v1 skills
This commit is contained in:
@@ -202,7 +202,7 @@ When changing tools, channels, file access, WebUI workspace behavior, or network
|
||||
| Channel | Export a `ChannelPlugin` descriptor, keep its runtime and optional setup surfaces in one package, and follow [`channel-package-guide.md`](./channel-package-guide.md) |
|
||||
| Tool | Implement a tool under `agent/tools/` or expose a plugin entry point |
|
||||
| MCP | Add `tools.mcpServers` config |
|
||||
| Skill | Add workspace skill files under `<workspace>/skills/` or built-in skills under `nanobot/skills/` |
|
||||
| Skill | Add workspace skills under `<workspace>/skills/`, Agent Plugins v1 under `<workspace>/plugins/`, or built-in skills under `nanobot/skills/` |
|
||||
|
||||
Prefer existing registry/discovery patterns over ad hoc wiring.
|
||||
|
||||
|
||||
@@ -2306,6 +2306,31 @@ Disabled skills are excluded from the main agent's skill summary, from always-on
|
||||
|--------|---------|-------------|
|
||||
| `agents.defaults.disabledSkills` | `[]` | List of skill directory names to exclude from loading. Applies to both built-in skills and workspace skills. |
|
||||
|
||||
### Agent Plugins v1 skills
|
||||
|
||||
nanobot also discovers portable [Agent Plugins](https://agent-plugins.org/) placed under
|
||||
`<workspace>/plugins/<plugin>/`. A supported package has a root `plugin.json` that targets
|
||||
Agent Plugins v1 and one or more direct-child skills:
|
||||
|
||||
```text
|
||||
plugins/
|
||||
└── release-tools/
|
||||
├── plugin.json
|
||||
└── skills/
|
||||
└── release-notes/
|
||||
└── SKILL.md
|
||||
```
|
||||
|
||||
Plugin skills use the same progressive loading and `$skill-name` invocation as workspace
|
||||
skills. A workspace skill wins when it has the same name as a plugin skill; plugin skills win
|
||||
over built-in skills. Invalid manifests, invalid Agent Skills, nested skill directories, and
|
||||
paths that resolve outside the plugin root are ignored.
|
||||
|
||||
This initial compatibility layer loads the portable `skills/` component only. Agent Plugins
|
||||
`mcp.json` is not started automatically: local MCP servers execute third-party processes and
|
||||
need an explicit trust and approval flow. Configure a reviewed MCP server through **Apps** or
|
||||
`tools.mcpServers` for now.
|
||||
|
||||
## Tool Hint Max Length
|
||||
|
||||
Tool hints are the short progress messages shown when the agent calls tools (e.g. `$ cd …/project && npm test`). By default, these are truncated at 40 characters, which can make long commands hard to read.
|
||||
|
||||
@@ -0,0 +1,210 @@
|
||||
"""Discover portable Agent Plugins from the agent workspace."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
import yaml
|
||||
from loguru import logger
|
||||
|
||||
AGENT_PLUGIN_SCHEMA = "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json"
|
||||
|
||||
_PLUGIN_NAME = re.compile(r"^(?!.*(?:--|\.\.))[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$")
|
||||
_SKILL_NAME = re.compile(r"^(?!.*--)[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
|
||||
_SKILL_FRONTMATTER = re.compile(r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", re.DOTALL)
|
||||
_MANIFEST_FIELDS = {
|
||||
"$schema",
|
||||
"name",
|
||||
"version",
|
||||
"description",
|
||||
"author",
|
||||
"homepage",
|
||||
"repository",
|
||||
"license",
|
||||
"keywords",
|
||||
"extensions",
|
||||
}
|
||||
_STRING_FIELDS = {"version", "description", "homepage", "repository", "license"}
|
||||
_AUTHOR_FIELDS = {"name", "email", "url"}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AgentPluginSkill:
|
||||
"""One skill supplied by a valid Agent Plugins v1 package."""
|
||||
|
||||
name: str
|
||||
path: Path
|
||||
plugin: str
|
||||
|
||||
|
||||
def discover_agent_plugin_skills(workspace: Path) -> list[AgentPluginSkill]:
|
||||
"""Discover direct-child skills under ``<workspace>/plugins/*``.
|
||||
|
||||
Agent Plugins does not prescribe an install location. nanobot uses the
|
||||
workspace ``plugins`` directory so packages stay explicit and portable
|
||||
with the rest of the agent workspace.
|
||||
"""
|
||||
workspace = workspace.expanduser().resolve()
|
||||
plugins_root = workspace / "plugins"
|
||||
if not plugins_root.is_dir():
|
||||
return []
|
||||
try:
|
||||
resolved_plugins_root = plugins_root.resolve(strict=True)
|
||||
except OSError:
|
||||
return []
|
||||
if not resolved_plugins_root.is_relative_to(workspace):
|
||||
logger.warning("Ignoring Agent Plugins directory outside the workspace")
|
||||
return []
|
||||
|
||||
try:
|
||||
candidates = sorted(plugins_root.iterdir(), key=lambda path: path.name)
|
||||
except OSError as exc:
|
||||
logger.warning("Could not inspect Agent Plugins directory: {}", exc)
|
||||
return []
|
||||
|
||||
skills: list[AgentPluginSkill] = []
|
||||
for candidate in candidates:
|
||||
plugin_root = _contained_directory(candidate, resolved_plugins_root)
|
||||
if plugin_root is None:
|
||||
continue
|
||||
plugin_name = _load_manifest_name(plugin_root)
|
||||
if plugin_name is None:
|
||||
continue
|
||||
skills.extend(_discover_plugin_skills(plugin_name, plugin_root))
|
||||
return skills
|
||||
|
||||
|
||||
def _load_manifest_name(plugin_root: Path) -> str | None:
|
||||
manifest = _contained_file(plugin_root / "plugin.json", plugin_root)
|
||||
if manifest is None:
|
||||
return None
|
||||
try:
|
||||
value = cast(object, json.loads(manifest.read_text(encoding="utf-8")))
|
||||
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
||||
logger.warning("Ignoring invalid Agent Plugin manifest '{}': {}", manifest, exc)
|
||||
return None
|
||||
if not isinstance(value, dict):
|
||||
logger.warning("Ignoring Agent Plugin manifest '{}': expected a JSON object", manifest)
|
||||
return None
|
||||
|
||||
payload = cast(dict[str, Any], value)
|
||||
if payload.get("$schema") != AGENT_PLUGIN_SCHEMA:
|
||||
return None
|
||||
name = payload.get("name")
|
||||
if (
|
||||
not isinstance(name, str)
|
||||
or len(name) > 64
|
||||
or _PLUGIN_NAME.fullmatch(name) is None
|
||||
):
|
||||
logger.warning("Ignoring Agent Plugin manifest '{}': invalid name", manifest)
|
||||
return None
|
||||
if not _valid_optional_fields(payload):
|
||||
logger.warning("Ignoring Agent Plugin manifest '{}': invalid metadata", manifest)
|
||||
return None
|
||||
|
||||
for field in payload.keys() - _MANIFEST_FIELDS:
|
||||
logger.warning("Ignoring unknown Agent Plugin manifest field '{}' in '{}'", field, manifest)
|
||||
if "extensions" in payload and not isinstance(payload["extensions"], dict):
|
||||
logger.warning("Ignoring non-object Agent Plugin extensions in '{}'", manifest)
|
||||
return name
|
||||
|
||||
|
||||
def _valid_optional_fields(payload: dict[str, Any]) -> bool:
|
||||
if any(field in payload and not isinstance(payload[field], str) for field in _STRING_FIELDS):
|
||||
return False
|
||||
keywords = payload.get("keywords")
|
||||
if "keywords" in payload and (
|
||||
not isinstance(keywords, list)
|
||||
or not all(isinstance(keyword, str) for keyword in cast(list[object], keywords))
|
||||
):
|
||||
return False
|
||||
author = payload.get("author")
|
||||
if "author" not in payload:
|
||||
return True
|
||||
if not isinstance(author, dict):
|
||||
return False
|
||||
author_payload = cast(dict[str, object], author)
|
||||
return not (author_payload.keys() - _AUTHOR_FIELDS) and all(
|
||||
isinstance(value, str) for value in author_payload.values()
|
||||
)
|
||||
|
||||
|
||||
def _discover_plugin_skills(plugin_name: str, plugin_root: Path) -> list[AgentPluginSkill]:
|
||||
skills_root = plugin_root / "skills"
|
||||
if not skills_root.exists():
|
||||
return []
|
||||
resolved_skills_root = _contained_directory(skills_root, plugin_root)
|
||||
if resolved_skills_root is None:
|
||||
logger.warning("Ignoring invalid skills component in Agent Plugin '{}'", plugin_name)
|
||||
return []
|
||||
|
||||
try:
|
||||
candidates = sorted(skills_root.iterdir(), key=lambda path: path.name)
|
||||
except OSError as exc:
|
||||
logger.warning("Could not inspect Agent Plugin '{}' skills: {}", plugin_name, exc)
|
||||
return []
|
||||
|
||||
skills: list[AgentPluginSkill] = []
|
||||
for candidate in candidates:
|
||||
skill_root = _contained_directory(candidate, resolved_skills_root)
|
||||
if skill_root is None:
|
||||
continue
|
||||
skill_file = _contained_file(skill_root / "SKILL.md", plugin_root)
|
||||
if skill_file is None or not _valid_skill(skill_file, candidate.name, plugin_name):
|
||||
continue
|
||||
skills.append(
|
||||
AgentPluginSkill(name=candidate.name, path=skill_file, plugin=plugin_name)
|
||||
)
|
||||
return skills
|
||||
|
||||
|
||||
def _valid_skill(path: Path, directory_name: str, plugin_name: str) -> bool:
|
||||
try:
|
||||
content = path.read_text(encoding="utf-8")
|
||||
except (OSError, UnicodeError):
|
||||
return False
|
||||
match = _SKILL_FRONTMATTER.match(content)
|
||||
if match is None:
|
||||
logger.warning("Ignoring Agent Plugin '{}' skill '{}': invalid frontmatter", plugin_name, directory_name)
|
||||
return False
|
||||
try:
|
||||
metadata = cast(object, yaml.safe_load(match.group(1)))
|
||||
except yaml.YAMLError:
|
||||
metadata = None
|
||||
if not isinstance(metadata, dict):
|
||||
logger.warning("Ignoring Agent Plugin '{}' skill '{}': invalid frontmatter", plugin_name, directory_name)
|
||||
return False
|
||||
payload = cast(dict[object, object], metadata)
|
||||
name = payload.get("name")
|
||||
description = payload.get("description")
|
||||
valid = (
|
||||
name == directory_name
|
||||
and isinstance(name, str)
|
||||
and len(name) <= 64
|
||||
and _SKILL_NAME.fullmatch(name) is not None
|
||||
and isinstance(description, str)
|
||||
and 1 <= len(description.strip()) <= 1024
|
||||
)
|
||||
if not valid:
|
||||
logger.warning("Ignoring Agent Plugin '{}' skill '{}': invalid metadata", plugin_name, directory_name)
|
||||
return valid
|
||||
|
||||
|
||||
def _contained_directory(path: Path, root: Path) -> Path | None:
|
||||
try:
|
||||
resolved = path.resolve(strict=True)
|
||||
except OSError:
|
||||
return None
|
||||
return resolved if resolved.is_dir() and resolved.is_relative_to(root) else None
|
||||
|
||||
|
||||
def _contained_file(path: Path, root: Path) -> Path | None:
|
||||
try:
|
||||
resolved = path.resolve(strict=True)
|
||||
except OSError:
|
||||
return None
|
||||
return resolved if resolved.is_file() and resolved.is_relative_to(root) else None
|
||||
+27
-8
@@ -9,6 +9,8 @@ from typing import Any, cast
|
||||
|
||||
import yaml
|
||||
|
||||
from nanobot.agent.agent_plugins import discover_agent_plugin_skills
|
||||
|
||||
# Default builtin skills directory (relative to this file)
|
||||
BUILTIN_SKILLS_DIR = Path(__file__).parent.parent / "skills"
|
||||
|
||||
@@ -33,6 +35,7 @@ class SkillsLoader:
|
||||
self.workspace_skills = workspace / "skills"
|
||||
self.builtin_skills = builtin_skills_dir or BUILTIN_SKILLS_DIR
|
||||
self.disabled_skills = disabled_skills or set()
|
||||
self.plugin_skills = discover_agent_plugin_skills(workspace)
|
||||
|
||||
def _skill_entries_from_dir(self, base: Path, source: str, *, skip_names: set[str] | None = None) -> list[dict[str, str]]:
|
||||
if not base.exists():
|
||||
@@ -61,10 +64,22 @@ class SkillsLoader:
|
||||
List of skill info dicts with 'name', 'path', 'source'.
|
||||
"""
|
||||
skills = self._skill_entries_from_dir(self.workspace_skills, "workspace")
|
||||
workspace_names = {entry["name"] for entry in skills}
|
||||
seen_names = {entry["name"] for entry in skills}
|
||||
for plugin_skill in self.plugin_skills:
|
||||
if plugin_skill.name in seen_names:
|
||||
continue
|
||||
skills.append(
|
||||
{
|
||||
"name": plugin_skill.name,
|
||||
"path": str(plugin_skill.path),
|
||||
"source": "plugin",
|
||||
"plugin": plugin_skill.plugin,
|
||||
}
|
||||
)
|
||||
seen_names.add(plugin_skill.name)
|
||||
if self.builtin_skills and self.builtin_skills.exists():
|
||||
skills.extend(
|
||||
self._skill_entries_from_dir(self.builtin_skills, "builtin", skip_names=workspace_names)
|
||||
self._skill_entries_from_dir(self.builtin_skills, "builtin", skip_names=seen_names)
|
||||
)
|
||||
|
||||
if self.disabled_skills:
|
||||
@@ -84,13 +99,16 @@ class SkillsLoader:
|
||||
Returns:
|
||||
Skill content or None if not found.
|
||||
"""
|
||||
roots = [self.workspace_skills]
|
||||
workspace_path = self.workspace_skills / name / "SKILL.md"
|
||||
if workspace_path.exists():
|
||||
return workspace_path.read_text(encoding="utf-8")
|
||||
for plugin_skill in self.plugin_skills:
|
||||
if plugin_skill.name == name:
|
||||
return plugin_skill.path.read_text(encoding="utf-8")
|
||||
if self.builtin_skills:
|
||||
roots.append(self.builtin_skills)
|
||||
for root in roots:
|
||||
path = root / name / "SKILL.md"
|
||||
if path.exists():
|
||||
return path.read_text(encoding="utf-8")
|
||||
builtin_path = self.builtin_skills / name / "SKILL.md"
|
||||
if builtin_path.exists():
|
||||
return builtin_path.read_text(encoding="utf-8")
|
||||
return None
|
||||
|
||||
def load_skills_for_context(self, skill_names: list[str]) -> str:
|
||||
@@ -145,6 +163,7 @@ class SkillsLoader:
|
||||
sections: list[str] = []
|
||||
groups = (
|
||||
("Workspace skills", "workspace", self.workspace_skills),
|
||||
("Agent Plugin skills", "plugin", self.workspace / "plugins"),
|
||||
("Built-in skills", "builtin", self.builtin_skills),
|
||||
)
|
||||
for label, source, root in groups:
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from nanobot.agent.agent_plugins import AGENT_PLUGIN_SCHEMA, discover_agent_plugin_skills
|
||||
from nanobot.agent.skills import SkillsLoader
|
||||
|
||||
|
||||
def _write_skill(root: Path, name: str, *, description: str = "Plugin skill.") -> Path:
|
||||
skill = root / "skills" / name
|
||||
skill.mkdir(parents=True)
|
||||
(skill / "SKILL.md").write_text(
|
||||
f"---\nname: {name}\ndescription: {description}\n---\n\n# {name}\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return skill
|
||||
|
||||
|
||||
def _write_plugin(
|
||||
workspace: Path,
|
||||
directory: str,
|
||||
*,
|
||||
name: str | None = None,
|
||||
manifest: dict[str, object] | None = None,
|
||||
) -> Path:
|
||||
root = workspace / "plugins" / directory
|
||||
root.mkdir(parents=True)
|
||||
payload = manifest or {
|
||||
"$schema": AGENT_PLUGIN_SCHEMA,
|
||||
"name": name or directory,
|
||||
}
|
||||
(root / "plugin.json").write_text(json.dumps(payload), encoding="utf-8")
|
||||
return root
|
||||
|
||||
|
||||
def test_skills_loader_discovers_agent_plugin_skill(tmp_path: Path) -> None:
|
||||
plugin = _write_plugin(tmp_path, "acme-tools")
|
||||
_write_skill(plugin, "release-notes", description="Draft release notes from changes.")
|
||||
|
||||
loader = SkillsLoader(tmp_path, builtin_skills_dir=tmp_path / "builtin")
|
||||
|
||||
assert loader.list_skills() == [
|
||||
{
|
||||
"name": "release-notes",
|
||||
"path": str(plugin / "skills" / "release-notes" / "SKILL.md"),
|
||||
"source": "plugin",
|
||||
"plugin": "acme-tools",
|
||||
}
|
||||
]
|
||||
assert loader.get_explicitly_invoked_skills("Use $release-notes") == ["release-notes"]
|
||||
assert "Draft release notes" in (loader.load_skill("release-notes") or "")
|
||||
assert "### Agent Plugin skills" in loader.build_skills_summary()
|
||||
assert "`acme-tools/skills/release-notes/SKILL.md`" in loader.build_skills_summary()
|
||||
|
||||
|
||||
def test_agent_plugin_skills_are_direct_children_only(tmp_path: Path) -> None:
|
||||
plugin = _write_plugin(tmp_path, "acme-tools")
|
||||
_write_skill(plugin, "direct")
|
||||
nested = plugin / "skills" / "group" / "nested"
|
||||
nested.mkdir(parents=True)
|
||||
(nested / "SKILL.md").write_text(
|
||||
"---\nname: nested\ndescription: Nested skill.\n---\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
assert [skill.name for skill in discover_agent_plugin_skills(tmp_path)] == ["direct"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"manifest",
|
||||
[
|
||||
{"$schema": "https://agent-plugins.org/schemas/2.0.0/plugin.schema.json", "name": "demo"},
|
||||
{"$schema": AGENT_PLUGIN_SCHEMA, "name": "Bad-Name"},
|
||||
{"$schema": AGENT_PLUGIN_SCHEMA, "name": "demo", "author": None},
|
||||
{"$schema": AGENT_PLUGIN_SCHEMA, "name": "demo", "keywords": None},
|
||||
],
|
||||
)
|
||||
def test_invalid_agent_plugin_manifest_is_skipped(
|
||||
tmp_path: Path,
|
||||
manifest: dict[str, object],
|
||||
) -> None:
|
||||
plugin = _write_plugin(tmp_path, "demo", manifest=manifest)
|
||||
_write_skill(plugin, "example")
|
||||
|
||||
assert discover_agent_plugin_skills(tmp_path) == []
|
||||
|
||||
|
||||
def test_unknown_manifest_fields_and_non_object_extensions_are_ignored(tmp_path: Path) -> None:
|
||||
plugin = _write_plugin(
|
||||
tmp_path,
|
||||
"demo",
|
||||
manifest={
|
||||
"$schema": AGENT_PLUGIN_SCHEMA,
|
||||
"name": "demo",
|
||||
"futureField": True,
|
||||
"extensions": "invalid but non-fatal",
|
||||
},
|
||||
)
|
||||
_write_skill(plugin, "example")
|
||||
|
||||
assert [skill.name for skill in discover_agent_plugin_skills(tmp_path)] == ["example"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("skill_name", "frontmatter"),
|
||||
[
|
||||
("wrong-directory", "name: another\ndescription: Mismatch."),
|
||||
("missing-description", "name: missing-description"),
|
||||
("Bad-Name", "name: Bad-Name\ndescription: Invalid name."),
|
||||
],
|
||||
)
|
||||
def test_invalid_agent_skill_is_skipped(
|
||||
tmp_path: Path,
|
||||
skill_name: str,
|
||||
frontmatter: str,
|
||||
) -> None:
|
||||
plugin = _write_plugin(tmp_path, "demo")
|
||||
skill = plugin / "skills" / skill_name
|
||||
skill.mkdir(parents=True)
|
||||
(skill / "SKILL.md").write_text(f"---\n{frontmatter}\n---\n", encoding="utf-8")
|
||||
|
||||
assert discover_agent_plugin_skills(tmp_path) == []
|
||||
|
||||
|
||||
def test_workspace_skill_overrides_plugin_skill(tmp_path: Path) -> None:
|
||||
plugin = _write_plugin(tmp_path, "demo")
|
||||
_write_skill(plugin, "shared", description="Plugin version.")
|
||||
workspace_skill = tmp_path / "skills" / "shared"
|
||||
workspace_skill.mkdir(parents=True)
|
||||
(workspace_skill / "SKILL.md").write_text(
|
||||
"---\nname: shared\ndescription: Workspace version.\n---\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
loader = SkillsLoader(tmp_path, builtin_skills_dir=tmp_path / "builtin")
|
||||
|
||||
assert [entry["source"] for entry in loader.list_skills()] == ["workspace"]
|
||||
assert "Workspace version" in (loader.load_skill("shared") or "")
|
||||
|
||||
|
||||
def test_plugin_skill_symlink_cannot_escape_plugin_root(tmp_path: Path) -> None:
|
||||
plugin = _write_plugin(tmp_path, "demo")
|
||||
outside = tmp_path / "outside"
|
||||
_write_skill(outside, "escaped")
|
||||
skills_root = plugin / "skills"
|
||||
skills_root.mkdir()
|
||||
try:
|
||||
(skills_root / "escaped").symlink_to(
|
||||
outside / "skills" / "escaped",
|
||||
target_is_directory=True,
|
||||
)
|
||||
except OSError as exc:
|
||||
pytest.skip(f"directory symlink unavailable: {exc}")
|
||||
|
||||
assert discover_agent_plugin_skills(tmp_path) == []
|
||||
Reference in New Issue
Block a user