# Copyright 2025 The EasyDeL/Calute Author @erfanzar (Erfan Zare Chavoshi).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Skill discovery and management for Calute.
Skills are self-contained instruction packages defined by SKILL.md files.
Each skill provides metadata, instructions, and optional resource references
that can be injected into agent prompts on demand.
Skill directory layout::
skills/
├── web_research/
│ └── SKILL.md
├── code_review/
│ └── SKILL.md
└── data_analysis/
├── SKILL.md
└── templates/
└── report.md
SKILL.md format (YAML frontmatter + markdown body)::
---
name: web_research
description: Search the web and synthesize findings
version: "1.0"
tags: [research, web]
resources:
- templates/query.md
---
# Web Research Skill
Instructions for conducting web research...
"""
from __future__ import annotations
import logging
import re
import typing as tp
from dataclasses import dataclass, field
from pathlib import Path
logger = logging.getLogger(__name__)
[docs]@dataclass
class Skill:
"""A fully loaded skill consisting of metadata, instructions, and resource references.
Attributes:
metadata: The :class:`SkillMetadata` parsed from the SKILL.md
frontmatter.
instructions: The markdown body of the SKILL.md file, containing
the instructions to be injected into the agent prompt.
source_path: The filesystem path to the SKILL.md file this skill
was loaded from.
resources_dir: The directory containing supplementary resource files,
or ``None`` if no resources are declared.
"""
metadata: SkillMetadata
instructions: str
source_path: Path
resources_dir: Path | None = None
@property
def name(self) -> str:
"""Return the skill's unique name from its metadata.
Returns:
The skill name string.
"""
return self.metadata.name
[docs] def to_prompt_section(self) -> str:
"""Format the skill as a markdown section for injection into a system prompt.
Produces a section with a ``## Skill: <name>`` heading, an optional
description line, and the full instruction body.
Returns:
A markdown-formatted string ready to be appended to a system
prompt.
"""
header = f"## Skill: {self.metadata.name}"
if self.metadata.description:
header += f"\n{self.metadata.description}"
return f"{header}\n\n{self.instructions}"
[docs]def parse_skill_md(content: str, source_path: Path) -> Skill:
"""Parse a SKILL.md file's content into a :class:`Skill` object.
The file is expected to contain optional YAML frontmatter delimited by
``---`` lines, followed by a markdown body. If PyYAML is installed the
frontmatter is parsed with ``yaml.safe_load``; otherwise a simple
line-by-line key-value parser is used as a fallback.
When no ``name`` key is present in the frontmatter, the parent directory
name of *source_path* is used as the skill name.
Args:
content: The full text content of the SKILL.md file.
source_path: The filesystem path to the SKILL.md file (used to
derive the skill name and resources directory).
Returns:
A :class:`Skill` instance populated with the parsed metadata and
instruction body.
Example:
>>> from pathlib import Path
>>> content = "---\\nname: demo\\n---\\nDo something."
>>> skill = parse_skill_md(content, Path("/skills/demo/SKILL.md"))
>>> skill.name
'demo'
"""
metadata_dict: dict = {}
body = content
fm_match = re.match(r"^---\s*\n(.*?)\n---\s*\n?(.*)", content, re.DOTALL)
if fm_match:
fm_text = fm_match.group(1)
body = fm_match.group(2).strip()
try:
import yaml
metadata_dict = yaml.safe_load(fm_text) or {}
except ImportError:
for line in fm_text.strip().splitlines():
line = line.strip()
if ":" in line:
key, _, value = line.partition(":")
key = key.strip()
value = value.strip().strip('"').strip("'")
if value.startswith("[") and value.endswith("]"):
value = [v.strip().strip('"').strip("'") for v in value[1:-1].split(",")]
metadata_dict[key] = value
name = metadata_dict.get("name", source_path.parent.name)
metadata = SkillMetadata(
name=name,
description=metadata_dict.get("description", ""),
version=str(metadata_dict.get("version", "1.0")),
tags=metadata_dict.get("tags", []),
resources=metadata_dict.get("resources", []),
author=metadata_dict.get("author", ""),
dependencies=metadata_dict.get("dependencies", []),
required_tools=metadata_dict.get("required_tools", []),
)
resources_dir = source_path.parent if metadata.resources else None
return Skill(
metadata=metadata,
instructions=body,
source_path=source_path,
resources_dir=resources_dir,
)
[docs]class SkillRegistry:
"""Discovers, indexes, and provides skills for prompt injection.
The registry scans one or more directories for ``SKILL.md`` files,
parses them, and stores the resulting :class:`Skill` objects for
later retrieval by name, tag, or free-text search.
Attributes:
_skills: Internal mapping of skill name to :class:`Skill` instance.
Example:
>>> registry = SkillRegistry()
>>> registry.discover("./skills")
>>> skill = registry.get("web_research")
>>> print(skill.to_prompt_section())
"""
def __init__(self) -> None:
"""Initialize the SkillRegistry with an empty internal skill store."""
self._skills: dict[str, Skill] = {}
@property
def skill_names(self) -> list[str]:
"""Return the names of all registered skills.
Returns:
A list of skill name strings in insertion order.
"""
return list(self._skills.keys())
[docs] def discover(self, *directories: str | Path) -> list[str]:
"""Recursively scan directories for SKILL.md files and register them.
Each directory is walked recursively. When a ``SKILL.md`` file is
found it is parsed and registered under its metadata name. Duplicate
skill names are skipped (first-discovered wins).
Args:
*directories: One or more directory paths to scan. Non-existent
directories are logged as warnings and skipped.
Returns:
A list of skill names that were newly registered during this
discovery pass.
"""
discovered = []
for directory in directories:
dir_path = Path(directory)
if not dir_path.is_dir():
logger.warning("Skill directory not found: %s", dir_path)
continue
for skill_file in dir_path.rglob("SKILL.md"):
try:
content = skill_file.read_text(encoding="utf-8")
skill = parse_skill_md(content, skill_file)
if skill.name not in self._skills:
self._skills[skill.name] = skill
discovered.append(skill.name)
logger.info("Discovered skill: %s at %s", skill.name, skill_file)
else:
logger.debug("Skill %s already registered, skipping %s", skill.name, skill_file)
except Exception:
logger.warning("Failed to parse skill at %s", skill_file, exc_info=True)
return discovered
[docs] def register(self, skill: Skill) -> None:
"""Manually register a pre-built :class:`Skill` instance.
Overwrites any existing skill with the same name.
Args:
skill: The :class:`Skill` to register.
"""
self._skills[skill.name] = skill
[docs] def get(self, name: str) -> Skill | None:
"""Look up a registered skill by name.
Args:
name: The skill name to look up.
Returns:
The :class:`Skill` instance, or ``None`` if not found.
"""
return self._skills.get(name)
[docs] def get_all(self) -> list[Skill]:
"""Return all registered skills.
Returns:
A list of all :class:`Skill` instances currently registered.
"""
return list(self._skills.values())
[docs] def search(self, query: str = "", tags: list[str] | None = None) -> list[Skill]:
"""Search skills by free-text query and/or tag matching.
When *query* is provided, skills whose name or description contains
the query (case-insensitive) are included. When *tags* is provided,
skills that have at least one matching tag are included. If neither
*query* nor *tags* is provided, all skills are returned.
Args:
query: A case-insensitive substring to search for in skill names
and descriptions. Defaults to ``""``.
tags: An optional list of tags to filter by. A skill matches if
it has any of the specified tags.
Returns:
A list of matching :class:`Skill` instances.
"""
results = []
query_lower = query.lower()
for skill in self._skills.values():
if query_lower and (query_lower in skill.name.lower() or query_lower in skill.metadata.description.lower()):
results.append(skill)
elif tags and any(tag in skill.metadata.tags for tag in tags):
results.append(skill)
elif not query and not tags:
results.append(skill)
return results
[docs] def validate_dependencies(self, plugin_registry: tp.Any = None) -> list[str]:
"""Validate that all registered skills have their dependencies met.
Args:
plugin_registry: Optional PluginRegistry instance to check
required_tools against.
Returns:
List of error messages (empty if all dependencies are satisfied).
"""
errors: list[str] = []
for name, skill in self._skills.items():
for dep in skill.metadata.dependencies:
if dep not in self._skills:
errors.append(f"Skill '{name}' requires missing dependency '{dep}'")
if plugin_registry is not None:
for tool_name in skill.metadata.required_tools:
if plugin_registry.get_tool(tool_name) is None:
errors.append(f"Skill '{name}' requires missing tool '{tool_name}'")
return errors
[docs] def build_skills_index(self) -> str:
"""Build a compact plain-text index of all registered skills.
The index lists each skill's name, description, and tags on a
single indented line, suitable for injection into a system prompt
so the agent knows which skills are available.
Returns:
A multi-line string listing available skills, or an empty string
if no skills are registered.
"""
if not self._skills:
return ""
lines = ["Available skills:"]
for skill in self._skills.values():
desc = skill.metadata.description or "No description"
tags = ", ".join(skill.metadata.tags) if skill.metadata.tags else ""
tag_str = f" [{tags}]" if tags else ""
lines.append(f" - {skill.name}: {desc}{tag_str}")
return "\n".join(lines)