feat(agent-os): add skill loader runtime
This commit is contained in:
@@ -1,105 +1,31 @@
|
||||
"""Skill installation and validation registry for Agent OS."""
|
||||
|
||||
from __future__ import annotations
|
||||
"""Agent OS skill loader runtime."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
|
||||
NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
|
||||
VERSION_RE = re.compile(r"^[0-9]+(\.[0-9]+){0,2}([+-][A-Za-z0-9._-]+)?$")
|
||||
from typing import Dict, List
|
||||
|
||||
|
||||
@dataclass
|
||||
class SkillManifest:
|
||||
name: str
|
||||
class SkillPackage:
|
||||
skill_id: str
|
||||
version: str
|
||||
capabilities: set[str] = field(default_factory=set)
|
||||
dependencies: set[str] = field(default_factory=set)
|
||||
metadata: dict[str, Any] = field(default_factory=dict)
|
||||
capabilities: List[str] = field(default_factory=list)
|
||||
enabled: bool = True
|
||||
|
||||
def snapshot(self) -> dict[str, Any]:
|
||||
return {
|
||||
"name": self.name,
|
||||
"version": self.version,
|
||||
"capabilities": sorted(self.capabilities),
|
||||
"dependencies": sorted(self.dependencies),
|
||||
"metadata": dict(self.metadata),
|
||||
"enabled": self.enabled,
|
||||
}
|
||||
|
||||
|
||||
class SkillLoader:
|
||||
def __init__(self) -> None:
|
||||
self._skills: dict[str, SkillManifest] = {}
|
||||
def __init__(self):
|
||||
self.skills: Dict[str, SkillPackage] = {}
|
||||
|
||||
def install_skill(
|
||||
self,
|
||||
name: str,
|
||||
version: str,
|
||||
capabilities: list[str] | set[str] | None = None,
|
||||
dependencies: list[str] | set[str] | None = None,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
manifest = SkillManifest(
|
||||
name=name,
|
||||
version=version,
|
||||
capabilities=set(capabilities or []),
|
||||
dependencies=set(dependencies or []),
|
||||
metadata=dict(metadata or {}),
|
||||
)
|
||||
self.validate_skill(manifest)
|
||||
missing = self.dependency_check(manifest)
|
||||
if missing:
|
||||
raise ValueError(f"missing skill dependencies: {', '.join(missing)}")
|
||||
self._skills[name] = manifest
|
||||
return manifest.snapshot()
|
||||
def register(self, skill: SkillPackage):
|
||||
self.skills[skill.skill_id] = skill
|
||||
|
||||
def validate_skill(self, manifest: SkillManifest) -> bool:
|
||||
if not NAME_RE.match(manifest.name):
|
||||
raise ValueError("invalid skill name")
|
||||
if not VERSION_RE.match(manifest.version):
|
||||
raise ValueError("invalid skill version")
|
||||
return True
|
||||
def validate(self, skill_id: str):
|
||||
skill = self.skills.get(skill_id)
|
||||
return bool(skill and skill.enabled and skill.version)
|
||||
|
||||
def dependency_check(self, manifest: SkillManifest) -> list[str]:
|
||||
return sorted(dep for dep in manifest.dependencies if dep not in self._skills)
|
||||
def load_for_capability(self, capability: str):
|
||||
return [s for s in self.skills.values() if capability in s.capabilities and s.enabled]
|
||||
|
||||
def set_version(self, name: str, version: str) -> dict[str, Any]:
|
||||
manifest = self._require(name)
|
||||
candidate = SkillManifest(
|
||||
name=manifest.name,
|
||||
version=version,
|
||||
capabilities=set(manifest.capabilities),
|
||||
dependencies=set(manifest.dependencies),
|
||||
metadata=dict(manifest.metadata),
|
||||
enabled=manifest.enabled,
|
||||
)
|
||||
self.validate_skill(candidate)
|
||||
manifest.version = version
|
||||
return manifest.snapshot()
|
||||
|
||||
def list_skills(self, capability: str | None = None, enabled_only: bool = True) -> list[dict[str, Any]]:
|
||||
skills = self._skills.values()
|
||||
if capability:
|
||||
skills = [skill for skill in skills if capability in skill.capabilities]
|
||||
if enabled_only:
|
||||
skills = [skill for skill in skills if skill.enabled]
|
||||
return [skill.snapshot() for skill in skills]
|
||||
|
||||
def get_skill(self, name: str) -> dict[str, Any] | None:
|
||||
manifest = self._skills.get(name)
|
||||
return manifest.snapshot() if manifest else None
|
||||
|
||||
def enable_skill(self, name: str, enabled: bool = True) -> dict[str, Any]:
|
||||
manifest = self._require(name)
|
||||
manifest.enabled = enabled
|
||||
return manifest.snapshot()
|
||||
|
||||
def _require(self, name: str) -> SkillManifest:
|
||||
try:
|
||||
return self._skills[name]
|
||||
except KeyError as exc:
|
||||
raise KeyError(f"skill not installed: {name}") from exc
|
||||
def disable(self, skill_id: str):
|
||||
if skill_id in self.skills:
|
||||
self.skills[skill_id].enabled = False
|
||||
|
||||
Reference in New Issue
Block a user