From 07225e72ef25bf7200239aac6af2692edf3131cb Mon Sep 17 00:00:00 2001 From: Davide Gallitelli Date: Wed, 8 Apr 2026 06:51:28 +0000 Subject: [PATCH 1/2] feat(skills): support loading skills from GitHub/Git URLs Add support for remote Git repository URLs as skill sources in AgentSkills, enabling teams to share and reference skills directly from GitHub without manual cloning. Closes #2090 Co-Authored-By: Claude Opus 4.6 (1M context) --- .../vended_plugins/skills/_url_loader.py | 174 +++++++++++ .../vended_plugins/skills/agent_skills.py | 20 +- src/strands/vended_plugins/skills/skill.py | 49 +++ .../skills/test_agent_skills.py | 106 +++++++ .../vended_plugins/skills/test_skill.py | 139 ++++++++- .../vended_plugins/skills/test_url_loader.py | 287 ++++++++++++++++++ 6 files changed, 773 insertions(+), 2 deletions(-) create mode 100644 src/strands/vended_plugins/skills/_url_loader.py create mode 100644 tests/strands/vended_plugins/skills/test_url_loader.py diff --git a/src/strands/vended_plugins/skills/_url_loader.py b/src/strands/vended_plugins/skills/_url_loader.py new file mode 100644 index 000000000..2f44dfcf1 --- /dev/null +++ b/src/strands/vended_plugins/skills/_url_loader.py @@ -0,0 +1,174 @@ +"""Utilities for loading skills from remote Git repository URLs. + +This module provides functions to detect URL-type skill sources, parse +optional version references, clone repositories with shallow depth, and +manage a local cache of cloned skill repositories. +""" + +from __future__ import annotations + +import hashlib +import logging +import re +import shutil +import subprocess +from pathlib import Path + +logger = logging.getLogger(__name__) + +_DEFAULT_CACHE_DIR = Path.home() / ".cache" / "strands" / "skills" + +# Patterns that indicate a string is a URL rather than a local path +_URL_PREFIXES = ("https://", "http://", "git@", "ssh://") + +# Regex to strip .git suffix from URLs before ref parsing +_GIT_SUFFIX = re.compile(r"\.git$") + + +def is_url(source: str) -> bool: + """Check whether a skill source string looks like a remote URL. + + Args: + source: The skill source string to check. + + Returns: + True if the source appears to be a URL. + """ + return any(source.startswith(prefix) for prefix in _URL_PREFIXES) + + +def parse_url_ref(url: str) -> tuple[str, str | None]: + """Parse a skill URL into a clone URL and an optional Git ref. + + Supports an ``@ref`` suffix for specifying a branch, tag, or commit:: + + https://github.com/org/skill-repo@v1.0.0 -> (https://github.com/org/skill-repo, v1.0.0) + https://github.com/org/skill-repo -> (https://github.com/org/skill-repo, None) + https://github.com/org/skill-repo.git@main -> (https://github.com/org/skill-repo.git, main) + git@github.com:org/skill-repo.git@v2 -> (git@github.com:org/skill-repo.git, v2) + + Args: + url: The skill URL, optionally with an ``@ref`` suffix. + + Returns: + Tuple of (clone_url, ref_or_none). + """ + if url.startswith(("https://", "http://", "ssh://")): + # Find the path portion after the host + scheme_end = url.index("//") + 2 + host_end = url.find("/", scheme_end) + if host_end == -1: + return url, None + + path_part = url[host_end:] + + # Strip .git suffix before looking for @ref so that + # "repo.git@v1" is handled correctly + clean_path = _GIT_SUFFIX.sub("", path_part) + had_git_suffix = clean_path != path_part + + if "@" in clean_path: + at_idx = clean_path.rfind("@") + ref = clean_path[at_idx + 1 :] + base_path = clean_path[:at_idx] + if had_git_suffix: + base_path += ".git" + return url[:host_end] + base_path, ref + + return url, None + + if url.startswith("git@"): + # SSH format: git@host:owner/repo.git@ref + # The first @ is part of the SSH URL format. + first_at = url.index("@") + rest = url[first_at + 1 :] + + clean_rest = _GIT_SUFFIX.sub("", rest) + had_git_suffix = clean_rest != rest + + if "@" in clean_rest: + at_idx = clean_rest.rfind("@") + ref = clean_rest[at_idx + 1 :] + base_rest = clean_rest[:at_idx] + if had_git_suffix: + base_rest += ".git" + return url[: first_at + 1] + base_rest, ref + + return url, None + + return url, None + + +def cache_key(url: str, ref: str | None) -> str: + """Generate a deterministic cache directory name from a URL and ref. + + Args: + url: The clone URL. + ref: The optional Git ref. + + Returns: + A short hex digest suitable for use as a directory name. + """ + key_input = f"{url}@{ref}" if ref else url + return hashlib.sha256(key_input.encode()).hexdigest()[:16] + + +def clone_skill_repo( + url: str, + *, + ref: str | None = None, + cache_dir: Path | None = None, +) -> Path: + """Clone a skill repository to a local cache directory. + + Uses ``git clone --depth 1`` for efficiency. If a ``ref`` is provided it + is passed as ``--branch`` (works for branches and tags). Repositories are + cached by a hash of (url, ref) so repeated loads are instant. + + Args: + url: The Git clone URL. + ref: Optional branch or tag to check out. + cache_dir: Override the default cache directory + (``~/.cache/strands/skills/``). + + Returns: + Path to the cloned repository root. + + Raises: + RuntimeError: If the clone fails or ``git`` is not installed. + """ + cache_dir = cache_dir or _DEFAULT_CACHE_DIR + cache_dir.mkdir(parents=True, exist_ok=True) + + key = cache_key(url, ref) + target = cache_dir / key + + if target.exists(): + logger.debug("url=<%s>, ref=<%s> | using cached skill at %s", url, ref, target) + return target + + logger.info("url=<%s>, ref=<%s> | cloning skill repository", url, ref) + + cmd: list[str] = ["git", "clone", "--depth", "1"] + if ref: + cmd.extend(["--branch", ref]) + cmd.extend([url, str(target)]) + + try: + subprocess.run( # noqa: S603 + cmd, + check=True, + capture_output=True, + text=True, + timeout=120, + ) + except subprocess.CalledProcessError as e: + # Clean up any partial clone + if target.exists(): + shutil.rmtree(target) + raise RuntimeError(f"url=<{url}>, ref=<{ref}> | failed to clone skill repository: {e.stderr.strip()}") from e + except FileNotFoundError as e: + raise RuntimeError("git is required to load skills from URLs but was not found on PATH") from e + + logger.debug("url=<%s>, ref=<%s> | cloned to %s", url, ref, target) + return target diff --git a/src/strands/vended_plugins/skills/agent_skills.py b/src/strands/vended_plugins/skills/agent_skills.py index 5e42b9230..2459be120 100644 --- a/src/strands/vended_plugins/skills/agent_skills.py +++ b/src/strands/vended_plugins/skills/agent_skills.py @@ -77,6 +77,7 @@ def __init__( state_key: str = _DEFAULT_STATE_KEY, max_resource_files: int = _DEFAULT_MAX_RESOURCE_FILES, strict: bool = False, + cache_dir: Path | None = None, ) -> None: """Initialize the AgentSkills plugin. @@ -86,11 +87,16 @@ 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 + - A remote Git URL (``https://``, ``git@``, or ``ssh://``) + with optional ``@ref`` suffix for branch/tag pinning 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. + cache_dir: Directory for caching cloned skill repositories. + Defaults to ``~/.cache/strands/skills/``. """ self._strict = strict + self._cache_dir = cache_dir self._skills: dict[str, Skill] = self._resolve_skills(_normalize_sources(skills)) self._state_key = state_key self._max_resource_files = max_resource_files @@ -284,7 +290,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 a remote + Git URL. Args: sources: List of skill sources to resolve. @@ -292,6 +299,8 @@ def _resolve_skills(self, sources: list[SkillSource]) -> dict[str, Skill]: Returns: Dict mapping skill names to Skill instances. """ + from ._url_loader import is_url + resolved: dict[str, Skill] = {} for source in sources: @@ -299,6 +308,15 @@ 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 is_url(source): + try: + url_skills = Skill.from_url(source, cache_dir=self._cache_dir, strict=self._strict) + for skill in url_skills: + 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..40724270f 100644 --- a/src/strands/vended_plugins/skills/skill.py +++ b/src/strands/vended_plugins/skills/skill.py @@ -333,6 +333,55 @@ def from_content(cls, content: str, *, strict: bool = False) -> Skill: return _build_skill_from_frontmatter(frontmatter, body) + @classmethod + def from_url( + cls, + url: str, + *, + cache_dir: Path | None = None, + strict: bool = False, + ) -> list[Skill]: + """Load skill(s) from a remote Git repository URL. + + Clones the repository (or uses a cached copy) and then loads skills + using the standard filesystem methods. If the repository root contains + a ``SKILL.md`` file it is treated as a single skill; otherwise it is + scanned for skill subdirectories. + + Supports an optional ``@ref`` suffix for branch or tag pinning:: + + skills = Skill.from_url("https://github.com/org/my-skill@v1.0.0") + + Args: + url: A Git-cloneable URL, optionally with an ``@ref`` suffix. + cache_dir: Override the default cache directory + (``~/.cache/strands/skills/``). + strict: If True, raise on any validation issue. If False (default), + warn and load anyway. + + Returns: + List of Skill instances loaded from the repository. + + Raises: + RuntimeError: If the repository cannot be cloned or ``git`` is not + available. + """ + from ._url_loader import clone_skill_repo, is_url, parse_url_ref + + if not is_url(url): + raise ValueError(f"url=<{url}> | not a valid remote URL") + + clean_url, ref = parse_url_ref(url) + repo_path = clone_skill_repo(clean_url, ref=ref, cache_dir=cache_dir) + + # If the repo root is itself a skill, load it directly + has_skill_md = (repo_path / "SKILL.md").is_file() or (repo_path / "skill.md").is_file() + + if has_skill_md: + return [cls.from_file(repo_path, strict=strict)] + else: + return cls.from_directory(repo_path, 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..bbe62cdea 100644 --- a/tests/strands/vended_plugins/skills/test_agent_skills.py +++ b/tests/strands/vended_plugins/skills/test_agent_skills.py @@ -661,6 +661,112 @@ def test_resolve_nonexistent_path(self, tmp_path): assert len(plugin._skills) == 0 +class TestResolveUrlSkills: + """Tests for _resolve_skills with URL sources.""" + + _URL_LOADER = "strands.vended_plugins.skills._url_loader" + + def _mock_clone(self, tmp_path, skill_name="url-skill", description="A URL skill"): + """Create a mock clone function that creates a skill directory.""" + skill_dir = tmp_path / "cloned" + + def fake_clone(url, *, ref=None, cache_dir=None): + skill_dir.mkdir(parents=True, exist_ok=True) + content = f"---\nname: {skill_name}\ndescription: {description}\n---\n# Instructions\n" + (skill_dir / "SKILL.md").write_text(content) + return skill_dir + + return fake_clone + + def test_resolve_url_source(self, tmp_path): + """Test resolving a URL string as a skill source.""" + from unittest.mock import patch + + fake_clone = self._mock_clone(tmp_path) + + with ( + patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=fake_clone), + patch( + f"{self._URL_LOADER}.parse_url_ref", + return_value=("https://github.com/org/url-skill", None), + ), + ): + plugin = AgentSkills(skills=["https://github.com/org/url-skill"]) + + 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") + fake_clone = self._mock_clone(tmp_path) + + with ( + patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=fake_clone), + patch( + f"{self._URL_LOADER}.parse_url_ref", + return_value=("https://github.com/org/url-skill", None), + ), + ): + plugin = AgentSkills( + skills=[ + "https://github.com/org/url-skill", + 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 clone is skipped with a warning.""" + import logging + from unittest.mock import patch + + with ( + patch( + f"{self._URL_LOADER}.parse_url_ref", + return_value=("https://github.com/org/broken", None), + ), + patch( + f"{self._URL_LOADER}.clone_skill_repo", + side_effect=RuntimeError("clone failed"), + ), + caplog.at_level(logging.WARNING), + ): + plugin = AgentSkills(skills=["https://github.com/org/broken"]) + + assert len(plugin.get_available_skills()) == 0 + assert "failed to load skill from URL" in caplog.text + + def test_cache_dir_forwarded(self, tmp_path): + """Test that custom cache_dir is forwarded through to clone.""" + from unittest.mock import patch + + fake_clone = self._mock_clone(tmp_path) + captured_cache = [] + + def tracking_clone(url, *, ref=None, cache_dir=None): + captured_cache.append(cache_dir) + return fake_clone(url, ref=ref, cache_dir=cache_dir) + + custom_cache = tmp_path / "my-cache" + + with ( + patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=tracking_clone), + patch( + f"{self._URL_LOADER}.parse_url_ref", + return_value=("https://github.com/org/url-skill", None), + ), + ): + AgentSkills(skills=["https://github.com/org/url-skill"], cache_dir=custom_cache) + + assert captured_cache == [custom_cache] + + 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..8857382fb 100644 --- a/tests/strands/vended_plugins/skills/test_skill.py +++ b/tests/strands/vended_plugins/skills/test_skill.py @@ -551,11 +551,148 @@ def test_strict_mode(self): Skill.from_content(content, strict=True) +class TestSkillFromUrl: + """Tests for Skill.from_url.""" + + _URL_LOADER = "strands.vended_plugins.skills._url_loader" + + def _mock_clone(self, tmp_path, skill_name="my-skill", description="A remote skill", body="Remote instructions."): + """Create a mock clone function that creates a skill directory.""" + skill_dir = tmp_path / "cloned" + + def fake_clone(url, *, ref=None, cache_dir=None): + skill_dir.mkdir(parents=True, exist_ok=True) + content = f"---\nname: {skill_name}\ndescription: {description}\n---\n{body}\n" + (skill_dir / "SKILL.md").write_text(content) + return skill_dir + + return fake_clone + + def _mock_clone_multi(self, tmp_path): + """Create a mock clone function that creates a parent dir with multiple skills.""" + parent_dir = tmp_path / "cloned" + + def fake_clone(url, *, ref=None, cache_dir=None): + parent_dir.mkdir(parents=True, exist_ok=True) + for name in ("skill-a", "skill-b"): + child = parent_dir / name + child.mkdir(exist_ok=True) + (child / "SKILL.md").write_text(f"---\nname: {name}\ndescription: Skill {name}\n---\nBody.\n") + return parent_dir + + return fake_clone + + def test_from_url_single_skill(self, tmp_path): + """Test loading a single skill from a URL.""" + from unittest.mock import patch + + fake_clone = self._mock_clone(tmp_path) + + with ( + patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=fake_clone), + patch(f"{self._URL_LOADER}.parse_url_ref", return_value=("https://github.com/org/my-skill", None)), + ): + skills = Skill.from_url("https://github.com/org/my-skill") + + assert len(skills) == 1 + assert skills[0].name == "my-skill" + assert skills[0].description == "A remote skill" + assert "Remote instructions." in skills[0].instructions + + def test_from_url_multiple_skills(self, tmp_path): + """Test loading multiple skills from a URL pointing to a parent directory.""" + from unittest.mock import patch + + fake_clone = self._mock_clone_multi(tmp_path) + + with ( + patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=fake_clone), + patch( + f"{self._URL_LOADER}.parse_url_ref", + return_value=("https://github.com/org/skills-collection", None), + ), + ): + skills = Skill.from_url("https://github.com/org/skills-collection") + + assert len(skills) == 2 + names = {s.name for s in skills} + assert names == {"skill-a", "skill-b"} + + def test_from_url_with_ref(self, tmp_path): + """Test that @ref is correctly parsed and forwarded.""" + from unittest.mock import patch + + fake_clone = self._mock_clone(tmp_path) + captured_ref = [] + + def tracking_clone(url, *, ref=None, cache_dir=None): + captured_ref.append(ref) + return fake_clone(url, ref=ref, cache_dir=cache_dir) + + with ( + patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=tracking_clone), + patch( + f"{self._URL_LOADER}.parse_url_ref", + return_value=("https://github.com/org/my-skill", "v1.0.0"), + ), + ): + Skill.from_url("https://github.com/org/my-skill@v1.0.0") + + assert captured_ref == ["v1.0.0"] + + def test_from_url_invalid_url_raises(self): + """Test that a non-URL string raises ValueError.""" + with pytest.raises(ValueError, match="not a valid remote URL"): + Skill.from_url("./local-path") + + def test_from_url_clone_failure_raises(self): + """Test that a clone failure propagates as RuntimeError.""" + from unittest.mock import patch + + with ( + patch( + f"{self._URL_LOADER}.parse_url_ref", + return_value=("https://github.com/org/broken", None), + ), + patch( + f"{self._URL_LOADER}.clone_skill_repo", + side_effect=RuntimeError("clone failed"), + ), + ): + with pytest.raises(RuntimeError, match="clone failed"): + Skill.from_url("https://github.com/org/broken") + + def test_from_url_passes_cache_dir(self, tmp_path): + """Test that cache_dir is forwarded to clone_skill_repo.""" + from unittest.mock import patch + + fake_clone = self._mock_clone(tmp_path) + captured_cache = [] + + def tracking_clone(url, *, ref=None, cache_dir=None): + captured_cache.append(cache_dir) + return fake_clone(url, ref=ref, cache_dir=cache_dir) + + custom_cache = tmp_path / "custom-cache" + + with ( + patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=tracking_clone), + patch( + f"{self._URL_LOADER}.parse_url_ref", + return_value=("https://github.com/org/my-skill", None), + ), + ): + Skill.from_url("https://github.com/org/my-skill", cache_dir=custom_cache) + + assert captured_cache == [custom_cache] + + 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)) diff --git a/tests/strands/vended_plugins/skills/test_url_loader.py b/tests/strands/vended_plugins/skills/test_url_loader.py new file mode 100644 index 000000000..994672c32 --- /dev/null +++ b/tests/strands/vended_plugins/skills/test_url_loader.py @@ -0,0 +1,287 @@ +"""Tests for the _url_loader module.""" + +from __future__ import annotations + +import subprocess +from pathlib import Path +from unittest.mock import patch + +import pytest + +from strands.vended_plugins.skills._url_loader import ( + cache_key, + clone_skill_repo, + is_url, + parse_url_ref, +) + + +class TestIsUrl: + """Tests for is_url.""" + + def test_https_github_url(self): + assert is_url("https://github.com/org/skill-repo") is True + + def test_http_url(self): + assert is_url("http://github.com/org/skill-repo") is True + + def test_ssh_url(self): + assert is_url("ssh://git@github.com/org/skill-repo") is True + + def test_git_at_url(self): + assert is_url("git@github.com:org/skill-repo.git") is True + + def test_local_relative_path(self): + assert is_url("./skills/my-skill") is False + + def test_local_absolute_path(self): + assert is_url("/home/user/skills/my-skill") is False + + def test_plain_directory_name(self): + assert is_url("my-skill") is False + + def test_empty_string(self): + assert is_url("") is False + + def test_https_with_ref(self): + assert is_url("https://github.com/org/skill@v1.0.0") is True + + +class TestParseUrlRef: + """Tests for parse_url_ref.""" + + def test_https_no_ref(self): + url, ref = parse_url_ref("https://github.com/org/skill-repo") + assert url == "https://github.com/org/skill-repo" + assert ref is None + + def test_https_with_tag_ref(self): + url, ref = parse_url_ref("https://github.com/org/skill-repo@v1.0.0") + assert url == "https://github.com/org/skill-repo" + assert ref == "v1.0.0" + + def test_https_with_branch_ref(self): + url, ref = parse_url_ref("https://github.com/org/skill-repo@main") + assert url == "https://github.com/org/skill-repo" + assert ref == "main" + + def test_https_git_suffix_no_ref(self): + url, ref = parse_url_ref("https://github.com/org/skill-repo.git") + assert url == "https://github.com/org/skill-repo.git" + assert ref is None + + def test_https_git_suffix_with_ref(self): + url, ref = parse_url_ref("https://github.com/org/skill-repo.git@v2.0") + assert url == "https://github.com/org/skill-repo.git" + assert ref == "v2.0" + + def test_ssh_no_ref(self): + url, ref = parse_url_ref("git@github.com:org/skill-repo.git") + assert url == "git@github.com:org/skill-repo.git" + assert ref is None + + def test_ssh_with_ref(self): + url, ref = parse_url_ref("git@github.com:org/skill-repo.git@v1.0") + assert url == "git@github.com:org/skill-repo.git" + assert ref == "v1.0" + + def test_ssh_no_git_suffix_with_ref(self): + url, ref = parse_url_ref("git@github.com:org/skill-repo@develop") + assert url == "git@github.com:org/skill-repo" + assert ref == "develop" + + def test_ssh_protocol_no_ref(self): + url, ref = parse_url_ref("ssh://git@github.com/org/skill-repo") + assert url == "ssh://git@github.com/org/skill-repo" + assert ref is None + + def test_ssh_protocol_with_ref(self): + url, ref = parse_url_ref("ssh://git@github.com/org/skill-repo@v3") + assert url == "ssh://git@github.com/org/skill-repo" + assert ref == "v3" + + def test_http_with_ref(self): + url, ref = parse_url_ref("http://gitlab.com/org/skill@feature-branch") + assert url == "http://gitlab.com/org/skill" + assert ref == "feature-branch" + + def test_url_host_only(self): + url, ref = parse_url_ref("https://example.com") + assert url == "https://example.com" + assert ref is None + + def test_non_url_passthrough(self): + url, ref = parse_url_ref("/local/path") + assert url == "/local/path" + assert ref is None + + +class TestCacheKey: + """Tests for cache_key.""" + + def test_deterministic(self): + key1 = cache_key("https://github.com/org/repo", None) + key2 = cache_key("https://github.com/org/repo", None) + assert key1 == key2 + + def test_different_url_different_key(self): + key1 = cache_key("https://github.com/org/repo-a", None) + key2 = cache_key("https://github.com/org/repo-b", None) + assert key1 != key2 + + def test_different_ref_different_key(self): + key1 = cache_key("https://github.com/org/repo", None) + key2 = cache_key("https://github.com/org/repo", "v1.0") + assert key1 != key2 + + def test_length(self): + key = cache_key("https://github.com/org/repo", "main") + assert len(key) == 16 + + def test_hex_characters_only(self): + key = cache_key("https://github.com/org/repo", "main") + assert all(c in "0123456789abcdef" for c in key) + + +class TestCloneSkillRepo: + """Tests for clone_skill_repo.""" + + def test_clone_success(self, tmp_path): + """Test successful clone by mocking subprocess.run.""" + cache = tmp_path / "cache" + + def fake_clone(cmd, **kwargs): + # Simulate a successful clone by creating the target directory + target_dir = Path(cmd[-1]) + target_dir.mkdir(parents=True, exist_ok=True) + (target_dir / "SKILL.md").write_text("---\nname: test\ndescription: test\n---\n") + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + with patch("strands.vended_plugins.skills._url_loader.subprocess.run", side_effect=fake_clone): + result = clone_skill_repo("https://github.com/org/skill", cache_dir=cache) + + assert result.exists() + assert (result / "SKILL.md").exists() + + def test_clone_with_ref(self, tmp_path): + """Test that ref is passed as --branch to git clone.""" + cache = tmp_path / "cache" + captured_cmd = [] + + def fake_clone(cmd, **kwargs): + captured_cmd.extend(cmd) + target_dir = Path(cmd[-1]) + target_dir.mkdir(parents=True, exist_ok=True) + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + with patch("strands.vended_plugins.skills._url_loader.subprocess.run", side_effect=fake_clone): + clone_skill_repo("https://github.com/org/skill", ref="v1.0", cache_dir=cache) + + assert "--branch" in captured_cmd + assert "v1.0" in captured_cmd + + def test_clone_without_ref(self, tmp_path): + """Test that --branch is not passed when ref is None.""" + cache = tmp_path / "cache" + captured_cmd = [] + + def fake_clone(cmd, **kwargs): + captured_cmd.extend(cmd) + target_dir = Path(cmd[-1]) + target_dir.mkdir(parents=True, exist_ok=True) + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + with patch("strands.vended_plugins.skills._url_loader.subprocess.run", side_effect=fake_clone): + clone_skill_repo("https://github.com/org/skill", cache_dir=cache) + + assert "--branch" not in captured_cmd + + def test_uses_cache_on_second_call(self, tmp_path): + """Test that the cache is used on the second call.""" + cache = tmp_path / "cache" + call_count = 0 + + def fake_clone(cmd, **kwargs): + nonlocal call_count + call_count += 1 + target_dir = Path(cmd[-1]) + target_dir.mkdir(parents=True, exist_ok=True) + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + with patch("strands.vended_plugins.skills._url_loader.subprocess.run", side_effect=fake_clone): + result1 = clone_skill_repo("https://github.com/org/skill", cache_dir=cache) + result2 = clone_skill_repo("https://github.com/org/skill", cache_dir=cache) + + assert call_count == 1 + assert result1 == result2 + + def test_clone_failure_raises_runtime_error(self, tmp_path): + """Test that a failed clone raises RuntimeError.""" + cache = tmp_path / "cache" + + with patch( + "strands.vended_plugins.skills._url_loader.subprocess.run", + side_effect=subprocess.CalledProcessError(128, "git", stderr="fatal: repo not found"), + ): + with pytest.raises(RuntimeError, match="failed to clone"): + clone_skill_repo("https://github.com/org/nonexistent", cache_dir=cache) + + def test_clone_failure_cleans_up_partial(self, tmp_path): + """Test that a failed clone removes any partial directory.""" + cache = tmp_path / "cache" + + def failing_clone(cmd, **kwargs): + # Create partial clone then fail + target_dir = Path(cmd[-1]) + target_dir.mkdir(parents=True, exist_ok=True) + raise subprocess.CalledProcessError(128, "git", stderr="fatal: error") + + with patch("strands.vended_plugins.skills._url_loader.subprocess.run", side_effect=failing_clone): + with pytest.raises(RuntimeError): + clone_skill_repo("https://github.com/org/broken", cache_dir=cache) + + # Verify no leftover directory + assert len(list(cache.iterdir())) == 0 + + def test_git_not_found_raises_runtime_error(self, tmp_path): + """Test that missing git binary raises RuntimeError.""" + cache = tmp_path / "cache" + + with patch( + "strands.vended_plugins.skills._url_loader.subprocess.run", + side_effect=FileNotFoundError(), + ): + with pytest.raises(RuntimeError, match="git is required"): + clone_skill_repo("https://github.com/org/skill", cache_dir=cache) + + def test_creates_cache_dir_if_missing(self, tmp_path): + """Test that the cache directory is created if it doesn't exist.""" + cache = tmp_path / "deep" / "nested" / "cache" + + def fake_clone(cmd, **kwargs): + target_dir = Path(cmd[-1]) + target_dir.mkdir(parents=True, exist_ok=True) + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + with patch("strands.vended_plugins.skills._url_loader.subprocess.run", side_effect=fake_clone): + clone_skill_repo("https://github.com/org/skill", cache_dir=cache) + + assert cache.exists() + + def test_shallow_clone_depth_one(self, tmp_path): + """Test that --depth 1 is always passed.""" + cache = tmp_path / "cache" + captured_cmd = [] + + def fake_clone(cmd, **kwargs): + captured_cmd.extend(cmd) + target_dir = Path(cmd[-1]) + target_dir.mkdir(parents=True, exist_ok=True) + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + with patch("strands.vended_plugins.skills._url_loader.subprocess.run", side_effect=fake_clone): + clone_skill_repo("https://github.com/org/skill", cache_dir=cache) + + assert "--depth" in captured_cmd + depth_idx = captured_cmd.index("--depth") + assert captured_cmd[depth_idx + 1] == "1" From 01a4fc34a11602e92280d77c19b74cfad0d63f0f Mon Sep 17 00:00:00 2001 From: Davide Gallitelli Date: Wed, 8 Apr 2026 08:00:19 +0000 Subject: [PATCH 2/2] feat(skills): support GitHub /tree/ URLs for nested skills Parse GitHub web URLs like /tree//path to extract the clone URL, branch, and subdirectory path. This enables loading skills from subdirectories within mono-repos. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../vended_plugins/skills/_url_loader.py | 114 +++++++++++------- src/strands/vended_plugins/skills/skill.py | 11 +- .../skills/test_agent_skills.py | 12 +- .../vended_plugins/skills/test_skill.py | 18 +-- .../vended_plugins/skills/test_url_loader.py | 112 +++++++++++++++-- 5 files changed, 193 insertions(+), 74 deletions(-) diff --git a/src/strands/vended_plugins/skills/_url_loader.py b/src/strands/vended_plugins/skills/_url_loader.py index 2f44dfcf1..9e4531917 100644 --- a/src/strands/vended_plugins/skills/_url_loader.py +++ b/src/strands/vended_plugins/skills/_url_loader.py @@ -24,6 +24,10 @@ # Regex to strip .git suffix from URLs before ref parsing _GIT_SUFFIX = re.compile(r"\.git$") +# Matches GitHub /tree/ or /tree// (also /blob/) +# e.g. /owner/repo/tree/main/skills/my-skill -> groups: (/owner/repo, main, skills/my-skill) +_GITHUB_TREE_PATTERN = re.compile(r"^(/[^/]+/[^/]+)/(?:tree|blob)/([^/]+)(?:/(.+?))?/?$") + def is_url(source: str) -> bool: """Check whether a skill source string looks like a remote URL. @@ -37,31 +41,43 @@ def is_url(source: str) -> bool: return any(source.startswith(prefix) for prefix in _URL_PREFIXES) -def parse_url_ref(url: str) -> tuple[str, str | None]: - """Parse a skill URL into a clone URL and an optional Git ref. +def parse_url_ref(url: str) -> tuple[str, str | None, str | None]: + """Parse a skill URL into a clone URL, optional Git ref, and optional subpath. Supports an ``@ref`` suffix for specifying a branch, tag, or commit:: - https://github.com/org/skill-repo@v1.0.0 -> (https://github.com/org/skill-repo, v1.0.0) - https://github.com/org/skill-repo -> (https://github.com/org/skill-repo, None) - https://github.com/org/skill-repo.git@main -> (https://github.com/org/skill-repo.git, main) - git@github.com:org/skill-repo.git@v2 -> (git@github.com:org/skill-repo.git, v2) + https://github.com/org/skill-repo@v1.0.0 -> (https://github.com/org/skill-repo, v1.0.0, None) + https://github.com/org/skill-repo -> (https://github.com/org/skill-repo, None, None) + + Also supports GitHub web URLs with ``/tree//path`` :: + + https://github.com/org/repo/tree/main/skills/my-skill + -> (https://github.com/org/repo, main, skills/my-skill) Args: - url: The skill URL, optionally with an ``@ref`` suffix. + url: The skill URL, optionally with an ``@ref`` suffix or ``/tree/`` path. Returns: - Tuple of (clone_url, ref_or_none). + Tuple of (clone_url, ref_or_none, subpath_or_none). """ if url.startswith(("https://", "http://", "ssh://")): # Find the path portion after the host scheme_end = url.index("//") + 2 host_end = url.find("/", scheme_end) if host_end == -1: - return url, None + return url, None, None path_part = url[host_end:] + # Handle GitHub /tree//path and /blob//path URLs + tree_match = _GITHUB_TREE_PATTERN.match(path_part) + if tree_match: + owner_repo = tree_match.group(1) + ref = tree_match.group(2) + subpath = tree_match.group(3) or None + clone_url = url[:host_end] + owner_repo + return clone_url, ref, subpath + # Strip .git suffix before looking for @ref so that # "repo.git@v1" is handled correctly clean_path = _GIT_SUFFIX.sub("", path_part) @@ -73,9 +89,9 @@ def parse_url_ref(url: str) -> tuple[str, str | None]: base_path = clean_path[:at_idx] if had_git_suffix: base_path += ".git" - return url[:host_end] + base_path, ref + return url[:host_end] + base_path, ref, None - return url, None + return url, None, None if url.startswith("git@"): # SSH format: git@host:owner/repo.git@ref @@ -92,11 +108,11 @@ def parse_url_ref(url: str) -> tuple[str, str | None]: base_rest = clean_rest[:at_idx] if had_git_suffix: base_rest += ".git" - return url[: first_at + 1] + base_rest, ref + return url[: first_at + 1] + base_rest, ref, None - return url, None + return url, None, None - return url, None + return url, None, None def cache_key(url: str, ref: str | None) -> str: @@ -117,6 +133,7 @@ def clone_skill_repo( url: str, *, ref: str | None = None, + subpath: str | None = None, cache_dir: Path | None = None, ) -> Path: """Clone a skill repository to a local cache directory. @@ -125,14 +142,19 @@ def clone_skill_repo( is passed as ``--branch`` (works for branches and tags). Repositories are cached by a hash of (url, ref) so repeated loads are instant. + If ``subpath`` is provided, the returned path points to that subdirectory + within the cloned repository (useful for mono-repos containing skills in + nested directories). + Args: url: The Git clone URL. ref: Optional branch or tag to check out. + subpath: Optional path within the repo to return (e.g. ``skills/my-skill``). cache_dir: Override the default cache directory (``~/.cache/strands/skills/``). Returns: - Path to the cloned repository root. + Path to the cloned repository root, or to ``subpath`` within it. Raises: RuntimeError: If the clone fails or ``git`` is not installed. @@ -143,32 +165,38 @@ def clone_skill_repo( key = cache_key(url, ref) target = cache_dir / key - if target.exists(): + if not target.exists(): + logger.info("url=<%s>, ref=<%s> | cloning skill repository", url, ref) + + cmd: list[str] = ["git", "clone", "--depth", "1"] + if ref: + cmd.extend(["--branch", ref]) + cmd.extend([url, str(target)]) + + try: + subprocess.run( # noqa: S603 + cmd, + check=True, + capture_output=True, + text=True, + timeout=120, + ) + except subprocess.CalledProcessError as e: + # Clean up any partial clone + if target.exists(): + shutil.rmtree(target) + raise RuntimeError( + f"url=<{url}>, ref=<{ref}> | failed to clone skill repository: {e.stderr.strip()}" + ) from e + except FileNotFoundError as e: + raise RuntimeError("git is required to load skills from URLs but was not found on PATH") from e + else: logger.debug("url=<%s>, ref=<%s> | using cached skill at %s", url, ref, target) - return target - - logger.info("url=<%s>, ref=<%s> | cloning skill repository", url, ref) - - cmd: list[str] = ["git", "clone", "--depth", "1"] - if ref: - cmd.extend(["--branch", ref]) - cmd.extend([url, str(target)]) - - try: - subprocess.run( # noqa: S603 - cmd, - check=True, - capture_output=True, - text=True, - timeout=120, - ) - except subprocess.CalledProcessError as e: - # Clean up any partial clone - if target.exists(): - shutil.rmtree(target) - raise RuntimeError(f"url=<{url}>, ref=<{ref}> | failed to clone skill repository: {e.stderr.strip()}") from e - except FileNotFoundError as e: - raise RuntimeError("git is required to load skills from URLs but was not found on PATH") from e - - logger.debug("url=<%s>, ref=<%s> | cloned to %s", url, ref, target) - return target + + result = target / subpath if subpath else target + + if subpath and not result.is_dir(): + raise RuntimeError(f"url=<{url}>, subpath=<{subpath}> | subdirectory does not exist in cloned repository") + + logger.debug("url=<%s>, ref=<%s> | resolved to %s", url, ref, result) + return result diff --git a/src/strands/vended_plugins/skills/skill.py b/src/strands/vended_plugins/skills/skill.py index 40724270f..8ce9194a9 100644 --- a/src/strands/vended_plugins/skills/skill.py +++ b/src/strands/vended_plugins/skills/skill.py @@ -352,8 +352,13 @@ def from_url( skills = Skill.from_url("https://github.com/org/my-skill@v1.0.0") + Also supports GitHub web URLs pointing to subdirectories:: + + skills = Skill.from_url("https://github.com/org/repo/tree/main/skills/my-skill") + Args: - url: A Git-cloneable URL, optionally with an ``@ref`` suffix. + url: A Git-cloneable URL, optionally with an ``@ref`` suffix or + a GitHub ``/tree//path`` URL. cache_dir: Override the default cache directory (``~/.cache/strands/skills/``). strict: If True, raise on any validation issue. If False (default), @@ -371,8 +376,8 @@ def from_url( if not is_url(url): raise ValueError(f"url=<{url}> | not a valid remote URL") - clean_url, ref = parse_url_ref(url) - repo_path = clone_skill_repo(clean_url, ref=ref, cache_dir=cache_dir) + clean_url, ref, subpath = parse_url_ref(url) + repo_path = clone_skill_repo(clean_url, ref=ref, subpath=subpath, cache_dir=cache_dir) # If the repo root is itself a skill, load it directly has_skill_md = (repo_path / "SKILL.md").is_file() or (repo_path / "skill.md").is_file() diff --git a/tests/strands/vended_plugins/skills/test_agent_skills.py b/tests/strands/vended_plugins/skills/test_agent_skills.py index bbe62cdea..6f150a792 100644 --- a/tests/strands/vended_plugins/skills/test_agent_skills.py +++ b/tests/strands/vended_plugins/skills/test_agent_skills.py @@ -670,7 +670,7 @@ def _mock_clone(self, tmp_path, skill_name="url-skill", description="A URL skill """Create a mock clone function that creates a skill directory.""" skill_dir = tmp_path / "cloned" - def fake_clone(url, *, ref=None, cache_dir=None): + def fake_clone(url, *, ref=None, subpath=None, cache_dir=None): skill_dir.mkdir(parents=True, exist_ok=True) content = f"---\nname: {skill_name}\ndescription: {description}\n---\n# Instructions\n" (skill_dir / "SKILL.md").write_text(content) @@ -688,7 +688,7 @@ def test_resolve_url_source(self, tmp_path): patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=fake_clone), patch( f"{self._URL_LOADER}.parse_url_ref", - return_value=("https://github.com/org/url-skill", None), + return_value=("https://github.com/org/url-skill", None, None), ), ): plugin = AgentSkills(skills=["https://github.com/org/url-skill"]) @@ -707,7 +707,7 @@ def test_resolve_mixed_url_and_local(self, tmp_path): patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=fake_clone), patch( f"{self._URL_LOADER}.parse_url_ref", - return_value=("https://github.com/org/url-skill", None), + return_value=("https://github.com/org/url-skill", None, None), ), ): plugin = AgentSkills( @@ -729,7 +729,7 @@ def test_resolve_url_failure_skips_gracefully(self, caplog): with ( patch( f"{self._URL_LOADER}.parse_url_ref", - return_value=("https://github.com/org/broken", None), + return_value=("https://github.com/org/broken", None, None), ), patch( f"{self._URL_LOADER}.clone_skill_repo", @@ -749,7 +749,7 @@ def test_cache_dir_forwarded(self, tmp_path): fake_clone = self._mock_clone(tmp_path) captured_cache = [] - def tracking_clone(url, *, ref=None, cache_dir=None): + def tracking_clone(url, *, ref=None, subpath=None, cache_dir=None): captured_cache.append(cache_dir) return fake_clone(url, ref=ref, cache_dir=cache_dir) @@ -759,7 +759,7 @@ def tracking_clone(url, *, ref=None, cache_dir=None): patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=tracking_clone), patch( f"{self._URL_LOADER}.parse_url_ref", - return_value=("https://github.com/org/url-skill", None), + return_value=("https://github.com/org/url-skill", None, None), ), ): AgentSkills(skills=["https://github.com/org/url-skill"], cache_dir=custom_cache) diff --git a/tests/strands/vended_plugins/skills/test_skill.py b/tests/strands/vended_plugins/skills/test_skill.py index 8857382fb..5b6753bd2 100644 --- a/tests/strands/vended_plugins/skills/test_skill.py +++ b/tests/strands/vended_plugins/skills/test_skill.py @@ -560,7 +560,7 @@ def _mock_clone(self, tmp_path, skill_name="my-skill", description="A remote ski """Create a mock clone function that creates a skill directory.""" skill_dir = tmp_path / "cloned" - def fake_clone(url, *, ref=None, cache_dir=None): + def fake_clone(url, *, ref=None, subpath=None, cache_dir=None): skill_dir.mkdir(parents=True, exist_ok=True) content = f"---\nname: {skill_name}\ndescription: {description}\n---\n{body}\n" (skill_dir / "SKILL.md").write_text(content) @@ -572,7 +572,7 @@ def _mock_clone_multi(self, tmp_path): """Create a mock clone function that creates a parent dir with multiple skills.""" parent_dir = tmp_path / "cloned" - def fake_clone(url, *, ref=None, cache_dir=None): + def fake_clone(url, *, ref=None, subpath=None, cache_dir=None): parent_dir.mkdir(parents=True, exist_ok=True) for name in ("skill-a", "skill-b"): child = parent_dir / name @@ -590,7 +590,7 @@ def test_from_url_single_skill(self, tmp_path): with ( patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=fake_clone), - patch(f"{self._URL_LOADER}.parse_url_ref", return_value=("https://github.com/org/my-skill", None)), + patch(f"{self._URL_LOADER}.parse_url_ref", return_value=("https://github.com/org/my-skill", None, None)), ): skills = Skill.from_url("https://github.com/org/my-skill") @@ -609,7 +609,7 @@ def test_from_url_multiple_skills(self, tmp_path): patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=fake_clone), patch( f"{self._URL_LOADER}.parse_url_ref", - return_value=("https://github.com/org/skills-collection", None), + return_value=("https://github.com/org/skills-collection", None, None), ), ): skills = Skill.from_url("https://github.com/org/skills-collection") @@ -625,7 +625,7 @@ def test_from_url_with_ref(self, tmp_path): fake_clone = self._mock_clone(tmp_path) captured_ref = [] - def tracking_clone(url, *, ref=None, cache_dir=None): + def tracking_clone(url, *, ref=None, subpath=None, cache_dir=None): captured_ref.append(ref) return fake_clone(url, ref=ref, cache_dir=cache_dir) @@ -633,7 +633,7 @@ def tracking_clone(url, *, ref=None, cache_dir=None): patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=tracking_clone), patch( f"{self._URL_LOADER}.parse_url_ref", - return_value=("https://github.com/org/my-skill", "v1.0.0"), + return_value=("https://github.com/org/my-skill", "v1.0.0", None), ), ): Skill.from_url("https://github.com/org/my-skill@v1.0.0") @@ -652,7 +652,7 @@ def test_from_url_clone_failure_raises(self): with ( patch( f"{self._URL_LOADER}.parse_url_ref", - return_value=("https://github.com/org/broken", None), + return_value=("https://github.com/org/broken", None, None), ), patch( f"{self._URL_LOADER}.clone_skill_repo", @@ -669,7 +669,7 @@ def test_from_url_passes_cache_dir(self, tmp_path): fake_clone = self._mock_clone(tmp_path) captured_cache = [] - def tracking_clone(url, *, ref=None, cache_dir=None): + def tracking_clone(url, *, ref=None, subpath=None, cache_dir=None): captured_cache.append(cache_dir) return fake_clone(url, ref=ref, cache_dir=cache_dir) @@ -679,7 +679,7 @@ def tracking_clone(url, *, ref=None, cache_dir=None): patch(f"{self._URL_LOADER}.clone_skill_repo", side_effect=tracking_clone), patch( f"{self._URL_LOADER}.parse_url_ref", - return_value=("https://github.com/org/my-skill", None), + return_value=("https://github.com/org/my-skill", None, None), ), ): Skill.from_url("https://github.com/org/my-skill", cache_dir=custom_cache) diff --git a/tests/strands/vended_plugins/skills/test_url_loader.py b/tests/strands/vended_plugins/skills/test_url_loader.py index 994672c32..0caf2b6d0 100644 --- a/tests/strands/vended_plugins/skills/test_url_loader.py +++ b/tests/strands/vended_plugins/skills/test_url_loader.py @@ -51,69 +51,123 @@ class TestParseUrlRef: """Tests for parse_url_ref.""" def test_https_no_ref(self): - url, ref = parse_url_ref("https://github.com/org/skill-repo") + url, ref, subpath = parse_url_ref("https://github.com/org/skill-repo") assert url == "https://github.com/org/skill-repo" assert ref is None + assert subpath is None def test_https_with_tag_ref(self): - url, ref = parse_url_ref("https://github.com/org/skill-repo@v1.0.0") + url, ref, subpath = parse_url_ref("https://github.com/org/skill-repo@v1.0.0") assert url == "https://github.com/org/skill-repo" assert ref == "v1.0.0" + assert subpath is None def test_https_with_branch_ref(self): - url, ref = parse_url_ref("https://github.com/org/skill-repo@main") + url, ref, subpath = parse_url_ref("https://github.com/org/skill-repo@main") assert url == "https://github.com/org/skill-repo" assert ref == "main" + assert subpath is None def test_https_git_suffix_no_ref(self): - url, ref = parse_url_ref("https://github.com/org/skill-repo.git") + url, ref, subpath = parse_url_ref("https://github.com/org/skill-repo.git") assert url == "https://github.com/org/skill-repo.git" assert ref is None + assert subpath is None def test_https_git_suffix_with_ref(self): - url, ref = parse_url_ref("https://github.com/org/skill-repo.git@v2.0") + url, ref, subpath = parse_url_ref("https://github.com/org/skill-repo.git@v2.0") assert url == "https://github.com/org/skill-repo.git" assert ref == "v2.0" + assert subpath is None def test_ssh_no_ref(self): - url, ref = parse_url_ref("git@github.com:org/skill-repo.git") + url, ref, subpath = parse_url_ref("git@github.com:org/skill-repo.git") assert url == "git@github.com:org/skill-repo.git" assert ref is None + assert subpath is None def test_ssh_with_ref(self): - url, ref = parse_url_ref("git@github.com:org/skill-repo.git@v1.0") + url, ref, subpath = parse_url_ref("git@github.com:org/skill-repo.git@v1.0") assert url == "git@github.com:org/skill-repo.git" assert ref == "v1.0" + assert subpath is None def test_ssh_no_git_suffix_with_ref(self): - url, ref = parse_url_ref("git@github.com:org/skill-repo@develop") + url, ref, subpath = parse_url_ref("git@github.com:org/skill-repo@develop") assert url == "git@github.com:org/skill-repo" assert ref == "develop" + assert subpath is None def test_ssh_protocol_no_ref(self): - url, ref = parse_url_ref("ssh://git@github.com/org/skill-repo") + url, ref, subpath = parse_url_ref("ssh://git@github.com/org/skill-repo") assert url == "ssh://git@github.com/org/skill-repo" assert ref is None + assert subpath is None def test_ssh_protocol_with_ref(self): - url, ref = parse_url_ref("ssh://git@github.com/org/skill-repo@v3") + url, ref, subpath = parse_url_ref("ssh://git@github.com/org/skill-repo@v3") assert url == "ssh://git@github.com/org/skill-repo" assert ref == "v3" + assert subpath is None def test_http_with_ref(self): - url, ref = parse_url_ref("http://gitlab.com/org/skill@feature-branch") + url, ref, subpath = parse_url_ref("http://gitlab.com/org/skill@feature-branch") assert url == "http://gitlab.com/org/skill" assert ref == "feature-branch" + assert subpath is None def test_url_host_only(self): - url, ref = parse_url_ref("https://example.com") + url, ref, subpath = parse_url_ref("https://example.com") assert url == "https://example.com" assert ref is None + assert subpath is None def test_non_url_passthrough(self): - url, ref = parse_url_ref("/local/path") + url, ref, subpath = parse_url_ref("/local/path") assert url == "/local/path" assert ref is None + assert subpath is None + + +class TestParseUrlRefGitHubTree: + """Tests for parse_url_ref with GitHub /tree/ and /blob/ URLs.""" + + def test_tree_with_subpath(self): + url, ref, subpath = parse_url_ref("https://github.com/org/repo/tree/main/skills/my-skill") + assert url == "https://github.com/org/repo" + assert ref == "main" + assert subpath == "skills/my-skill" + + def test_tree_branch_only(self): + url, ref, subpath = parse_url_ref("https://github.com/org/repo/tree/main") + assert url == "https://github.com/org/repo" + assert ref == "main" + assert subpath is None + + def test_tree_with_tag(self): + url, ref, subpath = parse_url_ref("https://github.com/org/repo/tree/v1.0.0/skills/brainstorming") + assert url == "https://github.com/org/repo" + assert ref == "v1.0.0" + assert subpath == "skills/brainstorming" + + def test_tree_deep_subpath(self): + url, ref, subpath = parse_url_ref("https://github.com/org/repo/tree/develop/a/b/c/d") + assert url == "https://github.com/org/repo" + assert ref == "develop" + assert subpath == "a/b/c/d" + + def test_blob_url(self): + """Test that /blob/ URLs are handled like /tree/.""" + url, ref, subpath = parse_url_ref("https://github.com/org/repo/blob/main/skills/my-skill") + assert url == "https://github.com/org/repo" + assert ref == "main" + assert subpath == "skills/my-skill" + + def test_tree_trailing_slash(self): + url, ref, subpath = parse_url_ref("https://github.com/org/repo/tree/main/skills/my-skill/") + assert url == "https://github.com/org/repo" + assert ref == "main" + assert subpath == "skills/my-skill" class TestCacheKey: @@ -268,6 +322,38 @@ def fake_clone(cmd, **kwargs): assert cache.exists() + def test_subpath_returns_subdirectory(self, tmp_path): + """Test that subpath parameter returns the subdirectory within the clone.""" + cache = tmp_path / "cache" + + def fake_clone(cmd, **kwargs): + target_dir = Path(cmd[-1]) + target_dir.mkdir(parents=True, exist_ok=True) + # Create a nested skill directory + skill_dir = target_dir / "skills" / "my-skill" + skill_dir.mkdir(parents=True) + (skill_dir / "SKILL.md").write_text("---\nname: my-skill\ndescription: test\n---\n") + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + with patch("strands.vended_plugins.skills._url_loader.subprocess.run", side_effect=fake_clone): + result = clone_skill_repo("https://github.com/org/repo", subpath="skills/my-skill", cache_dir=cache) + + assert result.name == "my-skill" + assert (result / "SKILL.md").exists() + + def test_subpath_nonexistent_raises(self, tmp_path): + """Test that a nonexistent subpath raises RuntimeError.""" + cache = tmp_path / "cache" + + def fake_clone(cmd, **kwargs): + target_dir = Path(cmd[-1]) + target_dir.mkdir(parents=True, exist_ok=True) + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + with patch("strands.vended_plugins.skills._url_loader.subprocess.run", side_effect=fake_clone): + with pytest.raises(RuntimeError, match="subdirectory does not exist"): + clone_skill_repo("https://github.com/org/repo", subpath="nonexistent/path", cache_dir=cache) + def test_shallow_clone_depth_one(self, tmp_path): """Test that --depth 1 is always passed.""" cache = tmp_path / "cache"