mirror of
https://github.com/andrewyng/openworker.git
synced 2026-09-11 14:50:14 +00:00
Rohit C PrasadandDevika
2b45018ffa
Imported from andrewyng/aisuite@1b4bbf303e (contents of its platform/ directory, hoisted to the repo root). Development history prior to this commit lives in that repository. Co-authored-by: Devika <devikaverma11@gmail.com>
116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
"""Skill loading — Anthropic SKILL.md format with progressive disclosure.
|
|
|
|
A skill is a folder containing `SKILL.md` (YAML frontmatter: name, description,
|
|
optional allowed-tools) + a markdown body of instructions + optional resources/scripts.
|
|
|
|
Progressive disclosure: at session start only the catalog (name + description) is injected
|
|
into the agent's context; the full body is loaded on demand via the `load_skill` tool.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import aisuite as ai
|
|
|
|
|
|
@dataclass
|
|
class Skill:
|
|
name: str
|
|
description: str
|
|
instructions: str = "" # full body — loaded on demand
|
|
path: Optional[str] = None
|
|
allowed_tools: list[str] = field(default_factory=list)
|
|
|
|
|
|
class SkillLoader:
|
|
def __init__(self, dirs: list[str | Path]) -> None:
|
|
self._skills: dict[str, Skill] = {}
|
|
for directory in dirs:
|
|
self._discover(Path(directory))
|
|
|
|
def _discover(self, directory: Path) -> None:
|
|
if not directory.is_dir():
|
|
return
|
|
for sub in sorted(directory.iterdir()):
|
|
md = sub / "SKILL.md"
|
|
if md.is_file():
|
|
skill = _parse_skill(md)
|
|
self._skills[skill.name] = skill
|
|
|
|
def names(self) -> list[str]:
|
|
return list(self._skills)
|
|
|
|
def get(self, name: str) -> Optional[Skill]:
|
|
return self._skills.get(name)
|
|
|
|
def catalog(self) -> list[dict]:
|
|
return [
|
|
{"name": s.name, "description": s.description}
|
|
for s in self._skills.values()
|
|
]
|
|
|
|
|
|
def _parse_skill(md: Path) -> Skill:
|
|
text = md.read_text(encoding="utf-8")
|
|
name, description, allowed, body = md.parent.name, "", [], text
|
|
if text.startswith("---"):
|
|
end = text.find("\n---", 3)
|
|
if end != -1:
|
|
frontmatter = text[3:end]
|
|
body = text[end + 4 :].lstrip("\n")
|
|
for line in frontmatter.splitlines():
|
|
if ":" not in line:
|
|
continue
|
|
key, value = line.split(":", 1)
|
|
key, value = key.strip().lower(), value.strip()
|
|
if key == "name" and value:
|
|
name = value
|
|
elif key == "description":
|
|
description = value
|
|
elif key in ("allowed-tools", "allowed_tools"):
|
|
allowed = [t.strip() for t in value.split(",") if t.strip()]
|
|
return Skill(
|
|
name=name,
|
|
description=description,
|
|
instructions=body.strip(),
|
|
path=str(md.parent),
|
|
allowed_tools=allowed,
|
|
)
|
|
|
|
|
|
def skill_catalog_text(loader: SkillLoader) -> str:
|
|
catalog = loader.catalog()
|
|
if not catalog:
|
|
return ""
|
|
lines = [f"- {c['name']}: {c['description']}" for c in catalog]
|
|
return (
|
|
"Available skills — call load_skill(name) to load one's full instructions when "
|
|
"it's relevant to the task:\n" + "\n".join(lines)
|
|
)
|
|
|
|
|
|
def skill_tools(loader: SkillLoader) -> list:
|
|
def load_skill(name: str) -> dict:
|
|
"""Load a skill's full instructions + resources path by name. Call this when a
|
|
skill from the catalog is relevant to the current task."""
|
|
skill = loader.get(name)
|
|
if skill is None:
|
|
return {"error": f"unknown skill: {name}", "available": loader.names()}
|
|
return {
|
|
"name": skill.name,
|
|
"instructions": skill.instructions,
|
|
"resources_path": skill.path,
|
|
}
|
|
|
|
return [
|
|
ai.tool(
|
|
load_skill,
|
|
metadata=ai.ToolMetadata(
|
|
category="skills", risk_level="low", capabilities=["load_skill"]
|
|
),
|
|
)
|
|
]
|