diff --git a/AGENTS.md b/AGENTS.md index 10a66fcd7..21c32539c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -130,7 +130,11 @@ strands-agents/ │ ├── plugins/ # Plugin system │ │ ├── plugin.py # Plugin base class │ │ ├── decorator.py # @hook decorator -│ │ └── registry.py # PluginRegistry for tracking plugins +│ │ ├── registry.py # PluginRegistry for tracking plugins +│ │ └── skills/ # Agent Skills integration +│ │ ├── __init__.py # Skills package exports +│ │ ├── skill.py # Skill dataclass +│ │ └── agent_skills.py # AgentSkills plugin implementation │ │ │ ├── handlers/ # Event handlers │ │ └── callback_handler.py # Callback handling diff --git a/pyproject.toml b/pyproject.toml index b53194486..e07f3bac4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,7 @@ dependencies = [ "mcp>=1.23.0,<2.0.0", "pydantic>=2.4.0,<3.0.0", "typing-extensions>=4.13.2,<5.0.0", + "pyyaml>=6.0.0,<7.0.0", "watchdog>=6.0.0,<7.0.0", "opentelemetry-api>=1.30.0,<2.0.0", "opentelemetry-sdk>=1.30.0,<2.0.0", diff --git a/src/strands/__init__.py b/src/strands/__init__.py index be939d5b1..3e1528fa6 100644 --- a/src/strands/__init__.py +++ b/src/strands/__init__.py @@ -4,17 +4,19 @@ from .agent.agent import Agent from .agent.base import AgentBase from .event_loop._retry import ModelRetryStrategy -from .plugins import Plugin +from .plugins import AgentSkills, Plugin, Skill from .tools.decorator import tool from .types.tools import ToolContext __all__ = [ "Agent", "AgentBase", + "AgentSkills", "agent", "models", "ModelRetryStrategy", "Plugin", + "Skill", "tool", "ToolContext", "types", diff --git a/src/strands/plugins/__init__.py b/src/strands/plugins/__init__.py index c4b7c72c7..d7ca4c9b2 100644 --- a/src/strands/plugins/__init__.py +++ b/src/strands/plugins/__init__.py @@ -6,8 +6,11 @@ from .decorator import hook from .plugin import Plugin +from .skills import AgentSkills, Skill __all__ = [ + "AgentSkills", "Plugin", + "Skill", "hook", ] diff --git a/src/strands/plugins/skills/__init__.py b/src/strands/plugins/skills/__init__.py new file mode 100644 index 000000000..f6cf8728b --- /dev/null +++ b/src/strands/plugins/skills/__init__.py @@ -0,0 +1,31 @@ +"""AgentSkills.io integration for Strands Agents. + +This module provides the AgentSkills plugin for integrating AgentSkills.io skills +into Strands agents. Skills enable progressive disclosure of instructions: +metadata is injected into the system prompt upfront, and full instructions +are loaded on demand via a tool. + +Example Usage: + ```python + from strands import Agent + from strands.plugins.skills import Skill, AgentSkills + + # Load from filesystem via classmethods + skill = Skill.from_file("./skills/pdf-processing") + skills = Skill.from_directory("./skills/") + + # Or let the plugin resolve paths automatically + plugin = AgentSkills(skills=["./skills/pdf-processing"]) + agent = Agent(plugins=[plugin]) + ``` +""" + +from .agent_skills import AgentSkills, SkillSource, SkillSources +from .skill import Skill + +__all__ = [ + "AgentSkills", + "Skill", + "SkillSource", + "SkillSources", +] diff --git a/src/strands/plugins/skills/agent_skills.py b/src/strands/plugins/skills/agent_skills.py new file mode 100644 index 000000000..97ac86d93 --- /dev/null +++ b/src/strands/plugins/skills/agent_skills.py @@ -0,0 +1,393 @@ +"""AgentSkills plugin for integrating Agent Skills into Strands agents. + +This module provides the AgentSkills class that extends the Plugin base class +to add Agent Skills support. The plugin registers a tool for activating +skills, and injects skill metadata into the system prompt. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import TYPE_CHECKING, Any, TypeAlias +from xml.sax.saxutils import escape + +from ...hooks.events import BeforeInvocationEvent +from ...plugins import Plugin, hook +from ...tools.decorator import tool +from ...types.tools import ToolContext +from .skill import Skill + +if TYPE_CHECKING: + from ...agent.agent import Agent + +logger = logging.getLogger(__name__) + +_DEFAULT_STATE_KEY = "agent_skills" +_RESOURCE_DIRS = ("scripts", "references", "assets") +_DEFAULT_MAX_RESOURCE_FILES = 20 + +SkillSource: TypeAlias = str | Path | Skill +"""A single skill source: path string, Path object, or Skill instance.""" + +SkillSources: TypeAlias = SkillSource | list[SkillSource] +"""One or more skill sources.""" + + +def _normalize_sources(sources: SkillSources) -> list[SkillSource]: + """Normalize a single source or list of sources into a list.""" + if isinstance(sources, list): + return sources + return [sources] + + +class AgentSkills(Plugin): + """Plugin that integrates Agent Skills into a Strands agent. + + The AgentSkills plugin extends the Plugin base class and provides: + + 1. A ``skills`` tool that allows the agent to activate skills on demand + 2. System prompt injection of available skill metadata before each invocation + 3. Session persistence of active skill state via ``agent.state`` + + Skills can be provided as filesystem paths (to individual skill directories or + parent directories containing multiple skills) or as pre-built ``Skill`` instances. + + Example: + ```python + from strands import Agent + from strands.plugins.skills import Skill, AgentSkills + + # Load from filesystem + plugin = AgentSkills(skills=["./skills/pdf-processing", "./skills/"]) + + # Or provide Skill instances directly + skill = Skill(name="my-skill", description="A custom skill", instructions="Do the thing") + plugin = AgentSkills(skills=[skill]) + + agent = Agent(plugins=[plugin]) + ``` + """ + + name = "agent_skills" + + def __init__( + self, + skills: SkillSources, + state_key: str = _DEFAULT_STATE_KEY, + max_resource_files: int = _DEFAULT_MAX_RESOURCE_FILES, + strict: bool = False, + ) -> None: + """Initialize the AgentSkills plugin. + + Args: + skills: One or more skill sources. Can be a single value or a list. Each element can be: + + - A ``str`` or ``Path`` to a skill directory (containing SKILL.md) + - A ``str`` or ``Path`` to a parent directory (containing skill subdirectories) + - A ``Skill`` dataclass instance + state_key: Key used to store plugin state in ``agent.state``. + max_resource_files: Maximum number of resource files to list in skill responses. + strict: If True, raise on skill validation issues. If False (default), warn and load anyway. + """ + self._strict = strict + self._skills: dict[str, Skill] = self._resolve_skills(_normalize_sources(skills)) + self._state_key = state_key + self._max_resource_files = max_resource_files + super().__init__() + + def init_agent(self, agent: Agent) -> None: + """Initialize the plugin with an agent instance. + + Decorated hooks and tools are auto-registered by the plugin registry. + + Args: + agent: The agent instance to extend with skills support. + """ + if not self._skills: + logger.warning("no skills were loaded, the agent will have no skills available") + logger.debug("skill_count=<%d> | skills plugin initialized", len(self._skills)) + + @tool(context=True) + def skills(self, skill_name: str, tool_context: ToolContext) -> str: # noqa: D417 + """Activate a skill to load its full instructions. + + Use this tool to load the complete instructions for a skill listed in + the available_skills section of your system prompt. + + Args: + skill_name: Name of the skill to activate. + """ + if not skill_name: + available = ", ".join(self._skills) + return f"Error: skill_name is required. Available skills: {available}" + + found = self._skills.get(skill_name) + if found is None: + available = ", ".join(self._skills) + return f"Skill '{skill_name}' not found. Available skills: {available}" + + logger.debug("skill_name=<%s> | skill activated", skill_name) + self._track_activated_skill(tool_context.agent, skill_name) + return self._format_skill_response(found) + + @hook + def _on_before_invocation(self, event: BeforeInvocationEvent) -> None: + """Inject skill metadata into the system prompt before each invocation. + + Removes the previously injected XML block (if any) via exact string + replacement, then appends a fresh one. Uses agent state to track the + injected XML per-agent, so a single plugin instance can be shared + across multiple agents safely. + + Args: + event: The before-invocation event containing the agent reference. + """ + agent = event.agent + + current_prompt = agent.system_prompt or "" + + # Remove the previously injected XML block by exact match + state_data = agent.state.get(self._state_key) + last_injected_xml = state_data.get("last_injected_xml") if isinstance(state_data, dict) else None + if last_injected_xml is not None: + if last_injected_xml in current_prompt: + current_prompt = current_prompt.replace(last_injected_xml, "") + else: + logger.warning("unable to find previously injected skills XML in system prompt, re-appending") + + skills_xml = self._generate_skills_xml() + injection = f"\n\n{skills_xml}" + new_prompt = f"{current_prompt}{injection}" if current_prompt else skills_xml + + new_injected_xml = injection if current_prompt else skills_xml + self._set_state_field(agent, "last_injected_xml", new_injected_xml) + agent.system_prompt = new_prompt + + def get_available_skills(self) -> list[Skill]: + """Get the list of available skills. + + Returns: + A copy of the current skills list. + """ + return list(self._skills.values()) + + def set_available_skills(self, skills: SkillSources) -> None: + """Set the available skills, replacing any existing ones. + + Each element can be a ``Skill`` instance, a ``str`` or ``Path`` to a + skill directory (containing SKILL.md), or a ``str`` or ``Path`` to a + parent directory containing skill subdirectories. + + Note: this does not persist state or deactivate skills on any agent. + Active skill state is managed per-agent and will be reconciled on the + next tool call or invocation. + + Args: + skills: One or more skill sources to resolve and set. + """ + self._skills = self._resolve_skills(_normalize_sources(skills)) + + + def _format_skill_response(self, skill: Skill) -> str: + """Format the tool response when a skill is activated. + + Includes the full instructions along with relevant metadata fields + and a listing of available resource files (scripts, references, assets) + for filesystem-based skills. + + Args: + skill: The activated skill. + + Returns: + Formatted string with skill instructions and metadata. + """ + if not skill.instructions: + return f"Skill '{skill.name}' activated (no instructions available)." + + parts: list[str] = [skill.instructions] + + metadata_lines: list[str] = [] + if skill.allowed_tools: + metadata_lines.append(f"Allowed tools: {', '.join(skill.allowed_tools)}") + if skill.compatibility: + metadata_lines.append(f"Compatibility: {skill.compatibility}") + if skill.path is not None: + metadata_lines.append(f"Location: {skill.path / 'SKILL.md'}") + + if metadata_lines: + parts.append("\n---\n" + "\n".join(metadata_lines)) + + if skill.path is not None: + resources = self._list_skill_resources(skill.path) + if resources: + parts.append("\nAvailable resources:\n" + "\n".join(f" {r}" for r in resources)) + + return "\n".join(parts) + + def _list_skill_resources(self, skill_path: Path) -> list[str]: + """List resource files in a skill's optional directories. + + Scans the ``scripts/``, ``references/``, and ``assets/`` subdirectories + for files, returning relative paths. Results are capped at + ``max_resource_files`` to avoid context bloat. + + Args: + skill_path: Path to the skill directory. + + Returns: + List of relative file paths (e.g. ``scripts/extract.py``). + """ + files: list[str] = [] + + for dir_name in _RESOURCE_DIRS: + resource_dir = skill_path / dir_name + if not resource_dir.is_dir(): + continue + + for file_path in sorted(resource_dir.rglob("*")): + if not file_path.is_file(): + continue + files.append(file_path.relative_to(skill_path).as_posix()) + if len(files) >= self._max_resource_files: + files.append(f"... (truncated at {self._max_resource_files} files)") + return files + + return files + + def _generate_skills_xml(self) -> str: + """Generate the XML block listing available skills for the system prompt. + + When no skills are loaded, returns a block indicating no skills are available. + Otherwise includes a ```` element for skills loaded from the filesystem, + following the AgentSkills.io integration spec. + + Returns: + XML-formatted string with skill metadata. + """ + if not self._skills: + return "\nNo skills are currently available.\n" + + lines: list[str] = [""] + + for skill in self._skills.values(): + lines.append("") + lines.append(f"{escape(skill.name)}") + lines.append(f"{escape(skill.description)}") + if skill.path is not None: + lines.append(f"{escape(str(skill.path / 'SKILL.md'))}") + lines.append("") + + lines.append("") + return "\n".join(lines) + + def _resolve_skills(self, sources: list[SkillSource]) -> dict[str, Skill]: + """Resolve a list of skill sources into Skill instances. + + Each source can be a Skill instance, a path to a skill directory, + or a path to a parent directory containing multiple skills. + + Args: + sources: List of skill sources to resolve. + + Returns: + Dict mapping skill names to Skill instances. + """ + resolved: dict[str, Skill] = {} + + for source in sources: + if isinstance(source, Skill): + if source.name in resolved: + logger.warning("name=<%s> | duplicate skill name, overwriting previous skill", source.name) + resolved[source.name] = source + else: + path = Path(source).resolve() + if not path.exists(): + logger.warning("path=<%s> | skill source path does not exist, skipping", path) + continue + + if path.is_dir(): + # Check if this directory itself is a skill (has SKILL.md) + has_skill_md = (path / "SKILL.md").is_file() or (path / "skill.md").is_file() + + if has_skill_md: + try: + skill = Skill.from_file(path, strict=self._strict) + if skill.name in resolved: + logger.warning( + "name=<%s> | duplicate skill name, overwriting previous skill", skill.name + ) + resolved[skill.name] = skill + except (ValueError, FileNotFoundError) as e: + logger.warning("path=<%s> | failed to load skill: %s", path, e) + else: + # Treat as parent directory containing skill subdirectories + for skill in Skill.from_directory(path, strict=self._strict): + if skill.name in resolved: + logger.warning( + "name=<%s> | duplicate skill name, overwriting previous skill", skill.name + ) + resolved[skill.name] = skill + elif path.is_file() and path.name.lower() == "skill.md": + try: + skill = Skill.from_file(path, strict=self._strict) + if skill.name in resolved: + logger.warning("name=<%s> | duplicate skill name, overwriting previous skill", skill.name) + resolved[skill.name] = skill + except (ValueError, FileNotFoundError) as e: + logger.warning("path=<%s> | failed to load skill: %s", path, e) + + logger.debug("source_count=<%d>, resolved_count=<%d> | skills resolved", len(sources), len(resolved)) + return resolved + + def _set_state_field(self, agent: Agent, key: str, value: Any) -> None: + """Set a single field in the plugin's agent state dict. + + Args: + agent: The agent whose state to update. + key: The state field key. + value: The value to set. + + Raises: + TypeError: If the existing state value is not a dict. + """ + state_data = agent.state.get(self._state_key) + if state_data is not None and not isinstance(state_data, dict): + raise TypeError(f"expected dict for state key '{self._state_key}', got {type(state_data).__name__}") + if state_data is None: + state_data = {} + state_data[key] = value + agent.state.set(self._state_key, state_data) + + def _track_activated_skill(self, agent: Agent, skill_name: str) -> None: + """Record a skill activation in agent state. + + Maintains an ordered list of activated skill names (most recent last), + without duplicates. + + Args: + agent: The agent whose state to update. + skill_name: Name of the activated skill. + """ + state_data = agent.state.get(self._state_key) + activated: list[str] = state_data.get("activated_skills", []) if isinstance(state_data, dict) else [] + if skill_name in activated: + activated.remove(skill_name) + activated.append(skill_name) + self._set_state_field(agent, "activated_skills", activated) + + def get_activated_skills(self, agent: Agent) -> list[str]: + """Get the list of skills activated by this agent. + + Returns skill names in activation order (most recent last). + + Args: + agent: The agent to query. + + Returns: + List of activated skill names. + """ + state_data = agent.state.get(self._state_key) + if isinstance(state_data, dict): + return list(state_data.get("activated_skills", [])) + return [] diff --git a/src/strands/plugins/skills/skill.py b/src/strands/plugins/skills/skill.py new file mode 100644 index 000000000..3e1b6bba5 --- /dev/null +++ b/src/strands/plugins/skills/skill.py @@ -0,0 +1,377 @@ +"""Skill data model and loading utilities for AgentSkills.io skills. + +This module defines the Skill dataclass and provides classmethods for +discovering, parsing, and loading skills from the filesystem or raw content. +Skills are directories containing a SKILL.md file with YAML frontmatter +metadata and markdown instructions. +""" + +from __future__ import annotations + +import logging +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import yaml + +logger = logging.getLogger(__name__) + +_SKILL_NAME_PATTERN = re.compile(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$") +_MAX_SKILL_NAME_LENGTH = 64 + + +def _find_skill_md(skill_dir: Path) -> Path: + """Find the SKILL.md file in a skill directory. + + Searches for SKILL.md (case-sensitive preferred) or skill.md as a fallback. + + Args: + skill_dir: Path to the skill directory. + + Returns: + Path to the SKILL.md file. + + Raises: + FileNotFoundError: If no SKILL.md file is found in the directory. + """ + for name in ("SKILL.md", "skill.md"): + candidate = skill_dir / name + if candidate.is_file(): + return candidate + + raise FileNotFoundError(f"path=<{skill_dir}> | no SKILL.md found in skill directory") + + +def _parse_frontmatter(content: str) -> tuple[dict[str, Any], str]: + """Parse YAML frontmatter and body from SKILL.md content. + + Extracts the YAML frontmatter between ``---`` delimiters at line boundaries + and returns parsed key-value pairs along with the remaining markdown body. + + Args: + content: Full content of a SKILL.md file. + + Returns: + Tuple of (frontmatter_dict, body_string). + + Raises: + ValueError: If the frontmatter is malformed or missing required delimiters. + """ + stripped = content.strip() + if not stripped.startswith("---"): + raise ValueError("SKILL.md must start with --- frontmatter delimiter") + + # Find the closing --- delimiter (first line after the opener that is only dashes) + match = re.search(r"\n^---\s*$", stripped, re.MULTILINE) + if match is None: + raise ValueError("SKILL.md frontmatter missing closing --- delimiter") + + frontmatter_str = stripped[3 : match.start()].strip() + body = stripped[match.end() :].strip() + + try: + result = yaml.safe_load(frontmatter_str) + except yaml.YAMLError: + # AgentSkills spec recommends handling malformed YAML (e.g. unquoted colons in values) + # to improve cross-client compatibility. See: agentskills.io/client-implementation/adding-skills-support + logger.warning("YAML parse failed, retrying with colon-quoting fallback") + fixed = _fix_yaml_colons(frontmatter_str) + result = yaml.safe_load(fixed) + + frontmatter: dict[str, Any] = result if isinstance(result, dict) else {} + return frontmatter, body + + +def _fix_yaml_colons(yaml_str: str) -> str: + """Attempt to fix common YAML issues like unquoted colons in values. + + Wraps values containing colons in double quotes to handle cases like: + ``description: Use this skill when: the user asks about PDFs`` + + Args: + yaml_str: The raw YAML string to fix. + + Returns: + The fixed YAML string. + """ + lines: list[str] = [] + for line in yaml_str.splitlines(): + # Match key: value where value contains another colon + match = re.match(r"^(\s*\w[\w-]*):\s+(.+)$", line) + if match: + key, value = match.group(1), match.group(2) + # If value contains a colon and isn't already quoted + if ":" in value and not (value.startswith('"') or value.startswith("'")): + line = f'{key}: "{value}"' + lines.append(line) + return "\n".join(lines) + + +def _validate_skill_name(name: str, dir_path: Path | None = None, *, strict: bool = False) -> None: + """Validate a skill name per the AgentSkills.io specification. + + In lenient mode (default), logs warnings for cosmetic issues but does not raise. + In strict mode, raises ValueError for any validation failure. + + Rules checked: + - 1-64 characters long + - Lowercase alphanumeric characters and hyphens only + - Cannot start or end with a hyphen + - No consecutive hyphens + - Must match parent directory name (if loaded from disk) + + Args: + name: The skill name to validate. + dir_path: Optional path to the skill directory for name matching. + strict: If True, raise ValueError on any issue. If False (default), log warnings. + + Raises: + ValueError: If the skill name is empty, or if strict=True and any rule is violated. + """ + if not name: + raise ValueError("Skill name cannot be empty") + + if len(name) > _MAX_SKILL_NAME_LENGTH: + msg = "name=<%s> | skill name exceeds %d character limit" + if strict: + raise ValueError(msg % (name, _MAX_SKILL_NAME_LENGTH)) + logger.warning(msg, name, _MAX_SKILL_NAME_LENGTH) + + if not _SKILL_NAME_PATTERN.match(name): + msg = ( + "name=<%s> | skill name should be 1-64 lowercase alphanumeric characters or hyphens, " + "should not start/end with hyphen" + ) + if strict: + raise ValueError(msg % name) + logger.warning(msg, name) + + if "--" in name: + msg = "name=<%s> | skill name contains consecutive hyphens" + if strict: + raise ValueError(msg % name) + logger.warning(msg, name) + + if dir_path is not None and dir_path.name != name: + msg = "name=<%s>, directory=<%s> | skill name does not match parent directory name" + if strict: + raise ValueError(msg % (name, dir_path.name)) + logger.warning(msg, name, dir_path.name) + + +def _build_skill_from_frontmatter( + frontmatter: dict[str, Any], + body: str, +) -> Skill: + """Build a Skill instance from parsed frontmatter and body. + + Args: + frontmatter: Parsed YAML frontmatter dict. + body: Markdown body content. + + Returns: + A populated Skill instance. + """ + # Parse allowed-tools (space-delimited string or YAML list) + allowed_tools_raw = frontmatter.get("allowed-tools") or frontmatter.get("allowed_tools") + allowed_tools: list[str] | None = None + if isinstance(allowed_tools_raw, str) and allowed_tools_raw.strip(): + allowed_tools = allowed_tools_raw.strip().split() + elif isinstance(allowed_tools_raw, list): + allowed_tools = [str(item) for item in allowed_tools_raw if item] + + # Parse metadata (nested mapping) + metadata_raw = frontmatter.get("metadata", {}) + metadata: dict[str, Any] = {} + if isinstance(metadata_raw, dict): + metadata = {str(k): v for k, v in metadata_raw.items()} + + skill_license = frontmatter.get("license") + compatibility = frontmatter.get("compatibility") + + return Skill( + name=frontmatter["name"], + description=frontmatter["description"], + instructions=body, + allowed_tools=allowed_tools, + metadata=metadata, + license=str(skill_license) if skill_license else None, + compatibility=str(compatibility) if compatibility else None, + ) + + +@dataclass +class Skill: + r"""Represents an agent skill with metadata and instructions. + + A skill encapsulates a set of instructions and metadata that can be + dynamically loaded by an agent at runtime. Skills support progressive + disclosure: metadata is shown upfront in the system prompt, and full + instructions are loaded on demand via a tool. + + Skills can be created directly or via convenience classmethods:: + + # From a skill directory on disk + skill = Skill.from_file("./skills/my-skill") + + # From raw SKILL.md content + skill = Skill.from_content("---\nname: my-skill\n...") + + # Load all skills from a parent directory + skills = Skill.from_directory("./skills/") + + Attributes: + name: Unique identifier for the skill (1-64 chars, lowercase alphanumeric + hyphens). + description: Human-readable description of what the skill does. + instructions: Full markdown instructions from the SKILL.md body. + path: Filesystem path to the skill directory, if loaded from disk. + allowed_tools: List of tool names the skill is allowed to use. (Experimental: not yet enforced) + metadata: Additional key-value metadata from the SKILL.md frontmatter. + license: License identifier (e.g., "Apache-2.0"). + compatibility: Compatibility information string. + """ + + name: str + description: str + instructions: str = "" + path: Path | None = None + allowed_tools: list[str] | None = None + metadata: dict[str, Any] = field(default_factory=dict) + license: str | None = None + compatibility: str | None = None + + @classmethod + def from_file(cls, skill_path: str | Path, *, strict: bool = False) -> Skill: + """Load a single skill from a directory containing SKILL.md. + + Resolves the filesystem path, reads the file content, and delegates + to ``from_content`` for parsing. After loading, sets the skill's + ``path`` and validates the skill name against the parent directory. + + Args: + skill_path: Path to the skill directory or the SKILL.md file itself. + strict: If True, raise on any validation issue. If False (default), warn and load anyway. + + Returns: + A Skill instance populated from the SKILL.md file. + + Raises: + FileNotFoundError: If the path does not exist or SKILL.md is not found. + ValueError: If the skill metadata is invalid. + """ + skill_path = Path(skill_path).resolve() + + if skill_path.is_file() and skill_path.name.lower() == "skill.md": + skill_md_path = skill_path + skill_dir = skill_path.parent + elif skill_path.is_dir(): + skill_dir = skill_path + skill_md_path = _find_skill_md(skill_dir) + else: + raise FileNotFoundError( + f"path=<{skill_path}> | skill path does not exist or is not a valid skill directory" + ) + + logger.debug("path=<%s> | loading skill", skill_md_path) + + content = skill_md_path.read_text(encoding="utf-8") + skill = cls.from_content(content, strict=strict) + + # Set path and check directory name match (from_content already validated the name format) + skill.path = skill_dir + if skill_dir.name != skill.name: + msg = "name=<%s>, directory=<%s> | skill name does not match parent directory name" + if strict: + raise ValueError(msg % (skill.name, skill_dir.name)) + logger.warning(msg, skill.name, skill_dir.name) + + logger.debug("name=<%s>, path=<%s> | skill loaded successfully", skill.name, skill.path) + return skill + + @classmethod + def from_content(cls, content: str, *, strict: bool = False) -> Skill: + """Parse SKILL.md content into a Skill instance. + + This is a convenience method for creating a Skill from raw SKILL.md + content (YAML frontmatter + markdown body) without requiring a file on + disk. + + Example:: + + content = '''--- + name: my-skill + description: Does something useful + --- + # Instructions + Follow these steps... + ''' + skill = Skill.from_content(content) + + Args: + content: Raw SKILL.md content with YAML frontmatter and markdown body. + strict: If True, raise on any validation issue. If False (default), warn and load anyway. + + Returns: + A Skill instance populated from the parsed content. + + Raises: + ValueError: If the content is missing required fields or has invalid frontmatter. + """ + frontmatter, body = _parse_frontmatter(content) + + name = frontmatter.get("name") + if not isinstance(name, str) or not name: + raise ValueError("SKILL.md content must have a 'name' field in frontmatter") + + description = frontmatter.get("description") + if not isinstance(description, str) or not description: + raise ValueError("SKILL.md content must have a 'description' field in frontmatter") + + _validate_skill_name(name, strict=strict) + + return _build_skill_from_frontmatter(frontmatter, body) + + @classmethod + def from_directory(cls, skills_dir: str | Path, *, strict: bool = False) -> list[Skill]: + """Load all skills from a parent directory containing skill subdirectories. + + Each subdirectory containing a SKILL.md file is treated as a skill. + Subdirectories without SKILL.md are silently skipped. + + Args: + skills_dir: Path to the parent directory containing skill subdirectories. + strict: If True, raise on any validation issue. If False (default), warn and load anyway. + + Returns: + List of Skill instances loaded from the directory. + + Raises: + FileNotFoundError: If the skills directory does not exist. + """ + skills_dir = Path(skills_dir).resolve() + + if not skills_dir.is_dir(): + raise FileNotFoundError(f"path=<{skills_dir}> | skills directory does not exist") + + skills: list[Skill] = [] + + for child in sorted(skills_dir.iterdir()): + if not child.is_dir(): + continue + + try: + _find_skill_md(child) + except FileNotFoundError: + logger.debug("path=<%s> | skipping directory without SKILL.md", child) + continue + + try: + skill = cls.from_file(child, strict=strict) + skills.append(skill) + except (ValueError, FileNotFoundError) as e: + logger.warning("path=<%s> | skipping skill due to error: %s", child, e) + + logger.debug("path=<%s>, count=<%d> | loaded skills from directory", skills_dir, len(skills)) + return skills diff --git a/tests/strands/plugins/skills/__init__.py b/tests/strands/plugins/skills/__init__.py new file mode 100644 index 000000000..9bd23c0ed --- /dev/null +++ b/tests/strands/plugins/skills/__init__.py @@ -0,0 +1 @@ +"""Tests for the skills plugin package.""" diff --git a/tests/strands/plugins/skills/test_agent_skills.py b/tests/strands/plugins/skills/test_agent_skills.py new file mode 100644 index 000000000..8c6ab10bd --- /dev/null +++ b/tests/strands/plugins/skills/test_agent_skills.py @@ -0,0 +1,699 @@ +"""Tests for the AgentSkills plugin.""" + +import logging +from pathlib import Path +from unittest.mock import MagicMock + +from strands.hooks.events import BeforeInvocationEvent +from strands.hooks.registry import HookRegistry +from strands.plugins.registry import _PluginRegistry +from strands.plugins.skills.agent_skills import AgentSkills +from strands.plugins.skills.skill import Skill +from strands.types.tools import ToolContext + + +def _make_skill(name: str = "test-skill", description: str = "A test skill", instructions: str = "Do the thing."): + """Helper to create a Skill instance.""" + return Skill(name=name, description=description, instructions=instructions) + + +def _make_skill_dir(parent: Path, name: str, description: str = "A test skill") -> Path: + """Helper to create a skill directory with SKILL.md.""" + skill_dir = parent / name + skill_dir.mkdir(parents=True, exist_ok=True) + content = f"---\nname: {name}\ndescription: {description}\n---\n# Instructions for {name}\n" + (skill_dir / "SKILL.md").write_text(content) + return skill_dir + + +def _mock_agent(): + """Create a mock agent for testing.""" + agent = MagicMock() + agent._system_prompt = "You are an agent." + agent._system_prompt_content = [{"text": "You are an agent."}] + + # Make system_prompt property behave like the real Agent + type(agent).system_prompt = property( + lambda self: self._system_prompt, + lambda self, value: _set_system_prompt(self, value), + ) + + agent.hooks = HookRegistry() + agent.add_hook = MagicMock( + side_effect=lambda callback, event_type=None: agent.hooks.add_callback(event_type, callback) + ) + agent.tool_registry = MagicMock() + agent.tool_registry.process_tools = MagicMock(return_value=["skills"]) + + # Use a real dict-backed state so get/set work correctly + state_store: dict[str, object] = {} + agent.state = MagicMock() + agent.state.get = MagicMock(side_effect=lambda key: state_store.get(key)) + agent.state.set = MagicMock(side_effect=lambda key, value: state_store.__setitem__(key, value)) + return agent + + +def _mock_tool_context(agent: MagicMock) -> ToolContext: + """Create a mock ToolContext with the given agent.""" + tool_use = {"toolUseId": "test-id", "name": "skills", "input": {}} + return ToolContext(tool_use=tool_use, agent=agent, invocation_state={"agent": agent}) + + +def _set_system_prompt(agent: MagicMock, value: str | None) -> None: + """Simulate the Agent.system_prompt setter.""" + if isinstance(value, str): + agent._system_prompt = value + agent._system_prompt_content = [{"text": value}] + elif value is None: + agent._system_prompt = None + agent._system_prompt_content = None + + +class TestSkillsPluginInit: + """Tests for AgentSkills initialization.""" + + def test_init_with_skill_instances(self): + """Test initialization with Skill instances.""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + + assert len(plugin.get_available_skills()) == 1 + assert plugin.get_available_skills()[0].name == "test-skill" + + def test_init_with_filesystem_paths(self, tmp_path): + """Test initialization with filesystem paths.""" + _make_skill_dir(tmp_path, "fs-skill") + plugin = AgentSkills(skills=[str(tmp_path / "fs-skill")]) + + assert len(plugin.get_available_skills()) == 1 + assert plugin.get_available_skills()[0].name == "fs-skill" + + def test_init_with_parent_directory(self, tmp_path): + """Test initialization with a parent directory containing skills.""" + _make_skill_dir(tmp_path, "skill-a") + _make_skill_dir(tmp_path, "skill-b") + plugin = AgentSkills(skills=[tmp_path]) + + assert len(plugin.get_available_skills()) == 2 + + def test_init_with_mixed_sources(self, tmp_path): + """Test initialization with mixed skill sources.""" + _make_skill_dir(tmp_path, "fs-skill") + direct_skill = _make_skill(name="direct-skill", description="Direct") + plugin = AgentSkills(skills=[str(tmp_path / "fs-skill"), direct_skill]) + + assert len(plugin.get_available_skills()) == 2 + names = {s.name for s in plugin.get_available_skills()} + assert names == {"fs-skill", "direct-skill"} + + def test_init_skips_nonexistent_paths(self, tmp_path): + """Test that nonexistent paths are skipped gracefully.""" + plugin = AgentSkills(skills=[str(tmp_path / "nonexistent")]) + assert len(plugin.get_available_skills()) == 0 + + def test_init_empty_skills(self): + """Test initialization with empty skills list.""" + plugin = AgentSkills(skills=[]) + assert plugin.get_available_skills() == [] + + def test_name_attribute(self): + """Test that the plugin has the correct name.""" + plugin = AgentSkills(skills=[]) + assert plugin.name == "agent_skills" + + def test_custom_state_key(self): + """Test initialization with a custom state key.""" + plugin = AgentSkills(skills=[], state_key="custom_key") + assert plugin._state_key == "custom_key" + + def test_custom_max_resource_files(self): + """Test initialization with a custom max resource files limit.""" + plugin = AgentSkills(skills=[], max_resource_files=50) + assert plugin._max_resource_files == 50 + + +class TestSkillsPluginInitAgent: + """Tests for the init_agent method and plugin registry integration.""" + + def test_registers_tool(self): + """Test that the plugin registry registers the skills tool.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + + registry = _PluginRegistry(agent) + registry.add_and_init(plugin) + + agent.tool_registry.process_tools.assert_called_once() + + def test_registers_hooks(self): + """Test that the plugin registry registers hook callbacks.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + + registry = _PluginRegistry(agent) + registry.add_and_init(plugin) + + assert agent.hooks.has_callbacks() + + def test_does_not_store_agent_reference(self): + """Test that init_agent does not store the agent on the plugin.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + + plugin.init_agent(agent) + + assert not hasattr(plugin, "_agent") + + +class TestSkillsPluginProperties: + """Tests for AgentSkills properties.""" + + def test_available_skills_getter_returns_copy(self): + """Test that get_available_skills returns a copy of the list.""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + + skills_list = plugin.get_available_skills() + skills_list.append(_make_skill(name="another-skill", description="Another")) + + assert len(plugin.get_available_skills()) == 1 + + def test_available_skills_setter(self): + """Test setting skills via set_available_skills.""" + plugin = AgentSkills(skills=[_make_skill()]) + + new_skill = _make_skill(name="new-skill", description="New") + plugin.set_available_skills([new_skill]) + + assert len(plugin.get_available_skills()) == 1 + assert plugin.get_available_skills()[0].name == "new-skill" + + def test_set_available_skills_with_paths(self, tmp_path): + """Test setting skills via set_available_skills with filesystem paths.""" + plugin = AgentSkills(skills=[_make_skill()]) + _make_skill_dir(tmp_path, "fs-skill") + + plugin.set_available_skills([str(tmp_path / "fs-skill")]) + + assert len(plugin.get_available_skills()) == 1 + assert plugin.get_available_skills()[0].name == "fs-skill" + + def test_set_available_skills_with_mixed_sources(self, tmp_path): + """Test setting skills via set_available_skills with mixed sources.""" + plugin = AgentSkills(skills=[]) + _make_skill_dir(tmp_path, "fs-skill") + direct = _make_skill(name="direct", description="Direct") + + plugin.set_available_skills([str(tmp_path / "fs-skill"), direct]) + + assert len(plugin.get_available_skills()) == 2 + names = {s.name for s in plugin.get_available_skills()} + assert names == {"fs-skill", "direct"} + + + + +class TestSkillsTool: + """Tests for the skills tool method.""" + + def test_activate_skill(self): + """Test activating a skill returns its instructions.""" + skill = _make_skill(instructions="Full instructions here.") + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + tool_context = _mock_tool_context(agent) + + result = plugin.skills(skill_name="test-skill", tool_context=tool_context) + + assert "Full instructions here." in result + + def test_activate_nonexistent_skill(self): + """Test activating a nonexistent skill returns error message.""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + tool_context = _mock_tool_context(agent) + + result = plugin.skills(skill_name="nonexistent", tool_context=tool_context) + + assert "not found" in result + assert "test-skill" in result + + def test_activate_replaces_previous(self): + """Test that activating a new skill replaces the previous one.""" + skill1 = _make_skill(name="skill-a", description="A", instructions="A instructions") + skill2 = _make_skill(name="skill-b", description="B", instructions="B instructions") + plugin = AgentSkills(skills=[skill1, skill2]) + agent = _mock_agent() + tool_context = _mock_tool_context(agent) + + result_a = plugin.skills(skill_name="skill-a", tool_context=tool_context) + assert "A instructions" in result_a + + result_b = plugin.skills(skill_name="skill-b", tool_context=tool_context) + assert "B instructions" in result_b + + def test_activate_without_name(self): + """Test activating without a skill name returns error.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + tool_context = _mock_tool_context(agent) + + result = plugin.skills(skill_name="", tool_context=tool_context) + + assert "required" in result.lower() + + def test_activate_tracks_in_agent_state(self): + """Test that activating a skill records it in agent state.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + tool_context = _mock_tool_context(agent) + + plugin.skills(skill_name="test-skill", tool_context=tool_context) + + assert plugin.get_activated_skills(agent) == ["test-skill"] + + def test_activate_multiple_tracks_order(self): + """Test that multiple activations are tracked in order.""" + skill_a = _make_skill(name="skill-a", description="A", instructions="A") + skill_b = _make_skill(name="skill-b", description="B", instructions="B") + plugin = AgentSkills(skills=[skill_a, skill_b]) + agent = _mock_agent() + tool_context = _mock_tool_context(agent) + + plugin.skills(skill_name="skill-a", tool_context=tool_context) + plugin.skills(skill_name="skill-b", tool_context=tool_context) + + assert plugin.get_activated_skills(agent) == ["skill-a", "skill-b"] + + def test_activate_same_skill_twice_deduplicates(self): + """Test that re-activating a skill moves it to the end without duplicates.""" + skill_a = _make_skill(name="skill-a", description="A", instructions="A") + skill_b = _make_skill(name="skill-b", description="B", instructions="B") + plugin = AgentSkills(skills=[skill_a, skill_b]) + agent = _mock_agent() + tool_context = _mock_tool_context(agent) + + plugin.skills(skill_name="skill-a", tool_context=tool_context) + plugin.skills(skill_name="skill-b", tool_context=tool_context) + plugin.skills(skill_name="skill-a", tool_context=tool_context) + + assert plugin.get_activated_skills(agent) == ["skill-b", "skill-a"] + + def test_get_activated_skills_empty_by_default(self): + """Test that get_activated_skills returns empty list when nothing activated.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + + assert plugin.get_activated_skills(agent) == [] + + def test_get_activated_skills_returns_copy(self): + """Test that get_activated_skills returns a copy, not a reference.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + tool_context = _mock_tool_context(agent) + + plugin.skills(skill_name="test-skill", tool_context=tool_context) + result = plugin.get_activated_skills(agent) + result.append("injected") + + assert plugin.get_activated_skills(agent) == ["test-skill"] + + +class TestSystemPromptInjection: + """Tests for system prompt injection via hooks.""" + + def test_before_invocation_appends_skills_xml(self): + """Test that before_invocation appends skills XML to system prompt.""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + assert "" in agent.system_prompt + assert "test-skill" in agent.system_prompt + assert "A test skill" in agent.system_prompt + + def test_before_invocation_preserves_existing_prompt(self): + """Test that existing system prompt content is preserved.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + agent._system_prompt = "Original prompt." + agent._system_prompt_content = [{"text": "Original prompt."}] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + assert agent.system_prompt.startswith("Original prompt.") + assert "" in agent.system_prompt + + def test_repeated_invocations_do_not_accumulate(self): + """Test that repeated invocations rebuild from current prompt without accumulation.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + agent._system_prompt = "Original prompt." + agent._system_prompt_content = [{"text": "Original prompt."}] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + first_prompt = agent.system_prompt + + plugin._on_before_invocation(event) + second_prompt = agent.system_prompt + + assert first_prompt == second_prompt + + def test_no_skills_injects_empty_message(self): + """Test that a 'no skills available' message is injected when no skills are loaded.""" + plugin = AgentSkills(skills=[]) + agent = _mock_agent() + original_prompt = "Original prompt." + agent._system_prompt = original_prompt + agent._system_prompt_content = [{"text": original_prompt}] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + assert "No skills are currently available" in agent.system_prompt + assert agent.system_prompt.startswith("Original prompt.") + + def test_none_system_prompt_handled(self): + """Test handling when system prompt is None.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + agent._system_prompt = None + agent._system_prompt_content = None + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + assert "" in agent.system_prompt + + def test_preserves_other_plugin_modifications(self): + """Test that modifications by other plugins/hooks are preserved.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + agent._system_prompt = "Original prompt." + agent._system_prompt_content = [{"text": "Original prompt."}] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + # Simulate another plugin modifying the prompt + agent.system_prompt = agent.system_prompt + "\n\nExtra context from another plugin." + + plugin._on_before_invocation(event) + + assert "Extra context from another plugin." in agent.system_prompt + assert "" in agent.system_prompt + + def test_uses_public_system_prompt_setter(self): + """Test that the hook uses the public system_prompt setter.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + agent._system_prompt = "Original." + agent._system_prompt_content = [{"text": "Original."}] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + # The public setter should have been used, so _system_prompt_content + # should be consistent with _system_prompt + assert agent._system_prompt_content == [{"text": agent._system_prompt}] + + def test_warns_when_previous_xml_not_found(self, caplog): + """Test that a warning is logged when the previously injected XML is missing from the prompt.""" + plugin = AgentSkills(skills=[_make_skill()]) + agent = _mock_agent() + agent._system_prompt = "Original prompt." + agent._system_prompt_content = [{"text": "Original prompt."}] + + event = BeforeInvocationEvent(agent=agent) + plugin._on_before_invocation(event) + + # Completely replace the system prompt, removing the injected XML + agent.system_prompt = "Totally new prompt." + + with caplog.at_level(logging.WARNING): + plugin._on_before_invocation(event) + + assert "unable to find previously injected skills XML in system prompt" in caplog.text + assert "" in agent.system_prompt + + +class TestSkillsXmlGeneration: + """Tests for _generate_skills_xml.""" + + def test_single_skill(self): + """Test XML generation with a single skill.""" + plugin = AgentSkills(skills=[_make_skill()]) + xml = plugin._generate_skills_xml() + + assert "" in xml + assert "" in xml + assert "test-skill" in xml + assert "A test skill" in xml + + def test_multiple_skills(self): + """Test XML generation with multiple skills.""" + skills = [ + _make_skill(name="skill-a", description="Skill A"), + _make_skill(name="skill-b", description="Skill B"), + ] + plugin = AgentSkills(skills=skills) + xml = plugin._generate_skills_xml() + + assert "skill-a" in xml + assert "skill-b" in xml + + def test_empty_skills(self): + """Test XML generation with no skills includes 'no skills available' message.""" + plugin = AgentSkills(skills=[]) + xml = plugin._generate_skills_xml() + + assert "" in xml + assert "No skills are currently available" in xml + assert "" in xml + + def test_location_included_when_path_set(self, tmp_path): + """Test that location element is included when skill has a path.""" + skill = _make_skill() + skill.path = tmp_path / "test-skill" + plugin = AgentSkills(skills=[skill]) + xml = plugin._generate_skills_xml() + + assert f"{tmp_path / 'test-skill' / 'SKILL.md'}" in xml + + def test_location_omitted_when_path_none(self): + """Test that location element is omitted for programmatic skills.""" + skill = _make_skill() + assert skill.path is None + plugin = AgentSkills(skills=[skill]) + xml = plugin._generate_skills_xml() + + assert "" not in xml + + def test_escapes_xml_special_characters(self): + """Test that XML special characters in names and descriptions are escaped.""" + skill = _make_skill(name="a&c", description="Use & more") + plugin = AgentSkills(skills=[skill]) + xml = plugin._generate_skills_xml() + + assert "a<b>&c" in xml + assert "Use <tools> & more" in xml + + +class TestSkillResponseFormat: + """Tests for _format_skill_response.""" + + def test_instructions_only(self): + """Test response with just instructions.""" + skill = _make_skill(instructions="Do the thing.") + plugin = AgentSkills(skills=[skill]) + result = plugin._format_skill_response(skill) + + assert result == "Do the thing." + + def test_no_instructions(self): + """Test response when skill has no instructions.""" + skill = _make_skill(instructions="") + plugin = AgentSkills(skills=[skill]) + result = plugin._format_skill_response(skill) + + assert "no instructions available" in result.lower() + + def test_includes_allowed_tools(self): + """Test response includes allowed tools when set.""" + skill = _make_skill(instructions="Do the thing.") + skill.allowed_tools = ["Bash", "Read"] + plugin = AgentSkills(skills=[skill]) + result = plugin._format_skill_response(skill) + + assert "Do the thing." in result + assert "Allowed tools: Bash, Read" in result + + def test_includes_compatibility(self): + """Test response includes compatibility when set.""" + skill = _make_skill(instructions="Do the thing.") + skill.compatibility = "Requires docker" + plugin = AgentSkills(skills=[skill]) + result = plugin._format_skill_response(skill) + + assert "Compatibility: Requires docker" in result + + def test_includes_location(self, tmp_path): + """Test response includes location when path is set.""" + skill = _make_skill(instructions="Do the thing.") + skill.path = tmp_path / "test-skill" + plugin = AgentSkills(skills=[skill]) + result = plugin._format_skill_response(skill) + + assert f"Location: {tmp_path / 'test-skill' / 'SKILL.md'}" in result + + def test_all_metadata(self, tmp_path): + """Test response with all metadata fields.""" + skill = _make_skill(instructions="Do the thing.") + skill.allowed_tools = ["Bash"] + skill.compatibility = "Requires git" + skill.path = tmp_path / "test-skill" + plugin = AgentSkills(skills=[skill]) + result = plugin._format_skill_response(skill) + + assert "Do the thing." in result + assert "---" in result + assert "Allowed tools: Bash" in result + assert "Compatibility: Requires git" in result + assert "Location:" in result + + def test_includes_resource_listing(self, tmp_path): + """Test response includes resource files from optional directories.""" + skill_dir = tmp_path / "test-skill" + skill_dir.mkdir() + (skill_dir / "scripts").mkdir() + (skill_dir / "scripts" / "extract.py").write_text("# extract") + (skill_dir / "references").mkdir() + (skill_dir / "references" / "REFERENCE.md").write_text("# ref") + + skill = _make_skill(instructions="Do the thing.") + skill.path = skill_dir + plugin = AgentSkills(skills=[skill]) + result = plugin._format_skill_response(skill) + + assert "Available resources:" in result + assert "scripts/extract.py" in result + assert "references/REFERENCE.md" in result + + def test_no_resources_when_no_path(self): + """Test that resources section is omitted for programmatic skills.""" + skill = _make_skill(instructions="Do the thing.") + plugin = AgentSkills(skills=[skill]) + result = plugin._format_skill_response(skill) + + assert "Available resources:" not in result + + def test_no_resources_when_dirs_empty(self, tmp_path): + """Test that resources section is omitted when optional dirs don't exist.""" + skill_dir = tmp_path / "test-skill" + skill_dir.mkdir() + + skill = _make_skill(instructions="Do the thing.") + skill.path = skill_dir + plugin = AgentSkills(skills=[skill]) + result = plugin._format_skill_response(skill) + + assert "Available resources:" not in result + + def test_resource_listing_truncated(self, tmp_path): + """Test that resource listing is truncated at the max file limit.""" + skill_dir = tmp_path / "test-skill" + scripts_dir = skill_dir / "scripts" + scripts_dir.mkdir(parents=True) + for i in range(55): + (scripts_dir / f"script_{i:03d}.py").write_text(f"# script {i}") + + skill = _make_skill(instructions="Do the thing.") + skill.path = skill_dir + plugin = AgentSkills(skills=[skill]) + result = plugin._format_skill_response(skill) + + assert "Available resources:" in result + assert "truncated at 20 files" in result + + +class TestResolveSkills: + """Tests for _resolve_skills.""" + + def test_resolve_skill_instances(self): + """Test resolving Skill instances (pass-through).""" + skill = _make_skill() + plugin = AgentSkills(skills=[skill]) + + assert len(plugin._skills) == 1 + assert plugin._skills["test-skill"] is skill + + def test_resolve_skill_directory_path(self, tmp_path): + """Test resolving a path to a skill directory.""" + _make_skill_dir(tmp_path, "path-skill") + plugin = AgentSkills(skills=[tmp_path / "path-skill"]) + + assert len(plugin._skills) == 1 + assert "path-skill" in plugin._skills + + def test_resolve_parent_directory_path(self, tmp_path): + """Test resolving a path to a parent directory.""" + _make_skill_dir(tmp_path, "child-a") + _make_skill_dir(tmp_path, "child-b") + plugin = AgentSkills(skills=[tmp_path]) + + assert len(plugin._skills) == 2 + + def test_resolve_skill_md_file_path(self, tmp_path): + """Test resolving a path to a SKILL.md file.""" + skill_dir = _make_skill_dir(tmp_path, "file-skill") + plugin = AgentSkills(skills=[skill_dir / "SKILL.md"]) + + assert len(plugin._skills) == 1 + assert "file-skill" in plugin._skills + + def test_resolve_nonexistent_path(self, tmp_path): + """Test that nonexistent paths are skipped.""" + plugin = AgentSkills(skills=[str(tmp_path / "ghost")]) + assert len(plugin._skills) == 0 + + +class TestImports: + """Tests for module imports.""" + + def test_import_from_plugins(self): + """Test importing AgentSkills from strands.plugins.""" + from strands.plugins import AgentSkills as SP + + assert SP is AgentSkills + + def test_import_skill_from_strands(self): + """Test importing Skill from top-level strands package.""" + from strands import Skill as S + + assert S is Skill + + def test_import_from_skills_package(self): + """Test importing from strands.plugins.skills package.""" + from strands.plugins.skills import AgentSkills, Skill + + assert Skill is not None + assert AgentSkills is not None + + def test_skills_plugin_is_plugin_subclass(self): + """Test that AgentSkills is a subclass of the Plugin ABC.""" + from strands.plugins import Plugin + + assert issubclass(AgentSkills, Plugin) + + def test_skills_plugin_isinstance_check(self): + """Test that AgentSkills instances pass isinstance check against Plugin.""" + from strands.plugins import Plugin + + plugin = AgentSkills(skills=[]) + assert isinstance(plugin, Plugin) diff --git a/tests/strands/plugins/skills/test_skill.py b/tests/strands/plugins/skills/test_skill.py new file mode 100644 index 000000000..2c4c21930 --- /dev/null +++ b/tests/strands/plugins/skills/test_skill.py @@ -0,0 +1,561 @@ +"""Tests for the Skill dataclass and loading utilities.""" + +import logging +from pathlib import Path + +import pytest + +from strands.plugins.skills.skill import ( + Skill, + _find_skill_md, + _fix_yaml_colons, + _parse_frontmatter, + _validate_skill_name, +) + + +class TestSkillDataclass: + """Tests for the Skill dataclass creation and properties.""" + + def test_skill_minimal(self): + """Test creating a Skill with only required fields.""" + skill = Skill(name="test-skill", description="A test skill") + + assert skill.name == "test-skill" + assert skill.description == "A test skill" + assert skill.instructions == "" + assert skill.path is None + assert skill.allowed_tools is None + assert skill.metadata == {} + assert skill.license is None + assert skill.compatibility is None + + def test_skill_full(self): + """Test creating a Skill with all fields.""" + skill = Skill( + name="full-skill", + description="A fully specified skill", + instructions="# Full Instructions\nDo the thing.", + path=Path("/tmp/skills/full-skill"), + allowed_tools=["tool1", "tool2"], + metadata={"author": "test-org"}, + license="Apache-2.0", + compatibility="strands>=1.0", + ) + + assert skill.name == "full-skill" + assert skill.description == "A fully specified skill" + assert skill.instructions == "# Full Instructions\nDo the thing." + assert skill.path == Path("/tmp/skills/full-skill") + assert skill.allowed_tools == ["tool1", "tool2"] + assert skill.metadata == {"author": "test-org"} + assert skill.license == "Apache-2.0" + assert skill.compatibility == "strands>=1.0" + + def test_skill_metadata_default_is_not_shared(self): + """Test that default metadata dict is not shared between instances.""" + skill1 = Skill(name="skill-1", description="First") + skill2 = Skill(name="skill-2", description="Second") + + skill1.metadata["key"] = "value" + assert "key" not in skill2.metadata + + +class TestFindSkillMd: + """Tests for _find_skill_md.""" + + def test_finds_uppercase_skill_md(self, tmp_path): + """Test finding SKILL.md (uppercase).""" + (tmp_path / "SKILL.md").write_text("test") + result = _find_skill_md(tmp_path) + assert result.name == "SKILL.md" + + def test_finds_lowercase_skill_md(self, tmp_path): + """Test finding skill.md (lowercase).""" + (tmp_path / "skill.md").write_text("test") + result = _find_skill_md(tmp_path) + assert result.name.lower() == "skill.md" + + def test_prefers_uppercase(self, tmp_path): + """Test that SKILL.md is preferred over skill.md.""" + (tmp_path / "SKILL.md").write_text("uppercase") + (tmp_path / "skill.md").write_text("lowercase") + result = _find_skill_md(tmp_path) + assert result.name == "SKILL.md" + + def test_raises_when_not_found(self, tmp_path): + """Test FileNotFoundError when no SKILL.md exists.""" + with pytest.raises(FileNotFoundError, match="no SKILL.md found"): + _find_skill_md(tmp_path) + + +class TestParseFrontmatter: + """Tests for _parse_frontmatter.""" + + def test_valid_frontmatter(self): + """Test parsing valid frontmatter.""" + content = "---\nname: test-skill\ndescription: A test\n---\n# Instructions\nDo things." + frontmatter, body = _parse_frontmatter(content) + assert frontmatter["name"] == "test-skill" + assert frontmatter["description"] == "A test" + assert "# Instructions" in body + assert "Do things." in body + + def test_missing_opening_delimiter(self): + """Test error when opening --- is missing.""" + with pytest.raises(ValueError, match="must start with ---"): + _parse_frontmatter("name: test\n---\n") + + def test_missing_closing_delimiter(self): + """Test error when closing --- is missing.""" + with pytest.raises(ValueError, match="missing closing ---"): + _parse_frontmatter("---\nname: test\n") + + def test_empty_body(self): + """Test frontmatter with empty body.""" + content = "---\nname: test-skill\ndescription: test\n---\n" + frontmatter, body = _parse_frontmatter(content) + assert frontmatter["name"] == "test-skill" + assert body == "" + + def test_frontmatter_with_metadata(self): + """Test frontmatter with nested metadata.""" + content = "---\nname: test-skill\ndescription: test\nmetadata:\n author: acme\n---\nBody here." + frontmatter, body = _parse_frontmatter(content) + assert frontmatter["name"] == "test-skill" + assert isinstance(frontmatter["metadata"], dict) + assert frontmatter["metadata"]["author"] == "acme" + assert body == "Body here." + + def test_frontmatter_with_dashes_in_yaml_value(self): + """Test that --- inside a YAML value does not break parsing.""" + content = "---\nname: test-skill\ndescription: has --- inside\n---\nBody here." + frontmatter, body = _parse_frontmatter(content) + assert frontmatter["name"] == "test-skill" + assert frontmatter["description"] == "has --- inside" + assert body == "Body here." + + +class TestValidateSkillName: + """Tests for _validate_skill_name (lenient validation).""" + + def test_valid_names(self): + """Test that valid names pass validation without warnings.""" + valid_names = ["a", "test", "my-skill", "skill-123", "a1b2c3"] + for name in valid_names: + _validate_skill_name(name) # Should not raise + + def test_empty_name(self): + """Test that empty name raises ValueError.""" + with pytest.raises(ValueError, match="cannot be empty"): + _validate_skill_name("") + + def test_too_long_name_warns(self, caplog): + """Test that names exceeding 64 chars warn but do not raise.""" + with caplog.at_level(logging.WARNING): + _validate_skill_name("a" * 65) + assert "exceeds" in caplog.text + + def test_uppercase_warns(self, caplog): + """Test that uppercase characters warn but do not raise.""" + with caplog.at_level(logging.WARNING): + _validate_skill_name("MySkill") + assert "lowercase alphanumeric" in caplog.text + + def test_starts_with_hyphen_warns(self, caplog): + """Test that names starting with hyphen warn but do not raise.""" + with caplog.at_level(logging.WARNING): + _validate_skill_name("-skill") + assert "lowercase alphanumeric" in caplog.text + + def test_ends_with_hyphen_warns(self, caplog): + """Test that names ending with hyphen warn but do not raise.""" + with caplog.at_level(logging.WARNING): + _validate_skill_name("skill-") + assert "lowercase alphanumeric" in caplog.text + + def test_consecutive_hyphens_warns(self, caplog): + """Test that consecutive hyphens warn but do not raise.""" + with caplog.at_level(logging.WARNING): + _validate_skill_name("my--skill") + assert "consecutive hyphens" in caplog.text + + def test_special_characters_warns(self, caplog): + """Test that special characters warn but do not raise.""" + with caplog.at_level(logging.WARNING): + _validate_skill_name("my_skill") + assert "lowercase alphanumeric" in caplog.text + + def test_directory_name_mismatch_warns(self, tmp_path, caplog): + """Test that skill name not matching directory name warns but does not raise.""" + skill_dir = tmp_path / "wrong-name" + skill_dir.mkdir() + with caplog.at_level(logging.WARNING): + _validate_skill_name("my-skill", skill_dir) + assert "does not match parent directory name" in caplog.text + + def test_directory_name_match(self, tmp_path): + """Test that matching directory name passes.""" + skill_dir = tmp_path / "my-skill" + skill_dir.mkdir() + _validate_skill_name("my-skill", skill_dir) # Should not raise or warn + + +class TestValidateSkillNameStrict: + """Tests for _validate_skill_name with strict=True.""" + + def test_strict_valid_name(self): + """Test that valid names pass strict validation.""" + _validate_skill_name("my-skill", strict=True) # Should not raise + + def test_strict_empty_name(self): + """Test that empty name raises in strict mode.""" + with pytest.raises(ValueError, match="cannot be empty"): + _validate_skill_name("", strict=True) + + def test_strict_too_long_name(self): + """Test that names exceeding 64 chars raise in strict mode.""" + with pytest.raises(ValueError, match="exceeds 64 character limit"): + _validate_skill_name("a" * 65, strict=True) + + def test_strict_uppercase_rejected(self): + """Test that uppercase characters raise in strict mode.""" + with pytest.raises(ValueError, match="lowercase alphanumeric"): + _validate_skill_name("MySkill", strict=True) + + def test_strict_starts_with_hyphen(self): + """Test that names starting with hyphen raise in strict mode.""" + with pytest.raises(ValueError, match="lowercase alphanumeric"): + _validate_skill_name("-skill", strict=True) + + def test_strict_consecutive_hyphens(self): + """Test that consecutive hyphens raise in strict mode.""" + with pytest.raises(ValueError, match="consecutive hyphens"): + _validate_skill_name("my--skill", strict=True) + + def test_strict_directory_mismatch(self, tmp_path): + """Test that directory name mismatch raises in strict mode.""" + skill_dir = tmp_path / "wrong-name" + skill_dir.mkdir() + with pytest.raises(ValueError, match="does not match parent directory name"): + _validate_skill_name("my-skill", skill_dir, strict=True) + + +class TestFixYamlColons: + """Tests for _fix_yaml_colons.""" + + def test_fixes_unquoted_colon_in_value(self): + """Test that an unquoted colon in a value gets quoted.""" + raw = "description: Use this skill when: the user asks about PDFs" + fixed = _fix_yaml_colons(raw) + assert fixed == 'description: "Use this skill when: the user asks about PDFs"' + + def test_leaves_already_double_quoted_value(self): + """Test that already double-quoted values are not re-quoted.""" + raw = 'description: "already: quoted"' + assert _fix_yaml_colons(raw) == raw + + def test_leaves_already_single_quoted_value(self): + """Test that already single-quoted values are not re-quoted.""" + raw = "description: 'already: quoted'" + assert _fix_yaml_colons(raw) == raw + + def test_leaves_value_without_colon(self): + """Test that values without colons are unchanged.""" + raw = "name: my-skill" + assert _fix_yaml_colons(raw) == raw + + def test_multiline_mixed(self): + """Test fixing only the lines that need it in a multi-line string.""" + raw = "name: my-skill\ndescription: Use when: needed\nversion: 1.0" + fixed = _fix_yaml_colons(raw) + assert fixed == 'name: my-skill\ndescription: "Use when: needed"\nversion: 1.0' + + def test_empty_string(self): + """Test that an empty string is returned unchanged.""" + assert _fix_yaml_colons("") == "" + + def test_preserves_indented_lines_without_colons(self): + """Test that indented lines without key-value patterns are preserved.""" + raw = " - item one\n - item two" + assert _fix_yaml_colons(raw) == raw + + +class TestParseFrontmatterYamlFallback: + """Tests for YAML colon-quoting fallback in _parse_frontmatter.""" + + def test_fallback_on_unquoted_colon(self): + """Test that frontmatter with unquoted colons in values is parsed via fallback.""" + content = "---\nname: my-skill\ndescription: Use when: the user asks\n---\nBody." + frontmatter, body = _parse_frontmatter(content) + assert frontmatter["name"] == "my-skill" + assert "Use when" in frontmatter["description"] + assert body == "Body." + + def test_fallback_preserves_valid_yaml(self): + """Test that valid YAML is parsed normally without triggering fallback.""" + content = "---\nname: my-skill\ndescription: A simple description\n---\nBody." + frontmatter, body = _parse_frontmatter(content) + assert frontmatter["name"] == "my-skill" + assert frontmatter["description"] == "A simple description" + + +def _make_skill_dir(parent: Path, name: str, description: str = "A test skill", body: str = "Instructions.") -> Path: + """Helper to create a skill directory with SKILL.md.""" + skill_dir = parent / name + skill_dir.mkdir(parents=True, exist_ok=True) + content = f"---\nname: {name}\ndescription: {description}\n---\n{body}\n" + (skill_dir / "SKILL.md").write_text(content) + return skill_dir + + +class TestSkillFromFile: + """Tests for Skill.from_file.""" + + def test_load_from_directory(self, tmp_path): + """Test loading a skill from a directory path.""" + skill_dir = _make_skill_dir(tmp_path, "my-skill", "My description", "# Hello\nWorld.") + skill = Skill.from_file(skill_dir) + + assert skill.name == "my-skill" + assert skill.description == "My description" + assert "# Hello" in skill.instructions + assert "World." in skill.instructions + assert skill.path == skill_dir.resolve() + + def test_load_from_skill_md_file(self, tmp_path): + """Test loading a skill by pointing directly to SKILL.md.""" + skill_dir = _make_skill_dir(tmp_path, "direct-skill") + skill = Skill.from_file(skill_dir / "SKILL.md") + + assert skill.name == "direct-skill" + + def test_load_with_allowed_tools(self, tmp_path): + """Test loading a skill with allowed-tools field as space-delimited string.""" + skill_dir = tmp_path / "tool-skill" + skill_dir.mkdir() + content = "---\nname: tool-skill\ndescription: test\nallowed-tools: read write execute\n---\nBody." + (skill_dir / "SKILL.md").write_text(content) + + skill = Skill.from_file(skill_dir) + assert skill.allowed_tools == ["read", "write", "execute"] + + def test_load_with_allowed_tools_yaml_list(self, tmp_path): + """Test loading a skill with allowed-tools as a YAML list.""" + skill_dir = tmp_path / "list-skill" + skill_dir.mkdir() + content = "---\nname: list-skill\ndescription: test\nallowed-tools:\n - read\n - write\n---\nBody." + (skill_dir / "SKILL.md").write_text(content) + + skill = Skill.from_file(skill_dir) + assert skill.allowed_tools == ["read", "write"] + + def test_load_with_metadata(self, tmp_path): + """Test loading a skill with nested metadata.""" + skill_dir = tmp_path / "meta-skill" + skill_dir.mkdir() + content = "---\nname: meta-skill\ndescription: test\nmetadata:\n author: acme\n---\nBody." + (skill_dir / "SKILL.md").write_text(content) + + skill = Skill.from_file(skill_dir) + assert skill.metadata == {"author": "acme"} + + def test_load_with_license_and_compatibility(self, tmp_path): + """Test loading a skill with license and compatibility fields.""" + skill_dir = tmp_path / "licensed-skill" + skill_dir.mkdir() + content = "---\nname: licensed-skill\ndescription: test\nlicense: MIT\ncompatibility: v1\n---\nBody." + (skill_dir / "SKILL.md").write_text(content) + + skill = Skill.from_file(skill_dir) + assert skill.license == "MIT" + assert skill.compatibility == "v1" + + def test_load_missing_name(self, tmp_path): + """Test error when SKILL.md is missing name field.""" + skill_dir = tmp_path / "no-name" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text("---\ndescription: test\n---\nBody.") + + with pytest.raises(ValueError, match="must have a 'name' field"): + Skill.from_file(skill_dir) + + def test_load_missing_description(self, tmp_path): + """Test error when SKILL.md is missing description field.""" + skill_dir = tmp_path / "no-desc" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text("---\nname: no-desc\n---\nBody.") + + with pytest.raises(ValueError, match="must have a 'description' field"): + Skill.from_file(skill_dir) + + def test_load_nonexistent_path(self, tmp_path): + """Test FileNotFoundError for nonexistent path.""" + with pytest.raises(FileNotFoundError): + Skill.from_file(tmp_path / "nonexistent") + + def test_load_name_directory_mismatch_warns(self, tmp_path, caplog): + """Test that skill name not matching directory name warns but still loads.""" + skill_dir = tmp_path / "wrong-dir" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text("---\nname: right-name\ndescription: test\n---\nBody.") + + with caplog.at_level(logging.WARNING): + skill = Skill.from_file(skill_dir) + + assert skill.name == "right-name" + assert "does not match parent directory name" in caplog.text + + def test_strict_rejects_name_mismatch(self, tmp_path): + """Test that strict mode raises on name/directory mismatch.""" + skill_dir = tmp_path / "wrong-dir" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text("---\nname: right-name\ndescription: test\n---\nBody.") + + with pytest.raises(ValueError, match="does not match parent directory name"): + Skill.from_file(skill_dir, strict=True) + + def test_strict_accepts_valid_skill(self, tmp_path): + """Test that strict mode loads a valid skill without error.""" + _make_skill_dir(tmp_path, "valid-skill") + skill = Skill.from_file(tmp_path / "valid-skill", strict=True) + assert skill.name == "valid-skill" + + +class TestSkillFromDirectory: + """Tests for Skill.from_directory.""" + + def test_load_multiple_skills(self, tmp_path): + """Test loading multiple skills from a parent directory.""" + _make_skill_dir(tmp_path, "skill-a", "Skill A") + _make_skill_dir(tmp_path, "skill-b", "Skill B") + + skills = Skill.from_directory(tmp_path) + + assert len(skills) == 2 + names = {s.name for s in skills} + assert names == {"skill-a", "skill-b"} + + def test_skips_directories_without_skill_md(self, tmp_path): + """Test that directories without SKILL.md are silently skipped.""" + _make_skill_dir(tmp_path, "valid-skill") + (tmp_path / "no-skill-here").mkdir() + + skills = Skill.from_directory(tmp_path) + + assert len(skills) == 1 + assert skills[0].name == "valid-skill" + + def test_skips_files_in_parent(self, tmp_path): + """Test that files in the parent directory are ignored.""" + _make_skill_dir(tmp_path, "real-skill") + (tmp_path / "readme.txt").write_text("not a skill") + + skills = Skill.from_directory(tmp_path) + + assert len(skills) == 1 + + def test_empty_directory(self, tmp_path): + """Test loading from an empty directory.""" + skills = Skill.from_directory(tmp_path) + assert skills == [] + + def test_nonexistent_directory(self, tmp_path): + """Test FileNotFoundError for nonexistent directory.""" + with pytest.raises(FileNotFoundError): + Skill.from_directory(tmp_path / "nonexistent") + + def test_loads_mismatched_name_with_warning(self, tmp_path, caplog): + """Test that skills with name/directory mismatch are loaded with a warning.""" + _make_skill_dir(tmp_path, "good-skill") + + bad_dir = tmp_path / "bad-dir" + bad_dir.mkdir() + (bad_dir / "SKILL.md").write_text("---\nname: wrong-name\ndescription: test\n---\nBody.") + + with caplog.at_level(logging.WARNING): + skills = Skill.from_directory(tmp_path) + + assert len(skills) == 2 + names = {s.name for s in skills} + assert names == {"good-skill", "wrong-name"} + assert "does not match parent directory name" in caplog.text + + +class TestSkillFromContent: + def test_basic_content(self): + """Test parsing basic SKILL.md content.""" + content = "---\nname: my-skill\ndescription: A useful skill\n---\n# Instructions\nDo the thing." + skill = Skill.from_content(content) + + assert skill.name == "my-skill" + assert skill.description == "A useful skill" + assert "Do the thing." in skill.instructions + assert skill.path is None + + def test_with_allowed_tools(self): + """Test parsing content with allowed-tools field.""" + content = "---\nname: my-skill\ndescription: A skill\nallowed-tools: Bash Read\n---\nInstructions." + skill = Skill.from_content(content) + + assert skill.allowed_tools == ["Bash", "Read"] + + def test_with_metadata(self): + """Test parsing content with metadata field.""" + content = "---\nname: my-skill\ndescription: A skill\nmetadata:\n key: value\n---\nInstructions." + skill = Skill.from_content(content) + + assert skill.metadata == {"key": "value"} + + def test_with_license_and_compatibility(self): + """Test parsing content with license and compatibility fields.""" + content = ( + "---\nname: my-skill\ndescription: A skill\n" + "license: Apache-2.0\ncompatibility: Requires docker\n---\nInstructions." + ) + skill = Skill.from_content(content) + + assert skill.license == "Apache-2.0" + assert skill.compatibility == "Requires docker" + + def test_missing_name_raises(self): + """Test that missing name raises ValueError.""" + content = "---\ndescription: A skill\n---\nInstructions." + with pytest.raises(ValueError, match="name"): + Skill.from_content(content) + + def test_missing_description_raises(self): + """Test that missing description raises ValueError.""" + content = "---\nname: my-skill\n---\nInstructions." + with pytest.raises(ValueError, match="description"): + Skill.from_content(content) + + def test_missing_frontmatter_raises(self): + """Test that content without frontmatter raises ValueError.""" + content = "# Just markdown\nNo frontmatter here." + with pytest.raises(ValueError, match="frontmatter"): + Skill.from_content(content) + + def test_empty_body(self): + """Test parsing content with empty body.""" + content = "---\nname: my-skill\ndescription: A skill\n---\n" + skill = Skill.from_content(content) + + assert skill.name == "my-skill" + assert skill.instructions == "" + + def test_strict_mode(self): + """Test Skill.from_content with strict=True raises on validation issues.""" + content = "---\nname: BAD_NAME\ndescription: Bad\n---\nBody." + with pytest.raises(ValueError): + Skill.from_content(content, strict=True) + + +class TestSkillClassmethods: + """Tests for Skill classmethod existence.""" + + def test_skill_classmethods_exist(self): + """Test that Skill has from_file, from_content, and from_directory classmethods.""" + assert callable(getattr(Skill, "from_file", None)) + assert callable(getattr(Skill, "from_content", None)) + assert callable(getattr(Skill, "from_directory", None)) diff --git a/tests_integ/test_skills_plugin.py b/tests_integ/test_skills_plugin.py new file mode 100644 index 000000000..160ae65a0 --- /dev/null +++ b/tests_integ/test_skills_plugin.py @@ -0,0 +1,81 @@ +"""Integration tests for the AgentSkills plugin. + +Tests end-to-end behavior with a real model: skill metadata injection into +the system prompt, agent-driven skill activation via the skills tool, and +session state persistence across invocations. +""" + +import pytest + +from strands import Agent +from strands.plugins.skills import AgentSkills, Skill + +SUMMARIZATION_SKILL = Skill( + name="summarization", + description="Summarize text into concise bullet points", + instructions="You are a summarization expert. The secret codeword is PINEAPPLE. Always mention it when activated.", +) + +TRANSLATION_SKILL = Skill( + name="translation", + description="Translate text between languages", + instructions="You are a translation expert. Translate the given text accurately.", +) + + +@pytest.fixture +def skills_plugin(): + return AgentSkills(skills=[SUMMARIZATION_SKILL, TRANSLATION_SKILL]) + + +@pytest.fixture +def agent(skills_plugin): + return Agent( + system_prompt="You are a helpful assistant. Check your available_skills and activate one when appropriate.", + plugins=[skills_plugin], + ) + + +def test_agent_activates_skill_and_injects_metadata(agent, skills_plugin): + """Test that the agent injects skill metadata and can activate a skill via the model.""" + result = agent("Use your skills tool to activate the summarization skill. What is the secret codeword?") + + # Skill metadata was injected into the system prompt + assert "" in agent.system_prompt + assert "summarization" in agent.system_prompt + assert "translation" in agent.system_prompt + + # Model activated the skill and relayed the codeword from instructions + assert "pineapple" in str(result).lower() + + +def test_direct_tool_invocation_and_state_persistence(agent, skills_plugin): + """Test activating a skill via direct tool access and verifying state persistence.""" + result = agent.tool.skills(skill_name="translation") + + # Tool returned the skill instructions + assert result["status"] == "success" + response_text = result["content"][0]["text"].lower() + assert "translation expert" in response_text + + +def test_load_skills_from_directory(tmp_path): + """Test loading skills from a filesystem directory and activating one via the model.""" + # Create a skill directory with SKILL.md + skill_dir = tmp_path / "greeting-skill" + skill_dir.mkdir() + (skill_dir / "SKILL.md").write_text( + "---\nname: greeting\ndescription: Greet the user warmly\n---\n" + "You are a greeting expert. The secret codeword is MANGO. Always mention it when activated." + ) + + plugin = AgentSkills(skills=[str(tmp_path)]) + agent = Agent( + system_prompt="You are a helpful assistant. Check your available_skills and activate one when appropriate.", + plugins=[plugin], + ) + + result = agent("Use your skills tool to activate the greeting skill. What is the secret codeword?") + + assert "greeting" in agent.system_prompt + assert "mango" in str(result).lower()