diff --git a/src/strands/vended_plugins/skills/agent_skills.py b/src/strands/vended_plugins/skills/agent_skills.py index 5e42b9230..23217e81c 100644 --- a/src/strands/vended_plugins/skills/agent_skills.py +++ b/src/strands/vended_plugins/skills/agent_skills.py @@ -86,6 +86,7 @@ def __init__( - 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 + - An ``https://`` URL pointing directly to raw SKILL.md content 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. @@ -176,8 +177,9 @@ 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. + skill directory (containing SKILL.md), a ``str`` or ``Path`` to a + parent directory containing skill subdirectories, or an ``https://`` + URL pointing directly to raw SKILL.md content. 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 @@ -284,7 +286,8 @@ 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. + a path to a parent directory containing multiple skills, or an + HTTPS URL pointing to a SKILL.md file. Args: sources: List of skill sources to resolve. @@ -299,6 +302,14 @@ def _resolve_skills(self, sources: list[SkillSource]) -> dict[str, Skill]: if source.name in resolved: logger.warning("name=<%s> | duplicate skill name, overwriting previous skill", source.name) resolved[source.name] = source + elif isinstance(source, str) and source.startswith("https://"): + try: + skill = Skill.from_url(source, 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 (RuntimeError, ValueError) as e: + logger.warning("url=<%s> | failed to load skill from URL: %s", source, e) else: path = Path(source).resolve() if not path.exists(): diff --git a/src/strands/vended_plugins/skills/skill.py b/src/strands/vended_plugins/skills/skill.py index 3e1b6bba5..a60c1cd6c 100644 --- a/src/strands/vended_plugins/skills/skill.py +++ b/src/strands/vended_plugins/skills/skill.py @@ -1,15 +1,17 @@ """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. +discovering, parsing, and loading skills from the filesystem, raw content, +or HTTPS URLs. Skills are directories containing a SKILL.md file with YAML +frontmatter metadata and markdown instructions. """ from __future__ import annotations import logging import re +import urllib.error +import urllib.request from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -222,6 +224,9 @@ class Skill: # Load all skills from a parent directory skills = Skill.from_directory("./skills/") + # From an HTTPS URL + skill = Skill.from_url("https://example.com/SKILL.md") + Attributes: name: Unique identifier for the skill (1-64 chars, lowercase alphanumeric + hyphens). description: Human-readable description of what the skill does. @@ -333,6 +338,48 @@ def from_content(cls, content: str, *, strict: bool = False) -> Skill: return _build_skill_from_frontmatter(frontmatter, body) + @classmethod + def from_url(cls, url: str, *, strict: bool = False) -> Skill: + """Load a skill by fetching its SKILL.md content from an HTTPS URL. + + Fetches the raw SKILL.md content over HTTPS and parses it using + :meth:`from_content`. The URL must point directly to the raw + file content (not an HTML page). + + Example:: + + skill = Skill.from_url( + "https://raw.githubusercontent.com/org/repo/main/SKILL.md" + ) + + Args: + url: An ``https://`` URL pointing directly to raw SKILL.md content. + strict: If True, raise on any validation issue. If False (default), + warn and load anyway. + + Returns: + A Skill instance populated from the fetched SKILL.md content. + + Raises: + ValueError: If ``url`` is not an ``https://`` URL. + RuntimeError: If the SKILL.md content cannot be fetched. + """ + if not url.startswith("https://"): + raise ValueError(f"url=<{url}> | not a valid HTTPS URL") + + logger.info("url=<%s> | fetching skill content", url) + + try: + req = urllib.request.Request(url, headers={"User-Agent": "strands-agents-sdk"}) # noqa: S310 + with urllib.request.urlopen(req, timeout=30) as response: # noqa: S310 + content: str = response.read().decode("utf-8") + except urllib.error.HTTPError as e: + raise RuntimeError(f"url=<{url}> | HTTP {e.code}: {e.reason}") from e + except urllib.error.URLError as e: + raise RuntimeError(f"url=<{url}> | failed to fetch skill: {e.reason}") from e + + return cls.from_content(content, strict=strict) + @classmethod def from_directory(cls, skills_dir: str | Path, *, strict: bool = False) -> list[Skill]: """Load all skills from a parent directory containing skill subdirectories. diff --git a/tests/strands/vended_plugins/skills/test_agent_skills.py b/tests/strands/vended_plugins/skills/test_agent_skills.py index 52802a6c1..db82355a9 100644 --- a/tests/strands/vended_plugins/skills/test_agent_skills.py +++ b/tests/strands/vended_plugins/skills/test_agent_skills.py @@ -661,6 +661,95 @@ def test_resolve_nonexistent_path(self, tmp_path): assert len(plugin._skills) == 0 +class TestResolveUrlSkills: + """Tests for _resolve_skills with URL sources.""" + + _SKILL_MODULE = "strands.vended_plugins.skills.skill" + _SAMPLE_CONTENT = "---\nname: url-skill\ndescription: A URL skill\n---\n# Instructions\n" + + def _mock_urlopen(self, content): + """Create a mock urlopen context manager returning the given content.""" + mock_response = MagicMock() + mock_response.read.return_value = content.encode("utf-8") + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + return mock_response + + def test_resolve_url_source(self): + """Test resolving a URL string as a skill source.""" + from unittest.mock import patch + + with patch( + f"{self._SKILL_MODULE}.urllib.request.urlopen", return_value=self._mock_urlopen(self._SAMPLE_CONTENT) + ): + plugin = AgentSkills(skills=["https://example.com/SKILL.md"]) + + assert len(plugin.get_available_skills()) == 1 + assert plugin.get_available_skills()[0].name == "url-skill" + + def test_resolve_mixed_url_and_local(self, tmp_path): + """Test resolving a mix of URL and local filesystem sources.""" + from unittest.mock import patch + + _make_skill_dir(tmp_path, "local-skill") + + with patch( + f"{self._SKILL_MODULE}.urllib.request.urlopen", return_value=self._mock_urlopen(self._SAMPLE_CONTENT) + ): + plugin = AgentSkills( + skills=[ + "https://example.com/SKILL.md", + str(tmp_path / "local-skill"), + ] + ) + + assert len(plugin.get_available_skills()) == 2 + names = {s.name for s in plugin.get_available_skills()} + assert names == {"url-skill", "local-skill"} + + def test_resolve_url_failure_skips_gracefully(self, caplog): + """Test that a failed URL fetch is skipped with a warning.""" + import logging + import urllib.error + from unittest.mock import patch + + with ( + patch( + f"{self._SKILL_MODULE}.urllib.request.urlopen", + side_effect=urllib.error.HTTPError( + url="https://example.com", code=404, msg="Not Found", hdrs=None, fp=None + ), + ), + caplog.at_level(logging.WARNING), + ): + plugin = AgentSkills(skills=["https://example.com/broken/SKILL.md"]) + + assert len(plugin.get_available_skills()) == 0 + assert "failed to load skill from URL" in caplog.text + + def test_resolve_duplicate_url_skills_warns(self, caplog): + """Test that duplicate skill names from URLs log a warning.""" + import logging + from unittest.mock import patch + + with ( + patch( + f"{self._SKILL_MODULE}.urllib.request.urlopen", + return_value=self._mock_urlopen(self._SAMPLE_CONTENT), + ), + caplog.at_level(logging.WARNING), + ): + plugin = AgentSkills( + skills=[ + "https://example.com/a/SKILL.md", + "https://example.com/b/SKILL.md", + ] + ) + + assert len(plugin.get_available_skills()) == 1 + assert "duplicate skill name" in caplog.text + + class TestImports: """Tests for module imports.""" diff --git a/tests/strands/vended_plugins/skills/test_skill.py b/tests/strands/vended_plugins/skills/test_skill.py index 53d6f3507..cb67d71a2 100644 --- a/tests/strands/vended_plugins/skills/test_skill.py +++ b/tests/strands/vended_plugins/skills/test_skill.py @@ -551,11 +551,99 @@ def test_strict_mode(self): Skill.from_content(content, strict=True) +class TestSkillFromUrl: + """Tests for Skill.from_url.""" + + _SKILL_MODULE = "strands.vended_plugins.skills.skill" + _SAMPLE_CONTENT = "---\nname: my-skill\ndescription: A remote skill\n---\nRemote instructions.\n" + + def _mock_urlopen(self, content): + """Create a mock urlopen context manager returning the given content.""" + from unittest.mock import MagicMock + + mock_response = MagicMock() + mock_response.read.return_value = content.encode("utf-8") + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + return mock_response + + def test_from_url_returns_skill(self): + """Test loading a skill from a URL returns a single Skill.""" + from unittest.mock import patch + + mock_response = self._mock_urlopen(self._SAMPLE_CONTENT) + with patch(f"{self._SKILL_MODULE}.urllib.request.urlopen", return_value=mock_response): + skill = Skill.from_url("https://raw.githubusercontent.com/org/repo/main/SKILL.md") + + assert isinstance(skill, Skill) + assert skill.name == "my-skill" + assert skill.description == "A remote skill" + assert "Remote instructions." in skill.instructions + assert skill.path is None + + def test_from_url_invalid_url_raises(self): + """Test that a non-HTTPS URL raises ValueError.""" + with pytest.raises(ValueError, match="not a valid HTTPS URL"): + Skill.from_url("./local-path") + + def test_from_url_http_rejected(self): + """Test that http:// URLs are rejected.""" + with pytest.raises(ValueError, match="not a valid HTTPS URL"): + Skill.from_url("http://example.com/SKILL.md") + + def test_from_url_http_error_raises(self): + """Test that HTTP errors propagate as RuntimeError.""" + import urllib.error + from unittest.mock import patch + + with patch( + f"{self._SKILL_MODULE}.urllib.request.urlopen", + side_effect=urllib.error.HTTPError( + url="https://example.com", code=404, msg="Not Found", hdrs=None, fp=None + ), + ): + with pytest.raises(RuntimeError, match="HTTP 404"): + Skill.from_url("https://example.com/SKILL.md") + + def test_from_url_network_error_raises(self): + """Test that network errors propagate as RuntimeError.""" + import urllib.error + from unittest.mock import patch + + with patch( + f"{self._SKILL_MODULE}.urllib.request.urlopen", + side_effect=urllib.error.URLError("Connection refused"), + ): + with pytest.raises(RuntimeError, match="failed to fetch"): + Skill.from_url("https://example.com/SKILL.md") + + def test_from_url_strict_mode(self): + """Test that strict mode is forwarded to from_content.""" + from unittest.mock import patch + + bad_content = "---\nname: BAD_NAME\ndescription: Bad\n---\nBody." + + with patch(f"{self._SKILL_MODULE}.urllib.request.urlopen", return_value=self._mock_urlopen(bad_content)): + with pytest.raises(ValueError): + Skill.from_url("https://example.com/SKILL.md", strict=True) + + def test_from_url_invalid_content_raises(self): + """Test that non-SKILL.md content (e.g. HTML page) raises ValueError.""" + from unittest.mock import patch + + html_content = "Not a SKILL.md" + + with patch(f"{self._SKILL_MODULE}.urllib.request.urlopen", return_value=self._mock_urlopen(html_content)): + with pytest.raises(ValueError, match="frontmatter"): + Skill.from_url("https://example.com/SKILL.md") + + 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.""" + """Test that Skill has from_file, from_content, from_directory, and from_url classmethods.""" assert callable(getattr(Skill, "from_file", None)) assert callable(getattr(Skill, "from_content", None)) assert callable(getattr(Skill, "from_directory", None)) + assert callable(getattr(Skill, "from_url", None))