Bases: BaseRegistry[str, Skill]
Registry for Claude Code Skills with auto-discovery.
Source code in src/llmling_agent/skills/registry.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157 | class SkillsRegistry(BaseRegistry[str, Skill]):
"""Registry for Claude Code Skills with auto-discovery."""
DEFAULT_SKILL_PATHS: ClassVar = ["~/.claude/skills/", ".claude/skills/"]
def __init__(self, skills_dirs: Sequence[JoinablePathLike] | None = None) -> None:
"""Initialize with custom skill directories or auto-detect."""
super().__init__()
if skills_dirs:
self.skills_dirs = [UPath(i) for i in skills_dirs or []]
else:
self.skills_dirs = [UPath(i) for i in self.DEFAULT_SKILL_PATHS or []]
async def discover_skills(self) -> None:
"""Scan filesystem and register all found skills.
Args:
filesystem: Optional async filesystem to use. If None, will use upath_to_fs()
to get appropriate filesystem for each skills directory.
"""
for skills_dir in self.skills_dirs:
await self.register_skills_from_path(skills_dir)
async def register_skills_from_path(
self,
skills_dir: UPath | AbstractFileSystem,
**storage_options: Any,
) -> None:
"""Register skills from a given path.
Args:
skills_dir: Path to the directory containing skills.
storage_options: Additional options to pass to the filesystem.
"""
if isinstance(skills_dir, AbstractFileSystem):
fs = skills_dir
if not isinstance(fs, AsyncFileSystem):
fs = AsyncFileSystemWrapper(fs)
else:
fs = upath_to_fs(skills_dir, **storage_options)
try:
# List entries in skills directory
entries = await fs._ls(fs.root_marker, detail=True)
except FileNotFoundError:
logger.warning("Skills directory not found", path=skills_dir)
return
# Filter for directories that might contain skills
skill_dirs = [entry for entry in entries if entry.get("type") == "directory"]
for skill_entry in skill_dirs:
skill_name = skill_entry["name"].lstrip("./")
skill_dir_path = skills_dir / skill_name
try:
await fs._cat(f"{skill_name}/SKILL.md")
except FileNotFoundError:
continue
try:
skill = self._parse_skill(skill_dir_path)
self.register(skill.name, skill, replace=True)
except Exception as e: # noqa: BLE001
# Log but don't fail discovery for one bad skill
print(f"Warning: Failed to parse skill at {skill_dir_path}: {e}")
def _parse_skill(self, skill_dir: JoinablePathLike) -> Skill:
"""Parse a SKILL.md file and extract metadata."""
skill_file = UPath(skill_dir) / "SKILL.md"
content = skill_file.read_text()
# Extract YAML frontmatter
frontmatter_match = re.match(r"^---\s*\n(.*?)\n---\s*\n", content, re.DOTALL)
if not frontmatter_match:
msg = f"No YAML frontmatter found in {skill_file}"
raise ToolError(msg)
import yamling
try:
metadata = yamling.load_yaml(frontmatter_match.group(1))
except yamling.YAMLError as e:
msg = f"Invalid YAML frontmatter in {skill_file}: {e}"
raise ToolError(msg) from e
# Validate required fields
if not isinstance(metadata, dict):
msg = f"YAML frontmatter must be a dictionary in {skill_file}"
raise ToolError(msg)
name = metadata.get("name")
description = metadata.get("description")
if not name:
msg = f"Missing 'name' field in {skill_file}"
raise ToolError(msg)
if not description:
msg = f"Missing 'description' field in {skill_file}"
raise ToolError(msg)
# Validate limits
if len(name) > SKILL_NAME_LIMIT:
msg = f"{skill_file}: Skill name exceeds {SKILL_NAME_LIMIT} chars"
raise ToolError(msg)
if len(description) > SKILL_DESCRIPTION_LIMIT:
msg = f"{skill_file}: Skill description exceeds {SKILL_DESCRIPTION_LIMIT} chars"
raise ToolError(msg)
return Skill(name=name, description=description, skill_path=UPath(skill_dir))
@property
def _error_class(self) -> type[ToolError]:
"""Error class to use for this registry."""
return ToolError
def _validate_item(self, item: Any) -> Skill:
"""Validate and possibly transform item before registration."""
if not isinstance(item, Skill):
msg = f"Expected Skill instance, got {type(item)}"
raise ToolError(msg)
return item
def get_skill_instructions(self, skill_name: str) -> str:
"""Lazy load full instructions for a skill."""
skill = self.get(skill_name)
return skill.load_instructions()
|
__init__(skills_dirs: Sequence[JoinablePathLike] | None = None) -> None
Initialize with custom skill directories or auto-detect.
Source code in src/llmling_agent/skills/registry.py
| def __init__(self, skills_dirs: Sequence[JoinablePathLike] | None = None) -> None:
"""Initialize with custom skill directories or auto-detect."""
super().__init__()
if skills_dirs:
self.skills_dirs = [UPath(i) for i in skills_dirs or []]
else:
self.skills_dirs = [UPath(i) for i in self.DEFAULT_SKILL_PATHS or []]
|
async
discover_skills() -> None
Scan filesystem and register all found skills.
Parameters:
| Name |
Type |
Description |
Default |
filesystem
|
|
Optional async filesystem to use. If None, will use upath_to_fs()
to get appropriate filesystem for each skills directory.
|
required
|
Source code in src/llmling_agent/skills/registry.py
46
47
48
49
50
51
52
53
54 | async def discover_skills(self) -> None:
"""Scan filesystem and register all found skills.
Args:
filesystem: Optional async filesystem to use. If None, will use upath_to_fs()
to get appropriate filesystem for each skills directory.
"""
for skills_dir in self.skills_dirs:
await self.register_skills_from_path(skills_dir)
|
get_skill_instructions(skill_name: str) -> str
Lazy load full instructions for a skill.
Source code in src/llmling_agent/skills/registry.py
| def get_skill_instructions(self, skill_name: str) -> str:
"""Lazy load full instructions for a skill."""
skill = self.get(skill_name)
return skill.load_instructions()
|
async
register_skills_from_path(
skills_dir: UPath | AbstractFileSystem, **storage_options: Any
) -> None
Register skills from a given path.
Parameters:
| Name |
Type |
Description |
Default |
skills_dir
|
UPath | AbstractFileSystem
|
Path to the directory containing skills.
|
required
|
storage_options
|
Any
|
Additional options to pass to the filesystem.
|
{}
|
Source code in src/llmling_agent/skills/registry.py
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97 | async def register_skills_from_path(
self,
skills_dir: UPath | AbstractFileSystem,
**storage_options: Any,
) -> None:
"""Register skills from a given path.
Args:
skills_dir: Path to the directory containing skills.
storage_options: Additional options to pass to the filesystem.
"""
if isinstance(skills_dir, AbstractFileSystem):
fs = skills_dir
if not isinstance(fs, AsyncFileSystem):
fs = AsyncFileSystemWrapper(fs)
else:
fs = upath_to_fs(skills_dir, **storage_options)
try:
# List entries in skills directory
entries = await fs._ls(fs.root_marker, detail=True)
except FileNotFoundError:
logger.warning("Skills directory not found", path=skills_dir)
return
# Filter for directories that might contain skills
skill_dirs = [entry for entry in entries if entry.get("type") == "directory"]
for skill_entry in skill_dirs:
skill_name = skill_entry["name"].lstrip("./")
skill_dir_path = skills_dir / skill_name
try:
await fs._cat(f"{skill_name}/SKILL.md")
except FileNotFoundError:
continue
try:
skill = self._parse_skill(skill_dir_path)
self.register(skill.name, skill, replace=True)
except Exception as e: # noqa: BLE001
# Log but don't fail discovery for one bad skill
print(f"Warning: Failed to parse skill at {skill_dir_path}: {e}")
|