diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py index 32746cbe1c..1cbcc7a8cb 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/python/packages/core/agent_framework/__init__.py @@ -59,7 +59,7 @@ register_state_type, ) from ._settings import SecretString, load_settings -from ._skills import FileAgentSkillsProvider +from ._skills import Skill, SkillResource, SkillsProvider from ._telemetry import ( AGENT_FRAMEWORK_USER_AGENT, APP_INFO, @@ -205,6 +205,9 @@ "AgentResponseUpdate", "AgentRunInputs", "AgentSession", + "Skill", + "SkillResource", + "SkillsProvider", "Annotation", "BaseAgent", "BaseChatClient", @@ -234,7 +237,6 @@ "Executor", "FanInEdgeGroup", "FanOutEdgeGroup", - "FileAgentSkillsProvider", "FileCheckpointStorage", "FinalT", "FinishReason", diff --git a/python/packages/core/agent_framework/_skills.py b/python/packages/core/agent_framework/_skills.py index 33d001b6f2..9e11ecbe96 100644 --- a/python/packages/core/agent_framework/_skills.py +++ b/python/packages/core/agent_framework/_skills.py @@ -1,31 +1,35 @@ # Copyright (c) Microsoft. All rights reserved. -"""File-based Agent Skills provider for the agent framework. +"""Agent Skills provider, models, and discovery utilities. -This module implements the progressive disclosure pattern from the +Defines :class:`SkillResource` and :class:`Skill`, the core data model classes +for the agent skills system, along with :class:`SkillsProvider` which implements +the progressive-disclosure pattern from the `Agent Skills specification `_: 1. **Advertise** — skill names and descriptions are injected into the system prompt. 2. **Load** — the full SKILL.md body is returned via the ``load_skill`` tool. -3. **Read resources** — supplementary files are read from disk on demand via +3. **Read resources** — supplementary content is returned on demand via the ``read_skill_resource`` tool. -Skills are discovered by searching configured directories for ``SKILL.md`` files. -Referenced resources are validated at initialization; invalid skills are excluded -and logged. +Skills can originate from two sources: -**Security:** this provider only reads static content. Skill metadata is XML-escaped -before prompt embedding, and resource reads are guarded against path traversal and -symlink escape. Only use skills from trusted sources. +- **File-based** — discovered by scanning configured directories for ``SKILL.md`` files. +- **Code-defined** — created as :class:`Skill` instances in Python code, + with optional callable resources attached via the ``@skill.resource`` decorator. + +**Security:** file-based skill metadata is XML-escaped before prompt injection, and +file-based resource reads are guarded against path traversal and symlink escape. +Only use skills from trusted sources. """ from __future__ import annotations +import inspect import logging import os import re -from collections.abc import Sequence -from dataclasses import dataclass, field +from collections.abc import Callable, Sequence from html import escape as xml_escape from pathlib import Path, PurePosixPath from typing import TYPE_CHECKING, Any, ClassVar, Final @@ -39,134 +43,590 @@ logger = logging.getLogger(__name__) +# region Models + + +class SkillResource: + """A named piece of supplementary content attached to a skill. + + .. warning:: Experimental + + This API is experimental and subject to change or removal + in future versions without notice. + + A resource provides data that an agent can retrieve on demand. It holds + either a static ``content`` string or a ``function`` that produces content + dynamically (sync or async). Exactly one must be provided. + + Attributes: + name: Resource identifier. + description: Optional human-readable summary, or ``None``. + content: Static content string, or ``None`` if backed by a callable. + function: Callable that returns content, or ``None`` if backed by static content. + + Examples: + Static resource: + + .. code-block:: python + + SkillResource(name="reference", content="Static docs here...") + + Callable resource: + + .. code-block:: python + + SkillResource(name="schema", function=get_schema_func) + """ + + def __init__( + self, + *, + name: str, + description: str | None = None, + content: str | None = None, + function: Callable[..., Any] | None = None, + ) -> None: + """Initialize a SkillResource. + + Args: + name: Identifier for this resource (e.g. ``"reference"``, ``"get-schema"``). + description: Optional human-readable summary shown when advertising the resource. + content: Static content string. Mutually exclusive with *function*. + function: Callable (sync or async) that returns content on demand. + Mutually exclusive with *content*. + """ + if not name or not name.strip(): + raise ValueError("Resource name cannot be empty.") + if content is None and function is None: + raise ValueError(f"Resource '{name}' must have either content or function.") + if content is not None and function is not None: + raise ValueError(f"Resource '{name}' must have either content or function, not both.") + + self.name = name + self.description = description + self.content = content + self.function = function + + +class Skill: + """A skill definition with optional resources. + + .. warning:: Experimental + + This API is experimental and subject to change or removal + in future versions without notice. + + A skill bundles a set of instructions (``content``) with metadata and + zero or more :class:`SkillResource` instances. Resources can be + supplied at construction time or added later via the :meth:`resource` + decorator. + + Attributes: + name: Skill name (lowercase letters, numbers, hyphens only). + description: Human-readable description of the skill. + content: The skill instructions body. + resources: Mutable list of :class:`SkillResource` instances. + path: Absolute path to the skill directory on disk, or ``None`` + for code-defined skills. + + Examples: + Direct construction: + + .. code-block:: python + + skill = Skill( + name="my-skill", + description="A skill example", + content="Use this skill for ...", + resources=[SkillResource(name="ref", content="...")], + ) + + With dynamic resources: + + .. code-block:: python + + skill = Skill( + name="db-skill", + description="Database operations", + content="Use this skill for DB tasks.", + ) + + @skill.resource + def get_schema() -> str: + return "CREATE TABLE ..." + """ + + def __init__( + self, + *, + name: str, + description: str, + content: str, + resources: list[SkillResource] | None = None, + path: str | None = None, + ) -> None: + """Initialize a Skill. + + Args: + name: Skill name (lowercase letters, numbers, hyphens only). + description: Human-readable description of the skill (≤1024 chars). + content: The skill instructions body. + resources: Pre-built resources to attach to this skill. + path: Absolute path to the skill directory on disk. Set automatically + for file-based skills; leave as ``None`` for code-defined skills. + """ + if not name or not name.strip(): + raise ValueError("Skill name cannot be empty.") + if not description or not description.strip(): + raise ValueError("Skill description cannot be empty.") + + self.name = name + self.description = description + self.content = content + self.resources: list[SkillResource] = resources if resources is not None else [] + self.path = path + + def resource( + self, + func: Callable[..., Any] | None = None, + *, + name: str | None = None, + description: str | None = None, + ) -> Any: + """Decorator that registers a callable as a resource on this skill. + + Supports bare usage (``@skill.resource``) and parameterized usage + (``@skill.resource(name="custom", description="...")``). The + decorated function is returned unchanged; a new + :class:`SkillResource` is appended to :attr:`resources`. + + Args: + func: The function being decorated. Populated automatically when + the decorator is applied without parentheses. + + Keyword Args: + name: Resource name override. Defaults to ``func.__name__``. + description: Resource description override. Defaults to the + function's docstring (via :func:`inspect.getdoc`). + + Returns: + The original function unchanged, or a secondary decorator when + called with keyword arguments. + + Examples: + Bare decorator: + + .. code-block:: python + + @skill.resource + def get_schema() -> str: + return "schema..." + + With arguments: + + .. code-block:: python + + @skill.resource(name="custom-name", description="Custom desc") + async def get_data() -> str: + return "data..." + """ + + def decorator(f: Callable[..., Any]) -> Callable[..., Any]: + resource_name = name or f.__name__ + resource_description = description or (inspect.getdoc(f) or None) + self.resources.append( + SkillResource( + name=resource_name, + description=resource_description, + function=f, + ) + ) + return f + + if func is None: + return decorator + return decorator(func) + + +# endregion + # region Constants SKILL_FILE_NAME: Final[str] = "SKILL.md" MAX_SEARCH_DEPTH: Final[int] = 2 MAX_NAME_LENGTH: Final[int] = 64 MAX_DESCRIPTION_LENGTH: Final[int] = 1024 +DEFAULT_RESOURCE_EXTENSIONS: Final[tuple[str, ...]] = ( + ".md", + ".json", + ".yaml", + ".yml", + ".csv", + ".xml", + ".txt", +) # endregion -# region Compiled regex patterns (ported from .NET FileAgentSkillLoader) +# region Patterns and prompt template # Matches YAML frontmatter delimited by "---" lines. # The \uFEFF? prefix allows an optional UTF-8 BOM. -_FRONTMATTER_RE = re.compile( +FRONTMATTER_RE = re.compile( r"\A\uFEFF?---\s*$(.+?)^---\s*$", re.MULTILINE | re.DOTALL, ) -# Matches resource file references in skill markdown. Group 1 = relative file path. -# Supports two forms: -# 1. Markdown links: [text](path/file.ext) -# 2. Backtick-quoted paths: `path/file.ext` -# Supports optional ./ or ../ prefixes; excludes URLs (no ":" in the path character class). -_RESOURCE_LINK_RE = re.compile( - r"(?:\[.*?\]\(|`)(\.?\.?/?[\w][\w\-./]*\.\w+)(?:\)|`)", -) - # Matches YAML "key: value" lines. Group 1 = key, Group 2 = quoted value, # Group 3 = unquoted value. -_YAML_KV_RE = re.compile( +YAML_KV_RE = re.compile( r"^\s*(\w+)\s*:\s*(?:[\"'](.+?)[\"']|(.+?))\s*$", re.MULTILINE, ) # Validates skill names: lowercase letters, numbers, hyphens only; # must not start or end with a hyphen. -_VALID_NAME_RE = re.compile(r"^[a-z0-9]([a-z0-9\-]*[a-z0-9])?$") +VALID_NAME_RE = re.compile(r"^[a-z0-9]([a-z0-9\-]*[a-z0-9])?$") -_DEFAULT_SKILLS_INSTRUCTION_PROMPT = """\ +# Default system prompt template for advertising available skills to the model. +# Use {skills} as the placeholder for the generated skills XML list. +DEFAULT_SKILLS_INSTRUCTION_PROMPT = """\ You have access to skills containing domain-specific knowledge and capabilities. Each skill provides specialized instructions, reference documents, and assets for specific tasks. -{0} +{skills} -When a task aligns with a skill's domain: -1. Use `load_skill` to retrieve the skill's instructions -2. Follow the provided guidance -3. Use `read_skill_resource` to read any references or other files mentioned by the skill, - always using the full path as written (e.g. `references/FAQ.md`, not just `FAQ.md`) +When a task aligns with a skill's domain, follow these steps in exact order: +1. Use `load_skill` to retrieve the skill's instructions. +2. Follow the provided guidance. +3. Use `read_skill_resource` to read any referenced resources, using the name exactly as listed + (e.g. `"style-guide"` not `"style-guide.md"`, `"references/FAQ.md"` not `"FAQ.md"`). Only load what is needed, when it is needed.""" # endregion -# region Private data classes +# region SkillsProvider + + +class SkillsProvider(BaseContextProvider): + """Context provider that advertises skills and exposes skill tools. + + .. warning:: Experimental + + This API is experimental and subject to change or removal + in future versions without notice. + + Supports both **file-based** skills (discovered from ``SKILL.md`` files) + and **code-defined** skills (passed as :class:`Skill` instances). + + Follows the progressive-disclosure pattern from the + `Agent Skills specification `_: + + 1. **Advertise** — injects skill names and descriptions into the system + prompt (~100 tokens per skill). + 2. **Load** — returns the full skill body via ``load_skill``. + 3. **Read resources** — returns supplementary content via + ``read_skill_resource``. + + **Security:** file-based metadata is XML-escaped before prompt injection, + and file-based resource reads are guarded against path traversal and + symlink escape. Only use skills from trusted sources. + + Examples: + File-based only: + + .. code-block:: python + + provider = SkillsProvider(skill_paths="./skills") + + Code-defined only: + + .. code-block:: python + + my_skill = Skill( + name="my-skill", + description="Example skill", + content="Use this skill for ...", + ) + provider = SkillsProvider(skills=[my_skill]) + + Combined: + + .. code-block:: python + + provider = SkillsProvider( + skill_paths="./skills", + skills=[my_skill], + ) + + Attributes: + DEFAULT_SOURCE_ID: Default value for the ``source_id`` used by this provider. + """ + + DEFAULT_SOURCE_ID: ClassVar[str] = "agent_skills" + + def __init__( + self, + skill_paths: str | Path | Sequence[str | Path] | None = None, + *, + skills: Sequence[Skill] | None = None, + instruction_template: str | None = None, + resource_extensions: tuple[str, ...] | None = None, + source_id: str | None = None, + ) -> None: + """Initialize a SkillsProvider. + + Args: + skill_paths: One or more directory paths to search for file-based + skills. Each path may point to an individual skill folder + (containing ``SKILL.md``) or to a parent that contains skill + subdirectories. + + Keyword Args: + skills: Code-defined :class:`Skill` instances to register. + instruction_template: Custom system-prompt template for + advertising skills. Must contain a ``{skills}`` placeholder for the + generated skills list. Uses a built-in template when ``None``. + resource_extensions: File extensions recognized as discoverable + resources. Defaults to ``DEFAULT_RESOURCE_EXTENSIONS`` + (``(".md", ".json", ".yaml", ".yml", ".csv", ".xml", ".txt")``). + source_id: Unique identifier for this provider instance. + """ + super().__init__(source_id or self.DEFAULT_SOURCE_ID) + + self._skills = _load_skills(skill_paths, skills, resource_extensions or DEFAULT_RESOURCE_EXTENSIONS) + + self._instructions = _create_instructions(instruction_template, self._skills) + + self._tools = self._create_tools() + + async def before_run( + self, + *, + agent: SupportsAgentRun, + session: AgentSession, + context: SessionContext, + state: dict[str, Any], + ) -> None: + """Inject skill instructions and tools into the session context. + + Called by the framework before the agent runs. When at least one + skill is registered, appends the skill-list system prompt and the + ``load_skill`` / ``read_skill_resource`` tools to *context*. + Args: + agent: The agent instance about to run. + session: The current agent session. + context: Session context to extend with instructions and tools. + state: Mutable per-run state dictionary (unused by this provider). + """ + if not self._skills: + return -@dataclass -class _SkillFrontmatter: - """Parsed YAML frontmatter from a SKILL.md file.""" + if self._instructions: + context.extend_instructions(self.source_id, self._instructions) + context.extend_tools(self.source_id, self._tools) - name: str - description: str + def _create_tools(self) -> list[FunctionTool]: + """Create the ``load_skill`` and ``read_skill_resource`` tool definitions. + Returns: + A two-element list of :class:`FunctionTool` instances. + """ + return [ + FunctionTool( + name="load_skill", + description="Loads the full instructions for a specific skill.", + func=self._load_skill, + input_model={ + "type": "object", + "properties": { + "skill_name": {"type": "string", "description": "The name of the skill to load."}, + }, + "required": ["skill_name"], + }, + ), + FunctionTool( + name="read_skill_resource", + description="Reads a resource associated with a skill, such as references, assets, or dynamic data.", + func=self._read_skill_resource, + input_model={ + "type": "object", + "properties": { + "skill_name": {"type": "string", "description": "The name of the skill."}, + "resource_name": { + "type": "string", + "description": "The name of the resource.", + }, + }, + "required": ["skill_name", "resource_name"], + }, + ), + ] -@dataclass -class _FileAgentSkill: - """Represents a loaded Agent Skill discovered from a filesystem directory.""" + def _load_skill(self, skill_name: str) -> str: + """Return the full instructions for the named skill. - frontmatter: _SkillFrontmatter - body: str - source_path: str - resource_names: list[str] = field(default_factory=list) + For file-based skills the raw ``SKILL.md`` content is returned as-is. + For code-defined skills the content is wrapped in XML metadata and, + when resources exist, an ```` element is appended. + + Args: + skill_name: The name of the skill to load. + + Returns: + The skill instructions text, or a user-facing error message if + *skill_name* is empty or not found. + """ + if not skill_name or not skill_name.strip(): + return "Error: Skill name cannot be empty." + + skill = self._skills.get(skill_name) + if skill is None: + return f"Error: Skill '{skill_name}' not found." + + logger.info("Loading skill: %s", skill_name) + + # File-based skills return raw content directly + if skill.path: + return skill.content + + # Code-defined skills: wrap in XML metadata + content = ( + f"{xml_escape(skill.name)}\n" + f"{xml_escape(skill.description)}\n" + "\n" + "\n" + f"{skill.content}\n" + "" + ) + + if skill.resources: + resource_lines = "\n".join(_create_resource_element(r) for r in skill.resources) + content += f"\n\n\n{resource_lines}\n" + + return content + + async def _read_skill_resource(self, skill_name: str, resource_name: str) -> str: + """Read a named resource from a skill. + + Resolves the resource by case-insensitive name lookup. Static + ``content`` is returned directly; callable resources are invoked + (awaited if async). + + Args: + skill_name: The name of the owning skill. + resource_name: The resource name to look up (case-insensitive). + + Returns: + The resource content string, or a user-facing error message on + failure. + """ + if not skill_name or not skill_name.strip(): + return "Error: Skill name cannot be empty." + + if not resource_name or not resource_name.strip(): + return "Error: Resource name cannot be empty." + + skill = self._skills.get(skill_name) + if skill is None: + return f"Error: Skill '{skill_name}' not found." + + # Find resource by name (case-insensitive) + resource_name_lower = resource_name.lower() + for resource in skill.resources: + if resource.name.lower() == resource_name_lower: + break + else: + return f"Error: Resource '{resource_name}' not found in skill '{skill_name}'." + + if resource.content is not None: + return resource.content + + if resource.function is not None: + try: + if inspect.iscoroutinefunction(resource.function): + result = await resource.function() + else: + result = resource.function() + return str(result) + except Exception as exc: + logger.exception("Failed to read resource '%s' from skill '%s'", resource_name, skill_name) + return ( + f"Error ({type(exc).__name__}): Failed to read resource" + f" '{resource_name}' from skill '{skill_name}'." + ) + + return f"Error: Resource '{resource.name}' has no content or function." # endregion -# region Private module-level functions (skill discovery, parsing, security) +# region Module-level helper functions def _normalize_resource_path(path: str) -> str: - """Normalize a relative resource path. + """Normalize a relative resource path to a canonical forward-slash form. + + Converts backslashes to forward slashes and strips leading ``./`` + prefixes so that ``./refs/doc.md`` and ``refs/doc.md`` resolve + identically. - Replaces backslashes with forward slashes and removes leading ``./`` prefixes - so that ``./refs/doc.md`` and ``refs/doc.md`` are treated as the same resource. + Args: + path: The relative path to normalize. + + Returns: + A clean forward-slash-separated path string. """ return PurePosixPath(path.replace("\\", "/")).as_posix() -def _extract_resource_paths(content: str) -> list[str]: - """Extract deduplicated resource paths from markdown link syntax.""" - seen: set[str] = set() - paths: list[str] = [] - for match in _RESOURCE_LINK_RE.finditer(content): - normalized = _normalize_resource_path(match.group(1)) - lower = normalized.lower() - if lower not in seen: - seen.add(lower) - paths.append(normalized) - return paths +def _is_path_within_directory(path: str, directory: str) -> bool: + """Return whether *path* resides under *directory*. + Comparison uses :meth:`pathlib.Path.is_relative_to`, which respects + per-platform case-sensitivity rules. -def _is_path_within_directory(full_path: str, directory_path: str) -> bool: - """Check that *full_path* is under *directory_path*. + Args: + path: Absolute path to check. + directory: Directory that must be an ancestor of *path*. - Uses :meth:`pathlib.Path.is_relative_to` for cross-platform comparison, - which handles case sensitivity correctly per platform. + Returns: + ``True`` if *path* is a descendant of *directory*. """ try: - return Path(full_path).is_relative_to(directory_path) + return Path(path).is_relative_to(directory) except (ValueError, OSError): return False -def _has_symlink_in_path(full_path: str, directory_path: str) -> bool: - """Check whether any segment in *full_path* below *directory_path* is a symlink. +def _has_symlink_in_path(path: str, directory: str) -> bool: + """Detect symlinks in the portion of *path* below *directory*. + + Only segments below *directory* are inspected; the directory itself + and anything above it are not checked. + + **Precondition:** *path* must be a descendant of *directory*. + Call :func:`_is_path_within_directory` first to verify containment. - Precondition: *full_path* must start with *directory_path*. Callers are - expected to verify containment via :func:`_is_path_within_directory` before - invoking this function. + Args: + path: Absolute path to inspect. + directory: Root directory; segments above it are not checked. + + Returns: + ``True`` if any intermediate segment below *directory* is a symlink. + + Raises: + ValueError: If *path* is not relative to *directory*. """ - dir_path = Path(directory_path) + dir_path = Path(directory) try: - relative = Path(full_path).relative_to(dir_path) + relative = Path(path).relative_to(dir_path) except ValueError as exc: - raise ValueError(f"full_path {full_path!r} does not start with directory_path {directory_path!r}") from exc + raise ValueError(f"path {path!r} does not start with directory {directory!r}") from exc current = dir_path for part in relative.parts: @@ -176,16 +636,122 @@ def _has_symlink_in_path(full_path: str, directory_path: str) -> bool: return False -def _try_parse_skill_document( +def _discover_resource_files( + skill_dir_path: str, + extensions: tuple[str, ...] = DEFAULT_RESOURCE_EXTENSIONS, +) -> list[str]: + """Scan a skill directory for resource files matching *extensions*. + + Recursively walks *skill_dir_path* and collects files whose extension + is in *extensions*, excluding ``SKILL.md`` itself. Each candidate is + validated against path-traversal and symlink-escape checks; unsafe + files are skipped with a warning. + + Args: + skill_dir_path: Absolute path to the skill directory to scan. + extensions: Tuple of allowed file extensions (e.g. ``(".md", ".json")``). + + Returns: + Relative resource paths (forward-slash-separated) for every + discovered file that passes security checks. + """ + skill_dir = Path(skill_dir_path).absolute() + root_directory_path = str(skill_dir) + resources: list[str] = [] + normalized_extensions = {e.lower() for e in extensions} + + for resource_file in skill_dir.rglob("*"): + if not resource_file.is_file(): + continue + + if resource_file.name.upper() == SKILL_FILE_NAME.upper(): + continue + + if resource_file.suffix.lower() not in normalized_extensions: + continue + + resource_full_path = str(Path(os.path.normpath(resource_file)).absolute()) + + if not _is_path_within_directory(resource_full_path, root_directory_path): + logger.warning( + "Skipping resource '%s': resolves outside skill directory '%s'", + resource_file, + skill_dir_path, + ) + continue + + if _has_symlink_in_path(resource_full_path, root_directory_path): + logger.warning( + "Skipping resource '%s': symlink detected in path under skill directory '%s'", + resource_file, + skill_dir_path, + ) + continue + + rel_path = resource_file.relative_to(skill_dir) + resources.append(_normalize_resource_path(str(rel_path))) + + return resources + + +def _validate_skill_metadata( + name: str | None, + description: str | None, + source: str, +) -> str | None: + """Validate a skill's name and description against naming rules. + + Enforces length limits, character-set restrictions, and non-emptiness + for both file-based and code-defined skills. + + Args: + name: Skill name to validate. + description: Skill description to validate. + source: Human-readable label for diagnostics (e.g. a file path + or ``"code skill"``). + + Returns: + A diagnostic error string if validation fails, or ``None`` if valid. + """ + if not name or not name.strip(): + return f"Skill from '{source}' is missing a name." + + if len(name) > MAX_NAME_LENGTH or not VALID_NAME_RE.match(name): + return ( + f"Skill from '{source}' has an invalid name '{name}': Must be {MAX_NAME_LENGTH} characters or fewer, " + "using only lowercase letters, numbers, and hyphens, and must not start or end with a hyphen." + ) + + if not description or not description.strip(): + return f"Skill '{name}' from '{source}' is missing a description." + + if len(description) > MAX_DESCRIPTION_LENGTH: + return ( + f"Skill '{name}' from '{source}' has an invalid description: " + f"Must be {MAX_DESCRIPTION_LENGTH} characters or fewer." + ) + + return None + + +def _extract_frontmatter( content: str, skill_file_path: str, -) -> tuple[_SkillFrontmatter, str] | None: - """Parse a SKILL.md file into frontmatter and body. +) -> tuple[str, str] | None: + """Extract and validate YAML frontmatter from a SKILL.md file. + + Parses the ``---``-delimited frontmatter block for ``name`` and + ``description`` fields. + + Args: + content: Raw text content of the SKILL.md file. + skill_file_path: Path to the file (used in diagnostic messages only). Returns: - A ``(frontmatter, body)`` tuple on success, or ``None`` if parsing fails. + A ``(name, description)`` tuple on success, or ``None`` if the + frontmatter is missing, malformed, or fails validation. """ - match = _FRONTMATTER_RE.search(content) + match = FRONTMATTER_RE.search(content) if not match: logger.error("SKILL.md at '%s' does not contain valid YAML frontmatter delimited by '---'", skill_file_path) return None @@ -194,7 +760,7 @@ def _try_parse_skill_document( name: str | None = None description: str | None = None - for kv_match in _YAML_KV_RE.finditer(yaml_content): + for kv_match in YAML_KV_RE.finditer(yaml_content): key = kv_match.group(1) value = kv_match.group(2) if kv_match.group(2) is not None else kv_match.group(3) @@ -203,75 +769,28 @@ def _try_parse_skill_document( elif key.lower() == "description": description = value - if not name or not name.strip(): - logger.error("SKILL.md at '%s' is missing a 'name' field in frontmatter", skill_file_path) + error = _validate_skill_metadata(name, description, skill_file_path) + if error: + logger.error(error) return None - if len(name) > MAX_NAME_LENGTH or not _VALID_NAME_RE.match(name): - logger.error( - "SKILL.md at '%s' has an invalid 'name' value: Must be %d characters or fewer, " - "using only lowercase letters, numbers, and hyphens, and must not start or end with a hyphen.", - skill_file_path, - MAX_NAME_LENGTH, - ) - return None - - if not description or not description.strip(): - logger.error("SKILL.md at '%s' is missing a 'description' field in frontmatter", skill_file_path) - return None - - if len(description) > MAX_DESCRIPTION_LENGTH: - logger.error( - "SKILL.md at '%s' has an invalid 'description' value: Must be %d characters or fewer.", - skill_file_path, - MAX_DESCRIPTION_LENGTH, - ) - return None + # name and description are guaranteed non-None after validation + return name, description # type: ignore[return-value] - body = content[match.end() :].lstrip() - return _SkillFrontmatter(name, description), body - -def _validate_resources( +def _read_and_parse_skill_file( skill_dir_path: str, - resource_names: list[str], - skill_name: str, -) -> bool: - """Validate that all resource paths exist and are safe.""" - skill_dir = Path(skill_dir_path).absolute() - - for resource_name in resource_names: - resource_path = Path(os.path.normpath(skill_dir / resource_name)) - - if not _is_path_within_directory(str(resource_path), str(skill_dir)): - logger.warning( - "Excluding skill '%s': resource '%s' references a path outside the skill directory", - skill_name, - resource_name, - ) - return False - - if not resource_path.is_file(): - logger.warning( - "Excluding skill '%s': referenced resource '%s' does not exist", - skill_name, - resource_name, - ) - return False - - if _has_symlink_in_path(str(resource_path), str(skill_dir)): - logger.warning( - "Excluding skill '%s': resource '%s' is a symlink that resolves outside the skill directory", - skill_name, - resource_name, - ) - return False - - return True +) -> tuple[str, str, str] | None: + """Read and parse the SKILL.md file in *skill_dir_path*. + Args: + skill_dir_path: Absolute path to the directory containing ``SKILL.md``. -def _parse_skill_file(skill_dir_path: str) -> _FileAgentSkill | None: - """Parse a SKILL.md file from the given directory.""" + Returns: + A ``(name, description, content)`` tuple where *content* is the + full raw file text, or ``None`` if the file cannot be read or + its frontmatter is invalid. + """ skill_file = Path(skill_dir_path) / SKILL_FILE_NAME try: @@ -280,321 +799,265 @@ def _parse_skill_file(skill_dir_path: str) -> _FileAgentSkill | None: logger.error("Failed to read SKILL.md at '%s'", skill_file) return None - result = _try_parse_skill_document(content, str(skill_file)) + result = _extract_frontmatter(content, str(skill_file)) if result is None: return None - frontmatter, body = result - resource_names = _extract_resource_paths(body) + name, description = result + return name, description, content - if not _validate_resources(skill_dir_path, resource_names, frontmatter.name): - return None - return _FileAgentSkill( - frontmatter=frontmatter, - body=body, - source_path=skill_dir_path, - resource_names=resource_names, - ) - - -def _search_directories_for_skills( - directory: str, - results: list[str], - current_depth: int, -) -> None: - """Recursively search for SKILL.md files up to *MAX_SEARCH_DEPTH*.""" - dir_path = Path(directory) - if (dir_path / SKILL_FILE_NAME).is_file(): - results.append(str(dir_path.absolute())) - - if current_depth >= MAX_SEARCH_DEPTH: - return - - try: - entries = list(dir_path.iterdir()) - except OSError: - return +def _discover_skill_directories(skill_paths: Sequence[str]) -> list[str]: + """Return absolute paths of all directories that contain a ``SKILL.md`` file. - for entry in entries: - if entry.is_dir(): - _search_directories_for_skills(str(entry), results, current_depth + 1) + Recursively searches each root path up to :data:`MAX_SEARCH_DEPTH`. + Args: + skill_paths: Root directory paths to search. -def _discover_skill_directories(skill_paths: Sequence[str]) -> list[str]: - """Discover all directories containing SKILL.md files.""" + Returns: + Absolute paths to directories containing ``SKILL.md``. + """ discovered: list[str] = [] - for root_dir in skill_paths: - if not root_dir or not root_dir.strip() or not Path(root_dir).is_dir(): - continue - _search_directories_for_skills(root_dir, discovered, current_depth=0) - return discovered + def _search(directory: str, current_depth: int) -> None: + dir_path = Path(directory) + if (dir_path / SKILL_FILE_NAME).is_file(): + discovered.append(str(dir_path.absolute())) -def _discover_and_load_skills(skill_paths: Sequence[str]) -> dict[str, _FileAgentSkill]: - """Discover and load all valid skills from the given paths.""" - skills: dict[str, _FileAgentSkill] = {} + if current_depth >= MAX_SEARCH_DEPTH: + return - discovered = _discover_skill_directories(skill_paths) - logger.info("Discovered %d potential skills", len(discovered)) + try: + entries = list(dir_path.iterdir()) + except OSError: + return - for skill_path in discovered: - skill = _parse_skill_file(skill_path) - if skill is None: - continue + for entry in entries: + if entry.is_dir(): + _search(str(entry), current_depth + 1) - if skill.frontmatter.name in skills: - existing = skills[skill.frontmatter.name] - logger.warning( - "Duplicate skill name '%s': skill from '%s' skipped in favor of existing skill from '%s'", - skill.frontmatter.name, - skill_path, - existing.source_path, - ) + for root_dir in skill_paths: + if not root_dir or not root_dir.strip() or not Path(root_dir).is_dir(): continue + _search(root_dir, current_depth=0) - skills[skill.frontmatter.name] = skill - logger.info("Loaded skill: %s", skill.frontmatter.name) + return discovered - logger.info("Successfully loaded %d skills", len(skills)) - return skills +def _read_file_skill_resource(skill: Skill, resource_name: str) -> str: + """Read a file-based resource from disk with security guards. -def _read_skill_resource(skill: _FileAgentSkill, resource_name: str) -> str: - """Read a resource file from disk with path traversal and symlink guards. + Validates that the resolved path stays within the skill directory and + does not traverse any symlinks before reading. Args: - skill: The skill that owns the resource. + skill: The owning skill (must have a non-``None`` :attr:`~Skill.path`). resource_name: Relative path of the resource within the skill directory. Returns: The UTF-8 text content of the resource file. Raises: - ValueError: The resource is not registered, resolves outside the skill - directory, or does not exist. + ValueError: If the resolved path escapes the skill directory, + the file does not exist, or a symlink is detected in the path. """ resource_name = _normalize_resource_path(resource_name) - # Find the registered resource name with the original casing so the - # file path is correct on case-sensitive filesystems. - registered_name: str | None = None - for r in skill.resource_names: - if r.lower() == resource_name.lower(): - registered_name = r - break + if not skill.path: + raise ValueError(f"Skill '{skill.name}' has no path set; cannot read file-based resources.") - if registered_name is None: - raise ValueError(f"Resource '{resource_name}' not found in skill '{skill.frontmatter.name}'.") + resource_full_path = os.path.normpath(Path(skill.path) / resource_name) + root_directory_path = os.path.normpath(skill.path) - full_path = os.path.normpath(Path(skill.source_path) / registered_name) - source_dir = str(Path(skill.source_path).absolute()) - - if not _is_path_within_directory(full_path, source_dir): + if not _is_path_within_directory(resource_full_path, root_directory_path): raise ValueError(f"Resource file '{resource_name}' references a path outside the skill directory.") - if not Path(full_path).is_file(): - raise ValueError(f"Resource file '{resource_name}' not found in skill '{skill.frontmatter.name}'.") + if not Path(resource_full_path).is_file(): + raise ValueError(f"Resource file '{resource_name}' not found in skill '{skill.name}'.") - if _has_symlink_in_path(full_path, source_dir): - raise ValueError(f"Resource file '{resource_name}' is a symlink that resolves outside the skill directory.") + if _has_symlink_in_path(resource_full_path, root_directory_path): + raise ValueError( + f"Resource file '{resource_name}' in skill '{skill.name}' " + "has a symlink in its path; symlinks are not allowed." + ) - logger.info("Reading resource '%s' from skill '%s'", resource_name, skill.frontmatter.name) - return Path(full_path).read_text(encoding="utf-8") + logger.info("Reading resource '%s' from skill '%s'", resource_name, skill.name) + return Path(resource_full_path).read_text(encoding="utf-8") -def _build_skills_instruction_prompt( - prompt_template: str | None, - skills: dict[str, _FileAgentSkill], -) -> str | None: - """Build the system prompt advertising available skills.""" - template = _DEFAULT_SKILLS_INSTRUCTION_PROMPT +def _discover_file_skills( + skill_paths: str | Path | Sequence[str | Path] | None, + resource_extensions: tuple[str, ...] = DEFAULT_RESOURCE_EXTENSIONS, +) -> dict[str, Skill]: + """Discover, parse, and load all file-based skills from the given paths. - if prompt_template is not None: - # Validate that the custom template contains a valid {0} placeholder - try: - prompt_template.format("") - template = prompt_template - except (KeyError, IndexError) as exc: - raise ValueError( - "The provided skills_instruction_prompt is not a valid format string. " - "It must contain a '{0}' placeholder and escape any literal '{' or '}' " - "by doubling them ('{{' or '}}')." - ) from exc + Each discovered ``SKILL.md`` is parsed for metadata, and resource files + in the same directory are wrapped in lazy-read closures that perform + security checks (path traversal, symlink escape) at read time. - if not skills: - return None + Args: + skill_paths: Directory path(s) to scan, or ``None`` to skip. + resource_extensions: File extensions recognized as resources. - lines: list[str] = [] - # Sort by name for deterministic output - for skill in sorted(skills.values(), key=lambda s: s.frontmatter.name): - lines.append(" ") - lines.append(f" {xml_escape(skill.frontmatter.name)}") - lines.append(f" {xml_escape(skill.frontmatter.description)}") - lines.append(" ") + Returns: + A dict mapping skill name → :class:`Skill`. + """ + if skill_paths is None: + return {} - return template.format("\n".join(lines)) + resolved_paths: list[str] = ( + [str(skill_paths)] if isinstance(skill_paths, (str, Path)) else [str(p) for p in skill_paths] + ) + skills: dict[str, Skill] = {} -# endregion + discovered = _discover_skill_directories(resolved_paths) + logger.info("Discovered %d potential skills", len(discovered)) -# region Public API + for skill_path in discovered: + parsed = _read_and_parse_skill_file(skill_path) + if parsed is None: + continue + name, description, content = parsed -class FileAgentSkillsProvider(BaseContextProvider): - """A context provider that discovers and exposes Agent Skills from filesystem directories. + if name in skills: + logger.warning( + "Duplicate skill name '%s': skill from '%s' skipped in favor of existing skill", + name, + skill_path, + ) + continue - This provider implements the progressive disclosure pattern from the - `Agent Skills specification `_: + file_skill = Skill( + name=name, + description=description, + content=content, + path=skill_path, + ) - 1. **Advertise** — skill names and descriptions are injected into the system prompt - (~100 tokens per skill). - 2. **Load** — the full SKILL.md body is returned via the ``load_skill`` tool. - 3. **Read resources** — supplementary files are read on demand via the - ``read_skill_resource`` tool. + # Discover and attach file-based resources as SkillResource closures + for rn in _discover_resource_files(skill_path, resource_extensions): + reader = (lambda s, r: lambda: _read_file_skill_resource(s, r))(file_skill, rn) + file_skill.resources.append(SkillResource(name=rn, function=reader)) - Skills are discovered by searching the configured directories for ``SKILL.md`` files. - Referenced resources are validated at initialization; invalid skills are excluded and - logged. + skills[file_skill.name] = file_skill + logger.info("Loaded skill: %s", file_skill.name) - **Security:** this provider only reads static content. Skill metadata is XML-escaped - before prompt embedding, and resource reads are guarded against path traversal and - symlink escape. Only use skills from trusted sources. + logger.info("Successfully loaded %d skills", len(skills)) + return skills - Args: - skill_paths: A single path or sequence of paths to search. Each can be an - individual skill folder (containing a SKILL.md file) or a parent folder - with skill subdirectories. - - Keyword Args: - skills_instruction_prompt: A custom system prompt template for advertising - skills. Use ``{0}`` as the placeholder for the generated skills list. - When ``None``, a default template is used. - source_id: Unique identifier for this provider instance. - logger: Optional logger instance. When ``None``, uses the module logger. - """ - DEFAULT_SOURCE_ID: ClassVar[str] = "file_agent_skills" +def _load_skills( + skill_paths: str | Path | Sequence[str | Path] | None, + skills: Sequence[Skill] | None, + resource_extensions: tuple[str, ...], +) -> dict[str, Skill]: + """Discover and merge skills from file paths and code-defined skills. - def __init__( - self, - skill_paths: str | Path | Sequence[str | Path], - *, - skills_instruction_prompt: str | None = None, - source_id: str | None = None, - ) -> None: - """Initialize the FileAgentSkillsProvider. + File-based skills are discovered first. Code-defined skills are then + merged in; if a code-defined skill has the same name as an existing + file-based skill, the code-defined one is skipped with a warning. - Args: - skill_paths: A single path or sequence of paths to search for skills. + Args: + skill_paths: Directory path(s) to scan for ``SKILL.md`` files, or ``None``. + skills: Code-defined :class:`Skill` instances, or ``None``. + resource_extensions: File extensions recognized as discoverable resources. - Keyword Args: - skills_instruction_prompt: Custom system prompt template with ``{0}`` placeholder. - source_id: Unique identifier for this provider instance. - """ - super().__init__(source_id or self.DEFAULT_SOURCE_ID) + Returns: + A dict mapping skill name → :class:`Skill`. + """ + result = _discover_file_skills(skill_paths, resource_extensions) - resolved_paths: Sequence[str] = ( - [str(skill_paths)] if isinstance(skill_paths, (str, Path)) else [str(p) for p in skill_paths] - ) + if skills: + for code_skill in skills: + error = _validate_skill_metadata( + code_skill.name, code_skill.description, "code skill" + ) + if error: + logger.warning(error) + continue + if code_skill.name in result: + logger.warning( + "Duplicate skill name '%s': code skill skipped in favor of existing skill", + code_skill.name, + ) + continue + result[code_skill.name] = code_skill + logger.info("Registered code skill: %s", code_skill.name) - self._skills = _discover_and_load_skills(resolved_paths) - self._skills_instruction_prompt = _build_skills_instruction_prompt(skills_instruction_prompt, self._skills) - self._tools = [ - FunctionTool( - name="load_skill", - description="Loads the full instructions for a specific skill.", - func=self._load_skill, - input_model={ - "type": "object", - "properties": { - "skill_name": {"type": "string", "description": "The name of the skill to load."}, - }, - "required": ["skill_name"], - }, - ), - FunctionTool( - name="read_skill_resource", - description="Reads a file associated with a skill, such as references or assets.", - func=self._read_skill_resource, - input_model={ - "type": "object", - "properties": { - "skill_name": {"type": "string", "description": "The name of the skill."}, - "resource_name": { - "type": "string", - "description": "The relative path of the resource file.", - }, - }, - "required": ["skill_name", "resource_name"], - }, - ), - ] + return result - async def before_run( - self, - *, - agent: SupportsAgentRun, - session: AgentSession, - context: SessionContext, - state: dict[str, Any], - ) -> None: - """Inject skill instructions and tools into the session context. - When skills are available, adds the skills instruction prompt and - ``load_skill`` / ``read_skill_resource`` tools. - """ - if not self._skills: - return +def _create_resource_element(resource: SkillResource) -> str: + """Create a self-closing ```` XML element from an :class:`SkillResource`. - if self._skills_instruction_prompt: - context.extend_instructions(self.source_id, self._skills_instruction_prompt) - context.extend_tools(self.source_id, self._tools) + Args: + resource: The resource to create the element from. - def _load_skill(self, skill_name: str) -> str: - """Load the full instructions for a specific skill. + Returns: + A single indented XML element string with ``name`` and optional + ``description`` attributes. + """ + attrs = f'name="{xml_escape(resource.name, quote=True)}"' + if resource.description: + attrs += f' description="{xml_escape(resource.description, quote=True)}"' + return f" " - Args: - skill_name: The name of the skill to load. - Returns: - The skill body text, or an error message if not found. - """ - if not skill_name or not skill_name.strip(): - return "Error: Skill name cannot be empty." +def _create_instructions( + prompt_template: str | None, + skills: dict[str, Skill], +) -> str | None: + """Create the system-prompt text that advertises available skills. - skill = self._skills.get(skill_name) - if skill is None: - return f"Error: Skill '{skill_name}' not found." + Generates an XML list of ```` elements (sorted by name) and + inserts it into *prompt_template* at the ``{skills}`` placeholder. - logger.info("Loading skill: %s", skill_name) - return skill.body + Args: + prompt_template: Custom template string with a ``{skills}`` placeholder, + or ``None`` to use the built-in default. + skills: Registered skills keyed by name. - def _read_skill_resource(self, skill_name: str, resource_name: str) -> str: - """Read a file associated with a skill. + Returns: + The formatted instruction string, or ``None`` when *skills* is empty. - Args: - skill_name: The name of the skill. - resource_name: The relative path of the resource file. + Raises: + ValueError: If *prompt_template* is not a valid format string + (e.g. missing ``{skills}`` placeholder). + """ + template = DEFAULT_SKILLS_INSTRUCTION_PROMPT - Returns: - The resource file content, or an error message if not found. - """ - if not skill_name or not skill_name.strip(): - return "Error: Skill name cannot be empty." + if prompt_template is not None: + # Validate that the custom template contains a valid {skills} placeholder + try: + result = prompt_template.format(skills="__PROBE__") + except (KeyError, IndexError, ValueError) as exc: + raise ValueError( + "The provided instruction_template is not a valid format string. " + "It must contain a '{skills}' placeholder and escape any literal" # noqa: RUF027 + " '{' or '}' " + "by doubling them ('{{' or '}}')." + ) from exc + if "__PROBE__" not in result: + raise ValueError( + "The provided instruction_template must contain a '{skills}' placeholder." # noqa: RUF027 + ) + template = prompt_template - if not resource_name or not resource_name.strip(): - return "Error: Resource name cannot be empty." + if not skills: + return None - skill = self._skills.get(skill_name) - if skill is None: - return f"Error: Skill '{skill_name}' not found." + lines: list[str] = [] + # Sort by name for deterministic output + for skill in sorted(skills.values(), key=lambda s: s.name): + lines.append(" ") + lines.append(f" {xml_escape(skill.name)}") + lines.append(f" {xml_escape(skill.description)}") + lines.append(" ") - try: - return _read_skill_resource(skill, resource_name) - except Exception: - logger.exception("Failed to read resource '%s' from skill '%s'", resource_name, skill_name) - return f"Error: Failed to read resource '{resource_name}' from skill '{skill_name}'." + return template.format(skills="\n".join(lines)) # endregion diff --git a/python/packages/core/tests/core/test_skills.py b/python/packages/core/tests/core/test_skills.py index a77f214718..c572f4727b 100644 --- a/python/packages/core/tests/core/test_skills.py +++ b/python/packages/core/tests/core/test_skills.py @@ -1,6 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. -"""Tests for file-based Agent Skills provider.""" +"""Tests for Agent Skills provider (file-based and code-defined).""" from __future__ import annotations @@ -10,17 +10,21 @@ import pytest -from agent_framework import FileAgentSkillsProvider, SessionContext +from agent_framework import Skill, SkillResource, SkillsProvider, SessionContext from agent_framework._skills import ( - _build_skills_instruction_prompt, - _discover_and_load_skills, - _extract_resource_paths, - _FileAgentSkill, + DEFAULT_RESOURCE_EXTENSIONS, + _create_instructions, + _create_resource_element, + _discover_file_skills, + _discover_resource_files, + _discover_skill_directories, + _extract_frontmatter, _has_symlink_in_path, + _is_path_within_directory, _normalize_resource_path, - _read_skill_resource, - _SkillFrontmatter, - _try_parse_skill_document, + _read_and_parse_skill_file, + _read_file_skill_resource, + _validate_skill_metadata, ) @@ -70,6 +74,19 @@ def _write_skill( return skill_dir +def _read_and_parse_skill_file_for_test(skill_dir: Path) -> Skill: + """Parse a SKILL.md file from the given directory, raising if invalid.""" + result = _read_and_parse_skill_file(str(skill_dir)) + assert result is not None, f"Failed to parse skill at {skill_dir}" + name, description, content = result + return Skill( + name=name, + description=description, + content=content, + path=str(skill_dir), + ) + + # --------------------------------------------------------------------------- # Tests: module-level helper functions # --------------------------------------------------------------------------- @@ -91,115 +108,150 @@ def test_no_change_for_clean_path(self) -> None: assert _normalize_resource_path("refs/doc.md") == "refs/doc.md" -class TestExtractResourcePaths: - """Tests for _extract_resource_paths.""" - - def test_extracts_markdown_links(self) -> None: - content = "See [doc](refs/FAQ.md) and [template](assets/template.md)." - paths = _extract_resource_paths(content) - assert paths == ["refs/FAQ.md", "assets/template.md"] +class TestDiscoverResourceFiles: + """Tests for _discover_resource_files (filesystem-based resource discovery).""" - def test_deduplicates_case_insensitive(self) -> None: - content = "See [a](refs/FAQ.md) and [b](refs/faq.md)." - paths = _extract_resource_paths(content) - assert len(paths) == 1 + def test_discovers_md_files(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text("---\nname: s\ndescription: d\n---\n", encoding="utf-8") + refs = skill_dir / "refs" + refs.mkdir() + (refs / "FAQ.md").write_text("FAQ content", encoding="utf-8") + resources = _discover_resource_files(str(skill_dir)) + assert "refs/FAQ.md" in resources + + def test_excludes_skill_md(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text("content", encoding="utf-8") + resources = _discover_resource_files(str(skill_dir)) + assert len(resources) == 0 - def test_normalizes_dot_slash_prefix(self) -> None: - content = "See [doc](./refs/FAQ.md)." - paths = _extract_resource_paths(content) - assert paths == ["refs/FAQ.md"] + def test_discovers_multiple_extensions(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + (skill_dir / "data.json").write_text("{}", encoding="utf-8") + (skill_dir / "config.yaml").write_text("key: val", encoding="utf-8") + (skill_dir / "notes.txt").write_text("notes", encoding="utf-8") + resources = _discover_resource_files(str(skill_dir)) + assert len(resources) == 3 + names = set(resources) + assert "data.json" in names + assert "config.yaml" in names + assert "notes.txt" in names + + def test_ignores_unsupported_extensions(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + (skill_dir / "image.png").write_bytes(b"\x89PNG") + (skill_dir / "binary.exe").write_bytes(b"\x00") + resources = _discover_resource_files(str(skill_dir)) + assert len(resources) == 0 - def test_ignores_urls(self) -> None: - content = "See [link](https://example.com/doc.md)." - paths = _extract_resource_paths(content) - assert paths == [] + def test_custom_extensions(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + (skill_dir / "data.json").write_text("{}", encoding="utf-8") + (skill_dir / "notes.txt").write_text("notes", encoding="utf-8") + resources = _discover_resource_files(str(skill_dir), extensions=(".json",)) + assert resources == ["data.json"] - def test_empty_content(self) -> None: - assert _extract_resource_paths("") == [] + def test_discovers_nested_files(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "my-skill" + sub = skill_dir / "refs" / "deep" + sub.mkdir(parents=True) + (sub / "doc.md").write_text("deep doc", encoding="utf-8") + resources = _discover_resource_files(str(skill_dir)) + assert "refs/deep/doc.md" in resources - def test_extracts_backtick_quoted_paths(self) -> None: - content = "Use the template at `assets/template.md` and the script `./scripts/run.py`." - paths = _extract_resource_paths(content) - assert paths == ["assets/template.md", "scripts/run.py"] + def test_empty_directory(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + resources = _discover_resource_files(str(skill_dir)) + assert resources == [] - def test_deduplicates_across_link_and_backtick(self) -> None: - content = "See [doc](refs/FAQ.md) and also `refs/FAQ.md`." - paths = _extract_resource_paths(content) - assert len(paths) == 1 + def test_default_extensions_match_constant(self) -> None: + assert ".md" in DEFAULT_RESOURCE_EXTENSIONS + assert ".json" in DEFAULT_RESOURCE_EXTENSIONS + assert ".yaml" in DEFAULT_RESOURCE_EXTENSIONS + assert ".yml" in DEFAULT_RESOURCE_EXTENSIONS + assert ".csv" in DEFAULT_RESOURCE_EXTENSIONS + assert ".xml" in DEFAULT_RESOURCE_EXTENSIONS + assert ".txt" in DEFAULT_RESOURCE_EXTENSIONS class TestTryParseSkillDocument: - """Tests for _try_parse_skill_document.""" + """Tests for _extract_frontmatter.""" def test_valid_skill(self) -> None: content = "---\nname: test-skill\ndescription: A test skill.\n---\n# Body\nInstructions here." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is not None - frontmatter, body = result - assert frontmatter.name == "test-skill" - assert frontmatter.description == "A test skill." - assert "Instructions here." in body + name, description = result + assert name == "test-skill" + assert description == "A test skill." def test_quoted_values(self) -> None: content = "---\nname: \"test-skill\"\ndescription: 'A test skill.'\n---\nBody." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is not None - assert result[0].name == "test-skill" - assert result[0].description == "A test skill." + assert result[0] == "test-skill" + assert result[1] == "A test skill." def test_utf8_bom(self) -> None: content = "\ufeff---\nname: test-skill\ndescription: A test skill.\n---\nBody." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is not None - assert result[0].name == "test-skill" + assert result[0] == "test-skill" def test_missing_frontmatter(self) -> None: content = "# Just a markdown file\nNo frontmatter here." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is None def test_missing_name(self) -> None: content = "---\ndescription: A test skill.\n---\nBody." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is None def test_missing_description(self) -> None: content = "---\nname: test-skill\n---\nBody." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is None def test_invalid_name_uppercase(self) -> None: content = "---\nname: Test-Skill\ndescription: A test skill.\n---\nBody." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is None def test_invalid_name_starts_with_hyphen(self) -> None: content = "---\nname: -test-skill\ndescription: A test skill.\n---\nBody." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is None def test_invalid_name_ends_with_hyphen(self) -> None: content = "---\nname: test-skill-\ndescription: A test skill.\n---\nBody." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is None def test_name_too_long(self) -> None: long_name = "a" * 65 content = f"---\nname: {long_name}\ndescription: A test skill.\n---\nBody." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is None def test_description_too_long(self) -> None: long_desc = "a" * 1025 content = f"---\nname: test-skill\ndescription: {long_desc}\n---\nBody." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is None def test_extra_metadata_ignored(self) -> None: content = "---\nname: test-skill\ndescription: A test skill.\nauthor: someone\nversion: 1.0\n---\nBody." - result = _try_parse_skill_document(content, "test.md") + result = _extract_frontmatter(content, "test.md") assert result is not None - assert result[0].name == "test-skill" + assert result[0] == "test-skill" # --------------------------------------------------------------------------- @@ -208,19 +260,19 @@ def test_extra_metadata_ignored(self) -> None: class TestDiscoverAndLoadSkills: - """Tests for _discover_and_load_skills.""" + """Tests for _discover_file_skills.""" def test_discovers_valid_skill(self, tmp_path: Path) -> None: _write_skill(tmp_path, "my-skill") - skills = _discover_and_load_skills([str(tmp_path)]) + skills = _discover_file_skills([str(tmp_path)]) assert "my-skill" in skills - assert skills["my-skill"].frontmatter.name == "my-skill" + assert skills["my-skill"].name == "my-skill" def test_discovers_nested_skills(self, tmp_path: Path) -> None: skills_dir = tmp_path / "skills" _write_skill(skills_dir, "skill-a") _write_skill(skills_dir, "skill-b") - skills = _discover_and_load_skills([str(skills_dir)]) + skills = _discover_file_skills([str(skills_dir)]) assert len(skills) == 2 assert "skill-a" in skills assert "skill-b" in skills @@ -229,7 +281,7 @@ def test_skips_invalid_skill(self, tmp_path: Path) -> None: skill_dir = tmp_path / "bad-skill" skill_dir.mkdir() (skill_dir / "SKILL.md").write_text("No frontmatter here.", encoding="utf-8") - skills = _discover_and_load_skills([str(tmp_path)]) + skills = _discover_file_skills([str(tmp_path)]) assert len(skills) == 0 def test_deduplicates_skill_names(self, tmp_path: Path) -> None: @@ -237,16 +289,16 @@ def test_deduplicates_skill_names(self, tmp_path: Path) -> None: dir2 = tmp_path / "dir2" _write_skill(dir1, "my-skill", body="First") _write_skill(dir2, "my-skill", body="Second") - skills = _discover_and_load_skills([str(dir1), str(dir2)]) + skills = _discover_file_skills([str(dir1), str(dir2)]) assert len(skills) == 1 - assert skills["my-skill"].body == "First" + assert "First" in skills["my-skill"].content def test_empty_directory(self, tmp_path: Path) -> None: - skills = _discover_and_load_skills([str(tmp_path)]) + skills = _discover_file_skills([str(tmp_path)]) assert len(skills) == 0 def test_nonexistent_directory(self) -> None: - skills = _discover_and_load_skills(["/nonexistent/path"]) + skills = _discover_file_skills(["/nonexistent/path"]) assert len(skills) == 0 def test_multiple_paths(self, tmp_path: Path) -> None: @@ -254,7 +306,7 @@ def test_multiple_paths(self, tmp_path: Path) -> None: dir2 = tmp_path / "dir2" _write_skill(dir1, "skill-a") _write_skill(dir2, "skill-b") - skills = _discover_and_load_skills([str(dir1), str(dir2)]) + skills = _discover_file_skills([str(dir1), str(dir2)]) assert len(skills) == 2 def test_depth_limit(self, tmp_path: Path) -> None: @@ -265,40 +317,33 @@ def test_depth_limit(self, tmp_path: Path) -> None: deep = tmp_path / "level1" / "level2" / "level3" deep.mkdir(parents=True) (deep / "SKILL.md").write_text("---\nname: deep-skill\ndescription: Too deep.\n---\nBody.", encoding="utf-8") - skills = _discover_and_load_skills([str(tmp_path)]) + skills = _discover_file_skills([str(tmp_path)]) assert "deep-skill" not in skills def test_skill_with_resources(self, tmp_path: Path) -> None: _write_skill( tmp_path, "my-skill", - body="See [doc](refs/FAQ.md).", + body="Instructions here.", resources={"refs/FAQ.md": "FAQ content"}, ) - skills = _discover_and_load_skills([str(tmp_path)]) + skills = _discover_file_skills([str(tmp_path)]) assert "my-skill" in skills - assert skills["my-skill"].resource_names == ["refs/FAQ.md"] - - def test_excludes_skill_with_missing_resource(self, tmp_path: Path) -> None: - _write_skill( - tmp_path, - "my-skill", - body="See [doc](refs/MISSING.md).", - ) - skills = _discover_and_load_skills([str(tmp_path)]) - assert len(skills) == 0 + assert [r.name for r in skills["my-skill"].resources] == ["refs/FAQ.md"] - def test_excludes_skill_with_path_traversal_resource(self, tmp_path: Path) -> None: + def test_skill_discovers_all_resource_files(self, tmp_path: Path) -> None: + """Resources are discovered by filesystem scan, not by markdown links.""" _write_skill( tmp_path, "my-skill", - body="See [doc](../secret.md).", - resources={}, # resource points outside + body="No links here.", + resources={"data.json": '{"key": "val"}', "refs/doc.md": "doc content"}, ) - # Create the file outside the skill directory - (tmp_path / "secret.md").write_text("secret", encoding="utf-8") - skills = _discover_and_load_skills([str(tmp_path)]) - assert len(skills) == 0 + skills = _discover_file_skills([str(tmp_path)]) + assert "my-skill" in skills + resource_names = sorted(r.name for r in skills["my-skill"].resources) + assert "data.json" in resource_names + assert "refs/doc.md" in resource_names # --------------------------------------------------------------------------- @@ -307,7 +352,7 @@ def test_excludes_skill_with_path_traversal_resource(self, tmp_path: Path) -> No class TestReadSkillResource: - """Tests for _read_skill_resource.""" + """Tests for _read_file_skill_resource.""" def test_reads_valid_resource(self, tmp_path: Path) -> None: _write_skill( @@ -316,8 +361,8 @@ def test_reads_valid_resource(self, tmp_path: Path) -> None: body="See [doc](refs/FAQ.md).", resources={"refs/FAQ.md": "FAQ content here"}, ) - skills = _discover_and_load_skills([str(tmp_path)]) - content = _read_skill_resource(skills["my-skill"], "refs/FAQ.md") + file_skill = _read_and_parse_skill_file_for_test(tmp_path / "my-skill") + content = _read_file_skill_resource(file_skill, "refs/FAQ.md") assert content == "FAQ content here" def test_normalizes_dot_slash(self, tmp_path: Path) -> None: @@ -327,74 +372,70 @@ def test_normalizes_dot_slash(self, tmp_path: Path) -> None: body="See [doc](refs/FAQ.md).", resources={"refs/FAQ.md": "FAQ content"}, ) - skills = _discover_and_load_skills([str(tmp_path)]) - content = _read_skill_resource(skills["my-skill"], "./refs/FAQ.md") + file_skill = _read_and_parse_skill_file_for_test(tmp_path / "my-skill") + content = _read_file_skill_resource(file_skill, "./refs/FAQ.md") assert content == "FAQ content" def test_unregistered_resource_raises(self, tmp_path: Path) -> None: _write_skill(tmp_path, "my-skill") - skills = _discover_and_load_skills([str(tmp_path)]) + file_skill = _read_and_parse_skill_file_for_test(tmp_path / "my-skill") with pytest.raises(ValueError, match="not found in skill"): - _read_skill_resource(skills["my-skill"], "nonexistent.md") + _read_file_skill_resource(file_skill, "nonexistent.md") - def test_case_insensitive_lookup_uses_registered_casing(self, tmp_path: Path) -> None: + def test_reads_resource_with_exact_casing(self, tmp_path: Path) -> None: + """Direct file read uses the given resource name for path resolution.""" _write_skill( tmp_path, "my-skill", body="See [doc](refs/FAQ.md).", resources={"refs/FAQ.md": "FAQ content"}, ) - skills = _discover_and_load_skills([str(tmp_path)]) - # Request with different casing; the registered name should be used for the file path - content = _read_skill_resource(skills["my-skill"], "REFS/faq.md") + file_skill = _read_and_parse_skill_file_for_test(tmp_path / "my-skill") + content = _read_file_skill_resource(file_skill, "refs/FAQ.md") assert content == "FAQ content" def test_path_traversal_raises(self, tmp_path: Path) -> None: - skill = _FileAgentSkill( - frontmatter=_SkillFrontmatter("test", "Test skill"), - body="Body", - source_path=str(tmp_path / "skill"), - resource_names=["../secret.md"], + skill = Skill( + name="test", + description="Test skill", + content="Body", + path=str(tmp_path / "skill"), ) (tmp_path / "secret.md").write_text("secret", encoding="utf-8") with pytest.raises(ValueError, match="outside the skill directory"): - _read_skill_resource(skill, "../secret.md") + _read_file_skill_resource(skill, "../secret.md") def test_similar_prefix_directory_does_not_match(self, tmp_path: Path) -> None: """A skill directory named 'skill-a-evil' must not access resources from 'skill-a'.""" - skill = _FileAgentSkill( - frontmatter=_SkillFrontmatter("test", "Test skill"), - body="Body", - source_path=str(tmp_path / "skill-a"), - resource_names=["../skill-a-evil/secret.md"], + skill = Skill( + name="test", + description="Test skill", + content="Body", + path=str(tmp_path / "skill-a"), ) evil_dir = tmp_path / "skill-a-evil" evil_dir.mkdir() (evil_dir / "secret.md").write_text("evil", encoding="utf-8") with pytest.raises(ValueError, match="outside the skill directory"): - _read_skill_resource(skill, "../skill-a-evil/secret.md") + _read_file_skill_resource(skill, "../skill-a-evil/secret.md") # --------------------------------------------------------------------------- -# Tests: _build_skills_instruction_prompt +# Tests: _create_instructions # --------------------------------------------------------------------------- class TestBuildSkillsInstructionPrompt: - """Tests for _build_skills_instruction_prompt.""" + """Tests for _create_instructions.""" def test_returns_none_for_empty_skills(self) -> None: - assert _build_skills_instruction_prompt(None, {}) is None + assert _create_instructions(None, {}) is None def test_default_prompt_contains_skills(self) -> None: skills = { - "my-skill": _FileAgentSkill( - frontmatter=_SkillFrontmatter("my-skill", "Does stuff."), - body="Body", - source_path="/tmp/skill", - ), + "my-skill": Skill(name="my-skill", description="Does stuff.", content="Body"), } - prompt = _build_skills_instruction_prompt(None, skills) + prompt = _create_instructions(None, skills) assert prompt is not None assert "my-skill" in prompt assert "Does stuff." in prompt @@ -402,18 +443,10 @@ def test_default_prompt_contains_skills(self) -> None: def test_skills_sorted_alphabetically(self) -> None: skills = { - "zebra": _FileAgentSkill( - frontmatter=_SkillFrontmatter("zebra", "Z skill."), - body="Body", - source_path="/tmp/z", - ), - "alpha": _FileAgentSkill( - frontmatter=_SkillFrontmatter("alpha", "A skill."), - body="Body", - source_path="/tmp/a", - ), + "zebra": Skill(name="zebra", description="Z skill.", content="Body"), + "alpha": Skill(name="alpha", description="A skill.", content="Body"), } - prompt = _build_skills_instruction_prompt(None, skills) + prompt = _create_instructions(None, skills) assert prompt is not None alpha_pos = prompt.index("alpha") zebra_pos = prompt.index("zebra") @@ -421,62 +454,57 @@ def test_skills_sorted_alphabetically(self) -> None: def test_xml_escapes_metadata(self) -> None: skills = { - "my-skill": _FileAgentSkill( - frontmatter=_SkillFrontmatter("my-skill", 'Uses & "quotes"'), - body="Body", - source_path="/tmp/skill", - ), + "my-skill": Skill(name="my-skill", description='Uses & "quotes"', content="Body"), } - prompt = _build_skills_instruction_prompt(None, skills) + prompt = _create_instructions(None, skills) assert prompt is not None assert "<tags>" in prompt assert "&" in prompt def test_custom_prompt_template(self) -> None: skills = { - "my-skill": _FileAgentSkill( - frontmatter=_SkillFrontmatter("my-skill", "Does stuff."), - body="Body", - source_path="/tmp/skill", - ), + "my-skill": Skill(name="my-skill", description="Does stuff.", content="Body"), } - custom = "Custom header:\n{0}\nCustom footer." - prompt = _build_skills_instruction_prompt(custom, skills) + custom = "Custom header:\n{skills}\nCustom footer." + prompt = _create_instructions(custom, skills) assert prompt is not None assert prompt.startswith("Custom header:") assert prompt.endswith("Custom footer.") def test_invalid_prompt_template_raises(self) -> None: skills = { - "my-skill": _FileAgentSkill( - frontmatter=_SkillFrontmatter("my-skill", "Does stuff."), - body="Body", - source_path="/tmp/skill", - ), + "my-skill": Skill(name="my-skill", description="Does stuff.", content="Body"), + } + with pytest.raises(ValueError, match="valid format string"): + _create_instructions("{invalid}", skills) + + def test_positional_placeholder_raises(self) -> None: + skills = { + "my-skill": Skill(name="my-skill", description="Does stuff.", content="Body"), } with pytest.raises(ValueError, match="valid format string"): - _build_skills_instruction_prompt("{invalid}", skills) + _create_instructions("Header {0} footer", skills) # --------------------------------------------------------------------------- -# Tests: FileAgentSkillsProvider +# Tests: SkillsProvider (file-based) # --------------------------------------------------------------------------- -class TestFileAgentSkillsProvider: - """Tests for the public FileAgentSkillsProvider class.""" +class TestSkillsProvider: + """Tests for file-based usage of SkillsProvider.""" def test_default_source_id(self, tmp_path: Path) -> None: - provider = FileAgentSkillsProvider(str(tmp_path)) - assert provider.source_id == "file_agent_skills" + provider = SkillsProvider(str(tmp_path)) + assert provider.source_id == "agent_skills" def test_custom_source_id(self, tmp_path: Path) -> None: - provider = FileAgentSkillsProvider(str(tmp_path), source_id="custom") + provider = SkillsProvider(str(tmp_path), source_id="custom") assert provider.source_id == "custom" def test_accepts_single_path_string(self, tmp_path: Path) -> None: _write_skill(tmp_path, "my-skill") - provider = FileAgentSkillsProvider(str(tmp_path)) + provider = SkillsProvider(str(tmp_path)) assert len(provider._skills) == 1 def test_accepts_sequence_of_paths(self, tmp_path: Path) -> None: @@ -484,12 +512,12 @@ def test_accepts_sequence_of_paths(self, tmp_path: Path) -> None: dir2 = tmp_path / "dir2" _write_skill(dir1, "skill-a") _write_skill(dir2, "skill-b") - provider = FileAgentSkillsProvider([str(dir1), str(dir2)]) + provider = SkillsProvider([str(dir1), str(dir2)]) assert len(provider._skills) == 2 async def test_before_run_with_skills(self, tmp_path: Path) -> None: _write_skill(tmp_path, "my-skill") - provider = FileAgentSkillsProvider(str(tmp_path)) + provider = SkillsProvider(str(tmp_path)) context = SessionContext(input_messages=[]) await provider.before_run( @@ -506,7 +534,7 @@ async def test_before_run_with_skills(self, tmp_path: Path) -> None: assert tool_names == {"load_skill", "read_skill_resource"} async def test_before_run_without_skills(self, tmp_path: Path) -> None: - provider = FileAgentSkillsProvider(str(tmp_path)) + provider = SkillsProvider(str(tmp_path)) context = SessionContext(input_messages=[]) await provider.before_run( @@ -521,53 +549,64 @@ async def test_before_run_without_skills(self, tmp_path: Path) -> None: def test_load_skill_returns_body(self, tmp_path: Path) -> None: _write_skill(tmp_path, "my-skill", body="Skill body content.") - provider = FileAgentSkillsProvider(str(tmp_path)) + provider = SkillsProvider(str(tmp_path)) result = provider._load_skill("my-skill") - assert result == "Skill body content." + assert "Skill body content." in result + + def test_load_skill_preserves_file_skill_content(self, tmp_path: Path) -> None: + _write_skill( + tmp_path, + "my-skill", + body="See [doc](refs/FAQ.md).", + resources={"refs/FAQ.md": "FAQ content"}, + ) + provider = SkillsProvider(str(tmp_path)) + result = provider._load_skill("my-skill") + assert "See [doc](refs/FAQ.md)." in result def test_load_skill_unknown_returns_error(self, tmp_path: Path) -> None: - provider = FileAgentSkillsProvider(str(tmp_path)) + provider = SkillsProvider(str(tmp_path)) result = provider._load_skill("nonexistent") assert result.startswith("Error:") def test_load_skill_empty_name_returns_error(self, tmp_path: Path) -> None: - provider = FileAgentSkillsProvider(str(tmp_path)) + provider = SkillsProvider(str(tmp_path)) result = provider._load_skill("") assert result.startswith("Error:") - def test_read_skill_resource_returns_content(self, tmp_path: Path) -> None: + async def test_read_skill_resource_returns_content(self, tmp_path: Path) -> None: _write_skill( tmp_path, "my-skill", body="See [doc](refs/FAQ.md).", resources={"refs/FAQ.md": "FAQ content"}, ) - provider = FileAgentSkillsProvider(str(tmp_path)) - result = provider._read_skill_resource("my-skill", "refs/FAQ.md") + provider = SkillsProvider(str(tmp_path)) + result = await provider._read_skill_resource("my-skill", "refs/FAQ.md") assert result == "FAQ content" - def test_read_skill_resource_unknown_skill_returns_error(self, tmp_path: Path) -> None: - provider = FileAgentSkillsProvider(str(tmp_path)) - result = provider._read_skill_resource("nonexistent", "file.md") + async def test_read_skill_resource_unknown_skill_returns_error(self, tmp_path: Path) -> None: + provider = SkillsProvider(str(tmp_path)) + result = await provider._read_skill_resource("nonexistent", "file.md") assert result.startswith("Error:") - def test_read_skill_resource_empty_name_returns_error(self, tmp_path: Path) -> None: + async def test_read_skill_resource_empty_name_returns_error(self, tmp_path: Path) -> None: _write_skill(tmp_path, "my-skill") - provider = FileAgentSkillsProvider(str(tmp_path)) - result = provider._read_skill_resource("my-skill", "") + provider = SkillsProvider(str(tmp_path)) + result = await provider._read_skill_resource("my-skill", "") assert result.startswith("Error:") - def test_read_skill_resource_unknown_resource_returns_error(self, tmp_path: Path) -> None: + async def test_read_skill_resource_unknown_resource_returns_error(self, tmp_path: Path) -> None: _write_skill(tmp_path, "my-skill") - provider = FileAgentSkillsProvider(str(tmp_path)) - result = provider._read_skill_resource("my-skill", "nonexistent.md") + provider = SkillsProvider(str(tmp_path)) + result = await provider._read_skill_resource("my-skill", "nonexistent.md") assert result.startswith("Error:") async def test_skills_sorted_in_prompt(self, tmp_path: Path) -> None: skills_dir = tmp_path / "skills" _write_skill(skills_dir, "zebra", description="Z skill.") _write_skill(skills_dir, "alpha", description="A skill.") - provider = FileAgentSkillsProvider(str(skills_dir)) + provider = SkillsProvider(str(skills_dir)) context = SessionContext(input_messages=[]) await provider.before_run( @@ -582,7 +621,7 @@ async def test_skills_sorted_in_prompt(self, tmp_path: Path) -> None: async def test_xml_escaping_in_prompt(self, tmp_path: Path) -> None: _write_skill(tmp_path, "my-skill", description="Uses & stuff") - provider = FileAgentSkillsProvider(str(tmp_path)) + provider = SkillsProvider(str(tmp_path)) context = SessionContext(input_messages=[]) await provider.before_run( @@ -656,25 +695,30 @@ def test_returns_false_for_regular_files(self, tmp_path: Path) -> None: directory_path = str(skill_dir) + os.sep assert _has_symlink_in_path(full_path, directory_path) is False - def test_validate_resources_rejects_symlinked_resource(self, tmp_path: Path) -> None: - """_discover_and_load_skills should exclude a skill whose resource is a symlink.""" + def test_discover_skips_symlinked_resource(self, tmp_path: Path) -> None: + """_discover_file_skills should skip a symlinked resource but keep the skill.""" skill_dir = tmp_path / "my-skill" skill_dir.mkdir() outside_file = tmp_path / "secret.md" outside_file.write_text("secret content", encoding="utf-8") - # Create SKILL.md referencing a resource + # Create SKILL.md (skill_dir / "SKILL.md").write_text( - "---\nname: my-skill\ndescription: A test skill.\n---\nSee [doc](refs/leak.md).\n", + "---\nname: my-skill\ndescription: A test skill.\n---\nInstructions.\n", encoding="utf-8", ) refs_dir = skill_dir / "refs" refs_dir.mkdir() (refs_dir / "leak.md").symlink_to(outside_file) + # Also add a safe resource + (refs_dir / "safe.md").write_text("safe content", encoding="utf-8") - skills = _discover_and_load_skills([str(tmp_path)]) - assert "my-skill" not in skills + skills = _discover_file_skills([str(tmp_path)]) + assert "my-skill" in skills + resource_names = [r.name for r in skills["my-skill"].resources] + assert "refs/leak.md" not in resource_names + assert "refs/safe.md" in resource_names def test_read_skill_resource_rejects_symlinked_resource(self, tmp_path: Path) -> None: """_read_skill_resource should raise ValueError for a symlinked resource.""" @@ -688,11 +732,966 @@ def test_read_skill_resource_rejects_symlinked_resource(self, tmp_path: Path) -> refs_dir.mkdir() (refs_dir / "leak.md").symlink_to(outside_file) - skill = _FileAgentSkill( - frontmatter=_SkillFrontmatter("test", "Test skill"), - body="See [doc](refs/leak.md).", - source_path=str(skill_dir), - resource_names=["refs/leak.md"], + skill = Skill( + name="test", + description="Test skill", + content="See [doc](refs/leak.md).", + path=str(skill_dir), ) with pytest.raises(ValueError, match="symlink"): - _read_skill_resource(skill, "refs/leak.md") + _read_file_skill_resource(skill, "refs/leak.md") + + +# --------------------------------------------------------------------------- +# Tests: SkillResource +# --------------------------------------------------------------------------- + + +class TestSkillResource: + """Tests for SkillResource dataclass.""" + + def test_static_content(self) -> None: + resource = SkillResource(name="ref", content="static content") + assert resource.name == "ref" + assert resource.content == "static content" + assert resource.function is None + + def test_callable_function(self) -> None: + def my_func() -> str: + return "dynamic" + + resource = SkillResource(name="func", function=my_func) + assert resource.name == "func" + assert resource.content is None + assert resource.function is my_func + + def test_with_description(self) -> None: + resource = SkillResource(name="ref", description="A reference doc.", content="data") + assert resource.description == "A reference doc." + + def test_requires_content_or_function(self) -> None: + with pytest.raises(ValueError, match="must have either content or function"): + SkillResource(name="empty") + + def test_content_and_function_mutually_exclusive(self) -> None: + with pytest.raises(ValueError, match="must have either content or function, not both"): + SkillResource(name="both", content="static", function=lambda: "dynamic") + + +# --------------------------------------------------------------------------- +# Tests: Skill +# --------------------------------------------------------------------------- + + +class TestSkill: + """Tests for Skill dataclass and .resource decorator.""" + + def test_basic_construction(self) -> None: + skill = Skill(name="my-skill", description="A test skill.", content="Instructions.") + assert skill.name == "my-skill" + assert skill.description == "A test skill." + assert skill.content == "Instructions." + assert skill.resources == [] + + def test_construction_with_static_resources(self) -> None: + skill = Skill( + name="my-skill", + description="A test skill.", + content="Instructions.", + resources=[ + SkillResource(name="ref", content="Reference content"), + ], + ) + assert len(skill.resources) == 1 + assert skill.resources[0].name == "ref" + + def test_empty_name_raises(self) -> None: + with pytest.raises(ValueError, match="cannot be empty"): + Skill(name="", description="A skill.", content="Body") + + def test_invalid_name_skipped(self) -> None: + invalid_skill = Skill(name="Invalid-Name", description="A skill.", content="Body") + provider = SkillsProvider(skills=[invalid_skill]) + assert len(provider._skills) == 0 + + def test_name_starts_with_hyphen_skipped(self) -> None: + invalid_skill = Skill(name="-bad-name", description="A skill.", content="Body") + provider = SkillsProvider(skills=[invalid_skill]) + assert len(provider._skills) == 0 + + def test_name_too_long_skipped(self) -> None: + invalid_skill = Skill(name="a" * 65, description="A skill.", content="Body") + provider = SkillsProvider(skills=[invalid_skill]) + assert len(provider._skills) == 0 + + def test_empty_description_raises(self) -> None: + with pytest.raises(ValueError, match="cannot be empty"): + Skill(name="my-skill", description="", content="Body") + + def test_description_too_long_skipped(self) -> None: + invalid_skill = Skill(name="my-skill", description="a" * 1025, content="Body") + provider = SkillsProvider(skills=[invalid_skill]) + assert len(provider._skills) == 0 + + def test_resource_decorator_bare(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource + def get_schema() -> str: + """Get the database schema.""" + return "CREATE TABLE users (id INT)" + + assert len(skill.resources) == 1 + assert skill.resources[0].name == "get_schema" + assert skill.resources[0].description == "Get the database schema." + assert skill.resources[0].function is get_schema + + def test_resource_decorator_with_args(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource(name="custom-name", description="Custom description") + def my_resource() -> str: + return "data" + + assert len(skill.resources) == 1 + assert skill.resources[0].name == "custom-name" + assert skill.resources[0].description == "Custom description" + + def test_resource_decorator_returns_function(self) -> None: + """Decorator should return the original function unchanged.""" + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource + def get_data() -> str: + return "data" + + assert callable(get_data) + assert get_data() == "data" + + def test_multiple_resources(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource + def resource_a() -> str: + return "A" + + @skill.resource + def resource_b() -> str: + return "B" + + assert len(skill.resources) == 2 + names = [r.name for r in skill.resources] + assert "resource_a" in names + assert "resource_b" in names + + def test_resource_decorator_async(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource + async def get_async_data() -> str: + return "async data" + + assert len(skill.resources) == 1 + assert skill.resources[0].function is get_async_data + + +# --------------------------------------------------------------------------- +# Tests: SkillsProvider with code-defined skills +# --------------------------------------------------------------------------- + + +class TestSkillsProviderCodeSkill: + """Tests for SkillsProvider with code-defined skills.""" + + def test_code_skill_only(self) -> None: + skill = Skill(name="prog-skill", description="A code-defined skill.", content="Do the thing.") + provider = SkillsProvider(skills=[skill]) + assert "prog-skill" in provider._skills + + def test_load_skill_returns_content(self) -> None: + skill = Skill(name="prog-skill", description="A skill.", content="Code-defined instructions.") + provider = SkillsProvider(skills=[skill]) + result = provider._load_skill("prog-skill") + assert "prog-skill" in result + assert "A skill." in result + assert "\nCode-defined instructions.\n" in result + assert "" not in result + + def test_load_skill_appends_resource_listing(self) -> None: + skill = Skill( + name="prog-skill", + description="A skill.", + content="Do things.", + resources=[ + SkillResource(name="ref-a", content="a", description="First resource"), + SkillResource(name="ref-b", content="b"), + ], + ) + provider = SkillsProvider(skills=[skill]) + result = provider._load_skill("prog-skill") + assert "prog-skill" in result + assert "A skill." in result + assert "Do things." in result + assert "" in result + assert '' in result + assert '' in result + + def test_load_skill_no_resources_no_listing(self) -> None: + skill = Skill(name="prog-skill", description="A skill.", content="Body only.") + provider = SkillsProvider(skills=[skill]) + result = provider._load_skill("prog-skill") + assert "Body only." in result + assert "" not in result + + async def test_read_static_resource(self) -> None: + skill = Skill( + name="prog-skill", + description="A skill.", + content="Body", + resources=[SkillResource(name="ref", content="static content")], + ) + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "ref") + assert result == "static content" + + async def test_read_callable_resource_sync(self) -> None: + skill = Skill(name="prog-skill", description="A skill.", content="Body") + + @skill.resource + def get_schema() -> str: + return "CREATE TABLE users" + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "get_schema") + assert result == "CREATE TABLE users" + + async def test_read_callable_resource_async(self) -> None: + skill = Skill(name="prog-skill", description="A skill.", content="Body") + + @skill.resource + async def get_data() -> str: + return "async data" + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "get_data") + assert result == "async data" + + async def test_read_resource_case_insensitive(self) -> None: + skill = Skill( + name="prog-skill", + description="A skill.", + content="Body", + resources=[SkillResource(name="MyRef", content="content")], + ) + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "myref") + assert result == "content" + + async def test_read_unknown_resource_returns_error(self) -> None: + skill = Skill(name="prog-skill", description="A skill.", content="Body") + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "nonexistent") + assert result.startswith("Error:") + + async def test_before_run_injects_code_skills(self) -> None: + skill = Skill(name="prog-skill", description="A code-defined skill.", content="Body") + provider = SkillsProvider(skills=[skill]) + context = SessionContext(input_messages=[]) + + await provider.before_run(agent=AsyncMock(), session=AsyncMock(), context=context, state={}) + + assert len(context.instructions) == 1 + assert "prog-skill" in context.instructions[0] + assert len(context.tools) == 2 + + async def test_before_run_empty_provider(self) -> None: + provider = SkillsProvider() + context = SessionContext(input_messages=[]) + + await provider.before_run(agent=AsyncMock(), session=AsyncMock(), context=context, state={}) + + assert len(context.instructions) == 0 + assert len(context.tools) == 0 + + def test_combined_file_and_code_skill(self, tmp_path: Path) -> None: + _write_skill(tmp_path, "file-skill") + prog_skill = Skill(name="prog-skill", description="Code-defined.", content="Body") + provider = SkillsProvider(skill_paths=str(tmp_path), skills=[prog_skill]) + assert "file-skill" in provider._skills + assert "prog-skill" in provider._skills + + def test_duplicate_name_file_wins(self, tmp_path: Path) -> None: + _write_skill(tmp_path, "my-skill", body="File version") + prog_skill = Skill(name="my-skill", description="Code-defined.", content="Prog version") + provider = SkillsProvider(skill_paths=str(tmp_path), skills=[prog_skill]) + # File-based is loaded first, so it wins + assert "File version" in provider._skills["my-skill"].content + + async def test_combined_prompt_includes_both(self, tmp_path: Path) -> None: + _write_skill(tmp_path, "file-skill") + prog_skill = Skill(name="prog-skill", description="A code-defined skill.", content="Body") + provider = SkillsProvider(skill_paths=str(tmp_path), skills=[prog_skill]) + context = SessionContext(input_messages=[]) + + await provider.before_run(agent=AsyncMock(), session=AsyncMock(), context=context, state={}) + + prompt = context.instructions[0] + assert "file-skill" in prompt + assert "prog-skill" in prompt + + def test_custom_resource_extensions(self, tmp_path: Path) -> None: + """SkillsProvider accepts custom resource_extensions.""" + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text( + "---\nname: my-skill\ndescription: A test skill.\n---\nBody.", + encoding="utf-8", + ) + (skill_dir / "data.json").write_text("{}", encoding="utf-8") + (skill_dir / "notes.txt").write_text("notes", encoding="utf-8") + + # Only discover .json files + provider = SkillsProvider(str(tmp_path), resource_extensions=(".json",)) + skill = provider._skills["my-skill"] + resource_names = [r.name for r in skill.resources] + assert "data.json" in resource_names + assert "notes.txt" not in resource_names + + +# --------------------------------------------------------------------------- +# Tests: File-based skill parsing and content +# --------------------------------------------------------------------------- + + +class TestFileBasedSkillParsing: + """Tests for file-based skills parsed from SKILL.md.""" + + def test_content_contains_full_raw_file(self, tmp_path: Path) -> None: + """content stores the entire SKILL.md file including frontmatter.""" + _write_skill(tmp_path, "my-skill", description="A test skill.", body="Instructions here.") + skill = _read_and_parse_skill_file_for_test(tmp_path / "my-skill") + assert "---" in skill.content + assert "name: my-skill" in skill.content + assert "description: A test skill." in skill.content + assert "Instructions here." in skill.content + + def test_name_and_description_from_frontmatter(self, tmp_path: Path) -> None: + _write_skill(tmp_path, "my-skill", description="Skill desc.") + skill = _read_and_parse_skill_file_for_test(tmp_path / "my-skill") + assert skill.name == "my-skill" + assert skill.description == "Skill desc." + + def test_path_set(self, tmp_path: Path) -> None: + _write_skill(tmp_path, "my-skill") + skill = _read_and_parse_skill_file_for_test(tmp_path / "my-skill") + assert skill.path == str(tmp_path / "my-skill") + + def test_resources_populated(self, tmp_path: Path) -> None: + _write_skill(tmp_path, "my-skill", resources={"refs/doc.md": "content"}) + skills = _discover_file_skills([str(tmp_path)]) + assert "my-skill" in skills + resource_names = [r.name for r in skills["my-skill"].resources] + assert "refs/doc.md" in resource_names + + +# --------------------------------------------------------------------------- +# Tests: _load_skill formatting +# --------------------------------------------------------------------------- + + +class TestLoadSkillFormatting: + """Tests for _load_skill output formatting differences between file-based and code-defined skills.""" + + def test_file_skill_returns_raw_content(self, tmp_path: Path) -> None: + """File-based skills return raw SKILL.md content without XML wrapping.""" + _write_skill(tmp_path, "my-skill", body="Do the thing.") + provider = SkillsProvider(str(tmp_path)) + result = provider._load_skill("my-skill") + assert "Do the thing." in result + assert "" not in result + assert "" not in result + + def test_code_skill_wraps_in_xml(self) -> None: + """Code-defined skills are wrapped with name, description, and instructions tags.""" + skill = Skill(name="prog-skill", description="A skill.", content="Do stuff.") + provider = SkillsProvider(skills=[skill]) + result = provider._load_skill("prog-skill") + assert "prog-skill" in result + assert "A skill." in result + assert "\nDo stuff.\n" in result + + def test_code_skill_single_resource_no_description(self) -> None: + """Resource without description omits the description attribute.""" + skill = Skill( + name="prog-skill", + description="A skill.", + content="Body.", + resources=[SkillResource(name="data", content="val")], + ) + provider = SkillsProvider(skills=[skill]) + result = provider._load_skill("prog-skill") + assert '' in result + assert "description=" not in result + + +# --------------------------------------------------------------------------- +# Tests: _discover_resource_files edge cases +# --------------------------------------------------------------------------- + + +class TestDiscoverResourceFilesEdgeCases: + """Additional edge-case tests for filesystem resource discovery.""" + + def test_excludes_skill_md_case_insensitive(self, tmp_path: Path) -> None: + """SKILL.md in any casing is excluded.""" + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + (skill_dir / "skill.md").write_text("lowercase name", encoding="utf-8") + (skill_dir / "other.md").write_text("keep me", encoding="utf-8") + resources = _discover_resource_files(str(skill_dir)) + names = [r.lower() for r in resources] + assert "skill.md" not in names + assert "other.md" in resources + + def test_skips_directories(self, tmp_path: Path) -> None: + """Directories are not included as resources even if their name matches an extension.""" + skill_dir = tmp_path / "my-skill" + subdir = skill_dir / "data.json" + subdir.mkdir(parents=True) + resources = _discover_resource_files(str(skill_dir)) + assert resources == [] + + def test_extension_matching_is_case_insensitive(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + (skill_dir / "NOTES.TXT").write_text("caps", encoding="utf-8") + resources = _discover_resource_files(str(skill_dir)) + assert len(resources) == 1 + + +# --------------------------------------------------------------------------- +# Tests: _is_path_within_directory +# --------------------------------------------------------------------------- + + +class TestIsPathWithinDirectory: + """Tests for _is_path_within_directory.""" + + def test_path_inside_directory(self, tmp_path: Path) -> None: + child = str(tmp_path / "sub" / "file.txt") + assert _is_path_within_directory(child, str(tmp_path)) is True + + def test_path_outside_directory(self, tmp_path: Path) -> None: + outside = str(tmp_path.parent / "other" / "file.txt") + assert _is_path_within_directory(outside, str(tmp_path)) is False + + def test_path_is_directory_itself(self, tmp_path: Path) -> None: + assert _is_path_within_directory(str(tmp_path), str(tmp_path)) is True + + def test_similar_prefix_not_matched(self, tmp_path: Path) -> None: + """'skill-a-evil' is not inside 'skill-a'.""" + dir_a = str(tmp_path / "skill-a") + evil = str(tmp_path / "skill-a-evil" / "file.txt") + assert _is_path_within_directory(evil, dir_a) is False + + +# --------------------------------------------------------------------------- +# Tests: _has_symlink_in_path edge cases +# --------------------------------------------------------------------------- + + +class TestHasSymlinkInPathEdgeCases: + """Edge-case tests for _has_symlink_in_path.""" + + def test_raises_when_path_not_relative(self, tmp_path: Path) -> None: + unrelated = str(tmp_path.parent / "other" / "file.txt") + with pytest.raises(ValueError, match="does not start with directory"): + _has_symlink_in_path(unrelated, str(tmp_path)) + + def test_returns_false_for_empty_relative(self, tmp_path: Path) -> None: + """When path equals directory, relative is empty so no symlinks.""" + assert _has_symlink_in_path(str(tmp_path), str(tmp_path)) is False + + +# --------------------------------------------------------------------------- +# Tests: _validate_skill_metadata +# --------------------------------------------------------------------------- + + +class TestValidateSkillMetadata: + """Tests for _validate_skill_metadata.""" + + def test_valid_metadata(self) -> None: + assert _validate_skill_metadata("my-skill", "A description.", "source") is None + + def test_none_name(self) -> None: + result = _validate_skill_metadata(None, "desc", "source") + assert result is not None + assert "missing a name" in result + + def test_empty_name(self) -> None: + result = _validate_skill_metadata("", "desc", "source") + assert result is not None + assert "missing a name" in result + + def test_whitespace_only_name(self) -> None: + result = _validate_skill_metadata(" ", "desc", "source") + assert result is not None + assert "missing a name" in result + + def test_name_at_max_length(self) -> None: + name = "a" * 64 + assert _validate_skill_metadata(name, "desc", "source") is None + + def test_name_exceeds_max_length(self) -> None: + name = "a" * 65 + result = _validate_skill_metadata(name, "desc", "source") + assert result is not None + assert "invalid name" in result + + def test_name_with_uppercase(self) -> None: + result = _validate_skill_metadata("BadName", "desc", "source") + assert result is not None + assert "invalid name" in result + + def test_name_starts_with_hyphen(self) -> None: + result = _validate_skill_metadata("-bad", "desc", "source") + assert result is not None + assert "invalid name" in result + + def test_name_ends_with_hyphen(self) -> None: + result = _validate_skill_metadata("bad-", "desc", "source") + assert result is not None + assert "invalid name" in result + + def test_single_char_name(self) -> None: + assert _validate_skill_metadata("a", "desc", "source") is None + + def test_none_description(self) -> None: + result = _validate_skill_metadata("my-skill", None, "source") + assert result is not None + assert "missing a description" in result + + def test_empty_description(self) -> None: + result = _validate_skill_metadata("my-skill", "", "source") + assert result is not None + assert "missing a description" in result + + def test_whitespace_only_description(self) -> None: + result = _validate_skill_metadata("my-skill", " ", "source") + assert result is not None + assert "missing a description" in result + + def test_description_at_max_length(self) -> None: + desc = "a" * 1024 + assert _validate_skill_metadata("my-skill", desc, "source") is None + + def test_description_exceeds_max_length(self) -> None: + desc = "a" * 1025 + result = _validate_skill_metadata("my-skill", desc, "source") + assert result is not None + assert "invalid description" in result + + +# --------------------------------------------------------------------------- +# Tests: _discover_skill_directories +# --------------------------------------------------------------------------- + + +class TestDiscoverSkillDirectories: + """Tests for _discover_skill_directories.""" + + def test_finds_skill_at_root(self, tmp_path: Path) -> None: + (tmp_path / "SKILL.md").write_text("---\nname: s\ndescription: d\n---\n", encoding="utf-8") + dirs = _discover_skill_directories([str(tmp_path)]) + assert len(dirs) == 1 + + def test_finds_nested_skill(self, tmp_path: Path) -> None: + sub = tmp_path / "sub" + sub.mkdir() + (sub / "SKILL.md").write_text("---\nname: s\ndescription: d\n---\n", encoding="utf-8") + dirs = _discover_skill_directories([str(tmp_path)]) + assert len(dirs) == 1 + assert str(sub.absolute()) in dirs[0] + + def test_skips_empty_path_string(self) -> None: + dirs = _discover_skill_directories(["", " "]) + assert dirs == [] + + def test_skips_nonexistent_path(self) -> None: + dirs = _discover_skill_directories(["/nonexistent/does/not/exist"]) + assert dirs == [] + + def test_depth_limit_excludes_deep_skill(self, tmp_path: Path) -> None: + deep = tmp_path / "l1" / "l2" / "l3" + deep.mkdir(parents=True) + (deep / "SKILL.md").write_text("---\nname: s\ndescription: d\n---\n", encoding="utf-8") + dirs = _discover_skill_directories([str(tmp_path)]) + assert len(dirs) == 0 + + def test_depth_limit_includes_at_boundary(self, tmp_path: Path) -> None: + at_boundary = tmp_path / "l1" / "l2" + at_boundary.mkdir(parents=True) + (at_boundary / "SKILL.md").write_text("---\nname: s\ndescription: d\n---\n", encoding="utf-8") + dirs = _discover_skill_directories([str(tmp_path)]) + assert len(dirs) == 1 + + +# --------------------------------------------------------------------------- +# Tests: _read_and_parse_skill_file edge cases +# --------------------------------------------------------------------------- + + +class TestReadAndParseSkillFile: + """Tests for _read_and_parse_skill_file.""" + + def test_valid_file(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text( + "---\nname: my-skill\ndescription: A skill.\n---\nBody.", encoding="utf-8" + ) + result = _read_and_parse_skill_file(str(skill_dir)) + assert result is not None + name, desc, content = result + assert name == "my-skill" + assert desc == "A skill." + assert "Body." in content + + def test_missing_skill_md_returns_none(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "no-skill" + skill_dir.mkdir() + result = _read_and_parse_skill_file(str(skill_dir)) + assert result is None + + def test_invalid_frontmatter_returns_none(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "bad-skill" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text("No frontmatter at all.", encoding="utf-8") + result = _read_and_parse_skill_file(str(skill_dir)) + assert result is None + + +# --------------------------------------------------------------------------- +# Tests: _create_resource_element +# --------------------------------------------------------------------------- + + +class TestCreateResourceElement: + """Tests for _create_resource_element.""" + + def test_name_only(self) -> None: + r = SkillResource(name="my-ref", content="data") + elem = _create_resource_element(r) + assert elem == ' ' + + def test_with_description(self) -> None: + r = SkillResource(name="my-ref", description="A reference.", content="data") + elem = _create_resource_element(r) + assert elem == ' ' + + def test_xml_escapes_name(self) -> None: + r = SkillResource(name='ref"special', content="data") + elem = _create_resource_element(r) + assert '"' in elem + + def test_xml_escapes_description(self) -> None: + r = SkillResource(name="ref", description='Uses & "quotes"', content="data") + elem = _create_resource_element(r) + assert "<tags>" in elem + assert "&" in elem + assert """ in elem + + +# --------------------------------------------------------------------------- +# Tests: _read_file_skill_resource edge cases +# --------------------------------------------------------------------------- + + +class TestReadFileSkillResourceEdgeCases: + """Edge-case tests for _read_file_skill_resource.""" + + def test_skill_with_no_path_raises(self) -> None: + skill = Skill(name="no-path", description="No path.", content="Body") + with pytest.raises(ValueError, match="has no path set"): + _read_file_skill_resource(skill, "some-file.md") + + def test_nonexistent_file_raises(self, tmp_path: Path) -> None: + skill_dir = tmp_path / "skill" + skill_dir.mkdir() + skill = Skill(name="test", description="Test.", content="Body", path=str(skill_dir)) + with pytest.raises(ValueError, match="not found in skill"): + _read_file_skill_resource(skill, "missing.md") + + +# --------------------------------------------------------------------------- +# Tests: _normalize_resource_path edge cases +# --------------------------------------------------------------------------- + + +class TestNormalizeResourcePathEdgeCases: + """Additional edge-case tests for _normalize_resource_path.""" + + def test_bare_filename(self) -> None: + assert _normalize_resource_path("file.md") == "file.md" + + def test_deeply_nested_path(self) -> None: + assert _normalize_resource_path("a/b/c/d.md") == "a/b/c/d.md" + + def test_mixed_separators(self) -> None: + assert _normalize_resource_path("a\\b/c\\d.md") == "a/b/c/d.md" + + def test_dot_prefix_only(self) -> None: + assert _normalize_resource_path("./file.md") == "file.md" + + +# --------------------------------------------------------------------------- +# Tests: _discover_file_skills edge cases +# --------------------------------------------------------------------------- + + +class TestDiscoverFileSkillsEdgeCases: + """Edge-case tests for _discover_file_skills.""" + + def test_none_path_returns_empty(self) -> None: + assert _discover_file_skills(None) == {} + + def test_accepts_path_object(self, tmp_path: Path) -> None: + _write_skill(tmp_path, "my-skill") + skills = _discover_file_skills(tmp_path) + assert "my-skill" in skills + + def test_accepts_single_string_path(self, tmp_path: Path) -> None: + _write_skill(tmp_path, "my-skill") + skills = _discover_file_skills(str(tmp_path)) + assert "my-skill" in skills + + +# --------------------------------------------------------------------------- +# Tests: _extract_frontmatter edge cases +# --------------------------------------------------------------------------- + + +class TestExtractFrontmatterEdgeCases: + """Additional edge-case tests for _extract_frontmatter.""" + + def test_whitespace_only_name(self) -> None: + content = "---\nname: ' '\ndescription: A skill.\n---\nBody." + result = _extract_frontmatter(content, "test.md") + assert result is None + + def test_whitespace_only_description(self) -> None: + content = "---\nname: test-skill\ndescription: ' '\n---\nBody." + result = _extract_frontmatter(content, "test.md") + assert result is None + + def test_name_exactly_max_length(self) -> None: + name = "a" * 64 + content = f"---\nname: {name}\ndescription: A skill.\n---\nBody." + result = _extract_frontmatter(content, "test.md") + assert result is not None + assert result[0] == name + + def test_description_exactly_max_length(self) -> None: + desc = "a" * 1024 + content = f"---\nname: test-skill\ndescription: {desc}\n---\nBody." + result = _extract_frontmatter(content, "test.md") + assert result is not None + assert result[1] == desc + + +# --------------------------------------------------------------------------- +# Tests: _create_instructions edge cases +# --------------------------------------------------------------------------- + + +class TestCreateInstructionsEdgeCases: + """Additional edge-case tests for _create_instructions.""" + + def test_custom_template_with_empty_skills_returns_none(self) -> None: + result = _create_instructions("Custom: {skills}", {}) + assert result is None + + def test_custom_template_with_literal_braces(self) -> None: + skills = { + "my-skill": Skill(name="my-skill", description="Skill.", content="Body"), + } + template = "Header {{literal}} {skills} footer." + result = _create_instructions(template, skills) + assert result is not None + assert "{literal}" in result + assert "my-skill" in result + + def test_multiple_skills_generates_sorted_xml(self) -> None: + skills = { + "charlie": Skill(name="charlie", description="C.", content="Body"), + "alpha": Skill(name="alpha", description="A.", content="Body"), + "bravo": Skill(name="bravo", description="B.", content="Body"), + } + result = _create_instructions(None, skills) + assert result is not None + alpha_pos = result.index("alpha") + bravo_pos = result.index("bravo") + charlie_pos = result.index("charlie") + assert alpha_pos < bravo_pos < charlie_pos + + +# --------------------------------------------------------------------------- +# Tests: SkillsProvider edge cases +# --------------------------------------------------------------------------- + + +class TestSkillsProviderEdgeCases: + """Additional edge-case tests for SkillsProvider.""" + + def test_accepts_path_object(self, tmp_path: Path) -> None: + _write_skill(tmp_path, "my-skill") + provider = SkillsProvider(tmp_path) + assert "my-skill" in provider._skills + + def test_load_skill_whitespace_name_returns_error(self, tmp_path: Path) -> None: + _write_skill(tmp_path, "my-skill") + provider = SkillsProvider(str(tmp_path)) + result = provider._load_skill(" ") + assert result.startswith("Error:") + assert "empty" in result + + async def test_read_skill_resource_whitespace_skill_name_returns_error(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource(" ", "ref") + assert result.startswith("Error:") + assert "empty" in result + + async def test_read_skill_resource_whitespace_resource_name_returns_error(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("my-skill", " ") + assert result.startswith("Error:") + assert "empty" in result + + async def test_read_callable_resource_exception_returns_error(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource + def exploding_resource() -> str: + raise RuntimeError("boom") + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("my-skill", "exploding_resource") + assert result.startswith("Error (RuntimeError):") + assert "Failed to read resource" in result + + async def test_read_async_callable_resource_exception_returns_error(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource + async def async_exploding() -> str: + raise ValueError("async boom") + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("my-skill", "async_exploding") + assert result.startswith("Error (ValueError):") + + def test_load_code_skill_xml_escapes_metadata(self) -> None: + skill = Skill(name="my-skill", description='Uses & "quotes"', content="Body") + provider = SkillsProvider(skills=[skill]) + result = provider._load_skill("my-skill") + assert "<tags>" in result + assert "&" in result + + def test_code_skill_deduplication(self) -> None: + skill1 = Skill(name="my-skill", description="First.", content="Body 1") + skill2 = Skill(name="my-skill", description="Second.", content="Body 2") + provider = SkillsProvider(skills=[skill1, skill2]) + assert len(provider._skills) == 1 + assert "First." in provider._skills["my-skill"].description + + async def test_before_run_extends_tools_even_without_instructions(self) -> None: + """If instructions are somehow None but skills exist, tools should still be added.""" + skill = Skill(name="my-skill", description="A skill.", content="Body") + provider = SkillsProvider(skills=[skill]) + context = SessionContext(input_messages=[]) + + await provider.before_run(agent=AsyncMock(), session=AsyncMock(), context=context, state={}) + + assert len(context.tools) == 2 + tool_names = {t.name for t in context.tools} + assert "load_skill" in tool_names + assert "read_skill_resource" in tool_names + + +# --------------------------------------------------------------------------- +# Tests: SkillResource edge cases +# --------------------------------------------------------------------------- + + +class TestSkillResourceEdgeCases: + """Additional edge-case tests for SkillResource.""" + + def test_empty_name_raises(self) -> None: + with pytest.raises(ValueError, match="cannot be empty"): + SkillResource(name="", content="data") + + def test_whitespace_only_name_raises(self) -> None: + with pytest.raises(ValueError, match="cannot be empty"): + SkillResource(name=" ", content="data") + + def test_description_defaults_to_none(self) -> None: + r = SkillResource(name="ref", content="data") + assert r.description is None + + +# --------------------------------------------------------------------------- +# Tests: Skill.resource decorator edge cases +# --------------------------------------------------------------------------- + + +class TestSkillResourceDecoratorEdgeCases: + """Additional edge-case tests for the @skill.resource decorator.""" + + def test_decorator_no_docstring_description_is_none(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource + def no_docs() -> str: + return "data" + + assert skill.resources[0].description is None + + def test_decorator_with_name_only(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource(name="custom-name") + def get_data() -> str: + """Some docs.""" + return "data" + + assert skill.resources[0].name == "custom-name" + # description falls back to docstring + assert skill.resources[0].description == "Some docs." + + def test_decorator_with_description_only(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource(description="Custom desc") + def get_data() -> str: + return "data" + + assert skill.resources[0].name == "get_data" + assert skill.resources[0].description == "Custom desc" + + def test_decorator_preserves_original_function_identity(self) -> None: + skill = Skill(name="my-skill", description="A skill.", content="Body") + + @skill.resource + def original() -> str: + return "original" + + @skill.resource(name="aliased") + def aliased() -> str: + return "aliased" + + # Both decorated functions should still be callable + assert original() == "original" + assert aliased() == "aliased" diff --git a/python/samples/02-agents/skills/basic_skill/README.md b/python/samples/02-agents/skills/basic_skill/README.md index 5c810aab06..1e8e4870e9 100644 --- a/python/samples/02-agents/skills/basic_skill/README.md +++ b/python/samples/02-agents/skills/basic_skill/README.md @@ -1,6 +1,6 @@ # Agent Skills Sample -This sample demonstrates how to use **Agent Skills** with a `FileAgentSkillsProvider` in the Microsoft Agent Framework. +This sample demonstrates how to use **Agent Skills** with a `SkillsProvider` in the Microsoft Agent Framework. ## What are Agent Skills? @@ -20,8 +20,8 @@ Policy-based expense filing with spending limits, receipt requirements, and appr ## Project Structure ``` -basic_skills/ -├── basic_file_skills.py +basic_skill/ +├── basic_skill.py ├── README.md └── skills/ └── expense-report/ @@ -52,7 +52,7 @@ This sample uses `AzureCliCredential` for authentication. Run `az login` in your ```bash cd python -uv run samples/02-agents/skills/basic_skills/basic_file_skills.py +uv run samples/02-agents/skills/basic_skill/basic_skill.py ``` ### Examples diff --git a/python/samples/02-agents/skills/basic_skill/basic_skill.py b/python/samples/02-agents/skills/basic_skill/basic_skill.py index 81cc6c1582..c2f18f73f8 100644 --- a/python/samples/02-agents/skills/basic_skill/basic_skill.py +++ b/python/samples/02-agents/skills/basic_skill/basic_skill.py @@ -4,18 +4,15 @@ import os from pathlib import Path -from agent_framework import Agent, FileAgentSkillsProvider +from agent_framework import Agent, SkillsProvider from agent_framework.azure import AzureOpenAIResponsesClient from azure.identity import AzureCliCredential from dotenv import load_dotenv -# Load environment variables from .env file -load_dotenv() - """ Agent Skills Sample -This sample demonstrates how to use file-based Agent Skills with a FileAgentSkillsProvider. +This sample demonstrates how to use file-based Agent Skills with a SkillsProvider. Agent Skills are modular packages of instructions and resources that extend an agent's capabilities. They follow the progressive disclosure pattern: @@ -27,6 +24,9 @@ - Policy-based expense filing with references and assets """ +# Load environment variables from .env file +load_dotenv() + async def main() -> None: """Run the Agent Skills demo.""" @@ -44,7 +44,7 @@ async def main() -> None: # --- 2. Create the skills provider --- # Discovers skills from the 'skills' directory and makes them available to the agent skills_dir = Path(__file__).parent / "skills" - skills_provider = FileAgentSkillsProvider(skill_paths=str(skills_dir)) + skills_provider = SkillsProvider(skill_paths=str(skills_dir)) # --- 3. Create the agent with skills --- async with Agent( diff --git a/python/samples/02-agents/skills/code_skill/README.md b/python/samples/02-agents/skills/code_skill/README.md new file mode 100644 index 0000000000..828e7c8e22 --- /dev/null +++ b/python/samples/02-agents/skills/code_skill/README.md @@ -0,0 +1,56 @@ +# Code-Defined Agent Skills Sample + +This sample demonstrates how to create **Agent Skills** in Python code, without needing `SKILL.md` files on disk. + +## What are Code-Defined Skills? + +While file-based skills use `SKILL.md` files discovered on disk, code-defined skills let you define skills entirely in Python using `Skill` and `SkillResource` classes. Two patterns are shown: + +1. **Basic Code Skill** — Create a `Skill` directly with static resources (inline content) +2. **Dynamic Resources** — Attach callable resources via the `@skill.resource` decorator that generate content at invocation time + +Both patterns can be combined with file-based skills in a single `SkillsProvider`. + +## Project Structure + +``` +code_skill/ +├── code_skill.py +└── README.md +``` + +## Running the Sample + +### Prerequisites +- An [Azure AI Foundry](https://ai.azure.com/) project with a deployed model (e.g. `gpt-4o-mini`) + +### Environment Variables + +Set the required environment variables in a `.env` file (see `python/.env.example`): + +- `AZURE_AI_PROJECT_ENDPOINT`: Your Azure AI Foundry project endpoint +- `AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME`: The name of your model deployment (defaults to `gpt-4o-mini`) + +### Authentication + +This sample uses `AzureCliCredential` for authentication. Run `az login` in your terminal before running the sample. + +### Run + +```bash +cd python +uv run samples/02-agents/skills/code_skill/code_skill.py +``` + +### Examples + +The sample runs two examples: + +1. **Code style question** — Uses Pattern 1 (static resources): the agent loads the `code-style` skill and reads the `style-guide` resource to answer naming convention questions +2. **Project info question** — Uses Pattern 2 (dynamic resources): the agent reads dynamically generated `environment` and `team-roster` resources + +## Learn More + +- [Agent Skills Specification](https://agentskills.io/) +- [File-based Skills Sample](../basic_skill/) +- [Microsoft Agent Framework Documentation](../../../../../docs/) diff --git a/python/samples/02-agents/skills/code_skill/code_skill.py b/python/samples/02-agents/skills/code_skill/code_skill.py new file mode 100644 index 0000000000..3c95688c49 --- /dev/null +++ b/python/samples/02-agents/skills/code_skill/code_skill.py @@ -0,0 +1,151 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +import os +import sys +from textwrap import dedent + +from agent_framework import Agent, Skill, SkillResource, SkillsProvider +from agent_framework.azure import AzureOpenAIResponsesClient +from azure.identity import AzureCliCredential +from dotenv import load_dotenv + +""" +Code-Defined Agent Skills — Define skills in Python code + +This sample demonstrates how to create Agent Skills in code, +without needing SKILL.md files on disk. Two patterns are shown: + +Pattern 1: Basic Code Skill + Create a Skill instance directly with static resources (inline content). + +Pattern 2: Dynamic Resources + Create a Skill and attach callable resources via the @skill.resource + decorator. Resources can be sync or async functions that generate content at + invocation time. + +Both patterns can be combined with file-based skills in a single SkillsProvider. +""" + +# Load environment variables from .env file +load_dotenv() + +# Pattern 1: Basic Code Skill — direct construction with static resources +code_style_skill = Skill( + name="code-style", + description="Coding style guidelines and conventions for the team", + content=dedent("""\ + Use this skill when answering questions about coding style, conventions, + or best practices for the team. + """), + resources=[ + SkillResource( + name="style-guide", + content=dedent("""\ + # Team Coding Style Guide + + ## General Rules + - Use 4-space indentation (no tabs) + - Maximum line length: 120 characters + - Use type annotations on all public functions + - Use Google-style docstrings + + ## Naming Conventions + - Classes: PascalCase (e.g., UserAccount) + - Functions/methods: snake_case (e.g., get_user_name) + - Constants: UPPER_SNAKE_CASE (e.g., MAX_RETRIES) + - Private members: prefix with underscore (e.g., _internal_state) + """), + ), + ], +) + +# Pattern 2: Dynamic Resources — @skill.resource decorator +project_info_skill = Skill( + name="project-info", + description="Project status and configuration information", + content=dedent("""\ + Use this skill for questions about the current project status, + environment configuration, or team structure. + """), +) + + +@project_info_skill.resource +def environment() -> str: + """Get current environment configuration.""" + env = os.environ.get("APP_ENV", "development") + region = os.environ.get("APP_REGION", "us-east-1") + return f"""\ + # Environment Configuration + - Environment: {env} + - Region: {region} + - Python: {sys.version} + """ + + +@project_info_skill.resource(name="team-roster", description="Current team members and roles") +def get_team_roster() -> str: + """Return the team roster.""" + return """\ + # Team Roster + | Name | Role | + |--------------|-------------------| + | Alice Chen | Tech Lead | + | Bob Smith | Backend Engineer | + | Carol Davis | Frontend Engineer | + """ + + +async def main() -> None: + """Run the code-defined skills demo.""" + endpoint = os.environ["AZURE_AI_PROJECT_ENDPOINT"] + deployment = os.environ.get("AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME", "gpt-4o-mini") + + client = AzureOpenAIResponsesClient( + project_endpoint=endpoint, + deployment_name=deployment, + credential=AzureCliCredential(), + ) + + # Create the skills provider with both code-defined skills + skills_provider = SkillsProvider( + skills=[code_style_skill, project_info_skill], + ) + + async with Agent( + client=client, + instructions="You are a helpful assistant for our development team.", + context_providers=[skills_provider], + ) as agent: + # Example 1: Code style question (Pattern 1 — static resources) + print("Example 1: Code style question") + print("-------------------------------") + response = await agent.run("What naming convention should I use for class attributes?") + print(f"Agent: {response}\n") + + # Example 2: Project info question (Pattern 2 — dynamic resources) + print("Example 2: Project info question") + print("---------------------------------") + response = await agent.run("What environment are we running in and who is on the team?") + print(f"Agent: {response}\n") + + """ + Expected output: + + Example 1: Code style question + ------------------------------- + Agent: Based on our team's coding style guide, class attributes should follow + snake_case naming. Private attributes use an underscore prefix (_internal_state). + Constants use UPPER_SNAKE_CASE (MAX_RETRIES). + + Example 2: Project info question + --------------------------------- + Agent: We're running in the development environment in us-east-1. + The team consists of Alice Chen (Tech Lead), Bob Smith (Backend Engineer), + and Carol Davis (Frontend Engineer). + """ + + +if __name__ == "__main__": + asyncio.run(main())