andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-02 21:58:11 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

feat: Skills Repository — registry, cache, CLI, and SDK integration (#5867)
Some checks failed
CodeQL Advanced / Analyze (actions) (push) Has been cancelled
Details
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Details
Vulnerability Scan / pip-audit (push) Has been cancelled
Details
Nightly Canary Release / Check for new commits (push) Has been cancelled
Details
Nightly Canary Release / Build nightly packages (push) Has been cancelled
Details
Nightly Canary Release / Publish nightly to PyPI (push) Has been cancelled
Details

Browse Source 
* feat: add Skills Repository — registry, cache, CLI, and SDK integration

Adds a Skills Repository feature allowing users to publish, install,
and use skills from the CrewAI registry with @org/skill-name refs.

## What's New

### SDK (lib/crewai/)
- SkillFrontmatter: added optional 'version' field (backward compatible)
- SkillCacheManager: manages ~/.crewai/skills/{org}/{name}/ with
  .crewai_meta.json tracking, path-traversal-safe tar extraction
- SkillRegistry: parse @org/skill-name refs, local-first resolution
  (./skills/ > cache > download), interactive prompt on first use,
  CI-mode guard (CREWAI_NONINTERACTIVE/CI env vars)
- Agent.skills and Crew.skills widened to accept str refs (@org/name)
- set_skills() resolves registry refs with org-prefixed dedup keys
- New events: SkillDownloadStartedEvent, SkillDownloadCompletedEvent

### CLI (lib/cli/)
- crewai skill create <name> — context-aware (project vs standalone)
- crewai skill install @org/name — downloads to ./skills/ or cache
- crewai skill publish — ZIP + upload to org registry
- crewai skill list — show installed skills

### PlusAPI (lib/crewai-core/)
- Added SKILLS_RESOURCE, get_skill(), publish_skill(), list_skills()

### Scaffolding
- crew and flow templates now include skills/ directory

### Tests
- 91 SDK skill tests + 15 CLI skill tests, all passing

* fix: address all CI failures and CodeRabbit review comments

Lint:
- Remove unused imports (click, pytest, json)
- Replace try-except-pass with logging (S110)
- Fix unprotected zipfile.extractall (S202)

Security:
- Path traversal: startswith → is_relative_to for tar extraction
- Add path traversal protection to ZIP extraction via _safe_extract_zip
- Both cache.py and CLI main.py hardened

Type checker:
- Fix import path: crewai.events.event_bus (not crewai_event_bus)
- Remove unused type: ignore comments
- Fix type mismatches in set_skills() variable types

Code quality:
- Fix f-string interpolation in SkillNotCachedError
- Use ValidationError instead of Exception in test

* style: ruff format + autofix remaining lint errors

* refactor: reuse SDK parser and SkillCacheManager in CLI

- _parse_frontmatter() now delegates to crewai.skills.parser.parse_frontmatter
  when available, with a minimal fallback for CLI-only installs
- install() global cache path now reuses SkillCacheManager.store() instead
  of duplicating metadata writing logic

* refactor: add _print_current_organization to SkillCommand (matches ToolCommand pattern)

* fix: write .crewai_meta.json in fallback install path

CodeRabbit caught that the ImportError fallback in install() didn't write
cache metadata, making skills invisible to 'crewai skill list'.

* fix: tighten @org/name ref validation to prevent path traversal

Reject refs with multiple slashes (@org/a/b), dot segments (@../skill),
or leading dots in org/name. Applied to both CLI install() and SDK
parse_registry_ref() so the contract is enforced consistently.

* fix: update test assertions to match tightened error messages

* fix: align OSS client with AMP API contract

- download_skill(): fetch download_url (presigned URL) instead of
  expecting inline base64. Falls back to 'file' field for compat.
- Read 'latest_version' field, fall back to 'version'
- Same fixes applied to CLI install() command

* fix: publish as tar.gz (matches AMP content_type validation) + add zip fallback to SDK cache

CLI publish:
- _build_skill_zip → _build_skill_tarball (tar.gz format)
- Content type: application/x-gzip (matches SkillVersion validation)

SDK cache:
- store() now tries tar.gz first, falls back to zip extraction
- Added _safe_extract_zip for path-traversal-safe zip handling
- Both formats work for download/install regardless of server format

---------

Co-authored-by: João Moura <joaomdmoura@gmail.com>
This commit is contained in:
alex-clawd
2026-05-20 10:38:25 -07:00
committed by GitHub
parent 7cc1a7bb41
commit 418afd29e7
20 changed files with 1439 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
lib/cli/src/crewai_cli/cli.py
View File

@@ -26,6 +26,7 @@ from crewai_cli.replay_from_task import replay_task_command
from crewai_cli.reset_memories_command import reset_memories_command
from crewai_cli.run_crew import run_crew
from crewai_cli.settings.main import SettingsCommand
from crewai_cli.skills.main import SkillCommand
from crewai_cli.task_outputs import load_task_outputs
from crewai_cli.tools.main import ToolCommand
from crewai_cli.train_crew import train_crew
@@ -546,6 +547,56 @@ def tool_publish(is_public: bool, force: bool) -> None:
tool_cmd.publish(is_public, force)
@crewai.group()
def skill() -> None:
"""Skill Repository related commands."""
@skill.command(name="create")
@click.argument("name")
@click.option(
"--no-project",
"in_project",
is_flag=True,
default=True,
flag_value=False,
help="Create skill in current dir instead of ./skills/",
)
def skill_create(name: str, in_project: bool) -> None:
skill_cmd = SkillCommand()
skill_cmd.create(name, in_project=in_project)
@skill.command(name="install")
@click.argument("ref")
def skill_install(ref: str) -> None:
skill_cmd = SkillCommand()
skill_cmd.install(ref)
@skill.command(name="publish")
@click.option(
"--force",
is_flag=True,
default=False,
show_default=True,
help="Skip git-state validation.",
)
@click.option("--public", "is_public", flag_value=True, default=False)
@click.option("--private", "is_public", flag_value=False)
@click.option("--org", default=None, help="Organisation slug (overrides settings).")
def skill_publish(is_public: bool, org: str | None, force: bool) -> None:
skill_cmd = SkillCommand()
skill_cmd.publish(is_public, org=org, force=force)
@skill.command(name="list")
def skill_list() -> None:
"""List locally installed skills."""
skill_cmd = SkillCommand()
skill_cmd.list_cached()
@crewai.group()
def template() -> None:
"""Browse and install project templates."""

2
lib/cli/src/crewai_cli/git.py
View File

@@ -40,7 +40,7 @@ class Repository:
encoding="utf-8",
).strip()
@cached_property # noqa: B019
@cached_property
def is_git_repo(self) -> bool:
"""Check if the current directory is a git repository."""
try:

0
lib/cli/src/crewai_cli/skills/__init__.py Normal file
View File

415
lib/cli/src/crewai_cli/skills/main.py Normal file
View File

@@ -0,0 +1,415 @@
"""Skill Repository CLI commands for CrewAI."""
from __future__ import annotations
import base64
import io
import json
import os
from pathlib import Path
import tarfile
import zipfile
from rich.console import Console
from rich.table import Table
from crewai_cli.command import BaseCommand, PlusAPIMixin
from crewai_cli.config import Settings
from crewai_cli.constants import DEFAULT_CREWAI_ENTERPRISE_URL
console = Console()
_SKILL_MD_TEMPLATE = """\
---
name: {name}
version: 0.1.0
description: |
A short description of what this skill does.
---
## Instructions
Describe the skill behaviour here. This section is shown to the agent at activation time.
"""
class SkillCommand(BaseCommand, PlusAPIMixin):
"""Skill Repository related operations for CrewAI projects."""
def __init__(self) -> None:
BaseCommand.__init__(self)
PlusAPIMixin.__init__(self, telemetry=self._telemetry)
# ------------------------------------------------------------------
# create
# ------------------------------------------------------------------
def create(self, name: str, in_project: bool = True) -> None:
"""Scaffold a new skill directory.
If pyproject.toml is present (crew project), creates ./skills/{name}/.
Otherwise creates ./{name}/.
"""
if in_project and os.path.isfile("pyproject.toml"):
skill_dir = Path("skills") / name
else:
skill_dir = Path(name)
if skill_dir.exists():
console.print(f"[red]Directory {skill_dir} already exists.[/red]")
raise SystemExit(1)
skill_dir.mkdir(parents=True)
(skill_dir / "scripts").mkdir()
(skill_dir / "references").mkdir()
(skill_dir / "assets").mkdir()
skill_md = skill_dir / "SKILL.md"
skill_md.write_text(_SKILL_MD_TEMPLATE.format(name=name))
console.print(
f"[green]Created skill [bold]{name}[/bold] at [bold]{skill_dir}[/bold].[/green]"
)
console.print(f"Edit [bold]{skill_md}[/bold] to define the skill instructions.")
# ------------------------------------------------------------------
# install
# ------------------------------------------------------------------
def install(self, ref: str) -> None:
"""Download and install a registry skill.
Format: @org/name
Inside a crew project (pyproject.toml present): installs to ./skills/{name}/
Outside a project: installs to ~/.crewai/skills/{org}/{name}/
"""
if not ref.startswith("@"):
console.print(
"[red]Invalid skill reference. Use the format @org/name.[/red]"
)
raise SystemExit(1)
without_at = ref[1:]
if without_at.count("/") != 1:
console.print(
"[red]Invalid skill reference. Use the format @org/name.[/red]"
)
raise SystemExit(1)
org, name = without_at.split("/", 1)
if (
not org
or not name
or org.startswith(".")
or name.startswith(".")
or len(Path(org).parts) != 1
or len(Path(name).parts) != 1
):
console.print(
"[red]Invalid skill reference: org and name must be single, "
"non-empty path segments (no slashes, no '..').[/red]"
)
raise SystemExit(1)
self._print_current_organization()
console.print(f"[bold blue]Downloading skill {ref}...[/bold blue]")
get_response = self.plus_api_client.get_skill(org, name)
if get_response.status_code == 404:
console.print(
f"[red]Skill {ref} not found. Ensure it has been published and you have access.[/red]"
)
raise SystemExit(1)
if get_response.status_code != 200:
console.print(
f"[red]Failed to download skill {ref}: {get_response.status_code}[/red]"
)
raise SystemExit(1)
data = get_response.json()
version = data.get("latest_version") or data.get("version")
download_url = data.get("download_url")
if download_url:
import httpx
dl_response = httpx.get(download_url, follow_redirects=True)
dl_response.raise_for_status()
archive_bytes = dl_response.content
else:
encoded = data.get("file", "")
if "," in encoded:
encoded = encoded.split(",", 1)[1]
archive_bytes = base64.b64decode(encoded)
in_project = os.path.isfile("pyproject.toml")
if in_project:
dest = Path("skills") / name
dest.mkdir(parents=True, exist_ok=True)
self._unpack_archive(archive_bytes, dest)
console.print(
f"[green]Installed [bold]{ref}[/bold]{' (' + version + ')' if version else ''} to [bold]{dest}[/bold].[/green]"
)
else:
try:
from crewai.skills.cache import SkillCacheManager
cache = SkillCacheManager()
cache.store(org, name, version, archive_bytes)
except ImportError:
# Fallback if SDK not installed — write directly
cache_dir = Path.home() / ".crewai" / "skills" / org / name
if cache_dir.exists():
import shutil
shutil.rmtree(cache_dir)
cache_dir.mkdir(parents=True, exist_ok=True)
self._unpack_archive(archive_bytes, cache_dir)
# Write metadata so `crewai skill list` can discover it
from datetime import datetime, timezone
meta = {
"org": org,
"name": name,
"version": version,
"installed_at": datetime.now(tz=timezone.utc).isoformat(),
}
(cache_dir / ".crewai_meta.json").write_text(json.dumps(meta, indent=2))
console.print(
f"[green]Installed [bold]{ref}[/bold]{' (' + version + ')' if version else ''} to global cache.[/green]"
)
# ------------------------------------------------------------------
# publish
# ------------------------------------------------------------------
def publish(self, is_public: bool, org: str | None, force: bool = False) -> None:
"""Publish the skill in the current directory to the registry."""
skill_md = Path("SKILL.md")
if not skill_md.exists():
console.print(
"[red]No SKILL.md found in current directory. "
"Run this command from inside a skill directory.[/red]"
)
raise SystemExit(1)
# Parse frontmatter to extract name + version
try:
frontmatter = self._parse_frontmatter(skill_md.read_text())
except ValueError as exc:
console.print(f"[red]Failed to parse SKILL.md frontmatter: {exc}[/red]")
raise SystemExit(1) from exc
name = frontmatter.get("name")
version = frontmatter.get("version")
description = frontmatter.get("description")
if not name:
console.print(
"[red]SKILL.md frontmatter must include a 'name' field.[/red]"
)
raise SystemExit(1)
if not version:
console.print(
"[red]SKILL.md frontmatter must include a 'version' field before publishing.[/red]"
)
raise SystemExit(1)
settings = Settings()
effective_org = org or settings.org_name
if not effective_org:
console.print(
"[red]No organisation set. Run `crewai org switch <org_id>` first, "
"or pass --org.[/red]"
)
raise SystemExit(1)
self._print_current_organization()
console.print(
f"[bold blue]Publishing skill [bold]{name}[/bold] v{version} to {effective_org}...[/bold blue]"
)
archive_bytes = self._build_skill_tarball()
encoded_file = "data:application/x-gzip;base64," + base64.b64encode(
archive_bytes
).decode("utf-8")
response = self.plus_api_client.publish_skill(
org=effective_org,
name=name,
version=version,
is_public=is_public,
description=description,
encoded_file=encoded_file,
)
self._validate_response(response)
base_url = settings.enterprise_base_url or DEFAULT_CREWAI_ENTERPRISE_URL
console.print(
f"[green]Published [bold]{effective_org}/{name}[/bold] v{version}.\n\n"
"Security checks are running in the background. "
"Your skill will be available once checks complete.\n"
f"Monitor status at: {base_url}/crewai_plus/skills/{effective_org}/{name}[/green]"
)
# ------------------------------------------------------------------
# list_cached
# ------------------------------------------------------------------
def list_cached(self) -> None:
"""Show locally installed skills."""
table = Table(title="Installed Skills", show_lines=True)
table.add_column("Source", style="dim")
table.add_column("Ref")
table.add_column("Version")
table.add_column("Path")
# Project-local ./skills/
local_skills_dir = Path("skills")
if local_skills_dir.is_dir():
for skill_dir in sorted(local_skills_dir.iterdir()):
if skill_dir.is_dir() and (skill_dir / "SKILL.md").exists():
version = self._read_version(skill_dir / "SKILL.md")
table.add_row(
"project",
skill_dir.name,
version or "-",
str(skill_dir),
)
# Global cache
cache_root = Path.home() / ".crewai" / "skills"
if cache_root.exists():
for org_dir in sorted(cache_root.iterdir()):
if not org_dir.is_dir():
continue
for skill_dir in sorted(org_dir.iterdir()):
meta_file = skill_dir / ".crewai_meta.json"
if meta_file.exists():
try:
meta = json.loads(meta_file.read_text())
table.add_row(
"cache",
f"@{meta['org']}/{meta['name']}",
meta.get("version") or "-",
str(skill_dir),
)
except (json.JSONDecodeError, KeyError):
console.print(
f"[yellow]Warning: skipping malformed cache entry at {meta_file}[/yellow]"
)
console.print(table)
# ------------------------------------------------------------------
# internal helpers
# ------------------------------------------------------------------
def _print_current_organization(self) -> None:
settings = Settings()
if settings.org_uuid:
console.print(
f"Current organization: {settings.org_name} ({settings.org_uuid})",
style="bold blue",
)
else:
console.print(
"No organization currently set. We recommend setting one before using: "
"`crewai org switch <org_id>` command.",
style="yellow",
)
def _unpack_archive(self, archive_bytes: bytes, dest: Path) -> None:
"""Unpack a .tar.gz or .zip archive into dest."""
# Try tar first, then zip
try:
with tarfile.open(fileobj=io.BytesIO(archive_bytes), mode="r:gz") as tf:
try:
tf.extractall(dest, filter="data")
except TypeError:
_safe_extractall(tf, dest)
return
except tarfile.TarError:
pass
# Fallback: zip
with zipfile.ZipFile(io.BytesIO(archive_bytes)) as zf:
_safe_extract_zip(zf, dest)
def _build_skill_tarball(self) -> bytes:
"""Build an in-memory .tar.gz of SKILL.md + scripts/ + references/ + assets/."""
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tf:
tf.add("SKILL.md")
for folder in ("scripts", "references", "assets"):
folder_path = Path(folder)
if folder_path.is_dir():
for fpath in sorted(folder_path.rglob("*")):
if fpath.is_file():
tf.add(str(fpath))
return buf.getvalue()
def _parse_frontmatter(self, content: str) -> dict[str, str]:
"""Extract YAML frontmatter fields from a SKILL.md string.
Reuses crewai.skills.parser when available, with a minimal
fallback for environments where the full SDK isn't installed.
"""
try:
from crewai.skills.parser import parse_frontmatter
fm_dict, _ = parse_frontmatter(content)
return fm_dict
except ImportError:
pass
# Fallback: minimal YAML parsing without SDK dependency
import re
match = re.match(r"^---\n(.*?)\n---", content, re.DOTALL)
if not match:
raise ValueError("No YAML frontmatter block found")
try:
import yaml
return yaml.safe_load(match.group(1)) or {}
except ImportError:
result: dict[str, str] = {}
for line in match.group(1).splitlines():
if ":" in line:
key, _, value = line.partition(":")
result[key.strip()] = value.strip()
return result
def _read_version(self, skill_md: Path) -> str | None:
"""Read the version field from a SKILL.md file, or None."""
try:
fm = self._parse_frontmatter(skill_md.read_text())
return fm.get("version")
except Exception:
return None
def _safe_extractall(tf: tarfile.TarFile, dest: Path) -> None:
"""Path-traversal-safe extraction for Python < 3.12."""
dest_resolved = dest.resolve()
for member in tf.getmembers():
member_path = (dest / member.name).resolve()
if not member_path.is_relative_to(dest_resolved):
raise ValueError(f"Blocked path traversal attempt: {member.name!r}")
tf.extractall(dest) # noqa: S202
def _safe_extract_zip(zf: zipfile.ZipFile, dest: Path) -> None:
"""Path-traversal-safe ZIP extraction."""
dest_resolved = dest.resolve()
for member in zf.namelist():
member_path = (dest / member).resolve()
if not member_path.is_relative_to(dest_resolved):
raise ValueError(f"Blocked path traversal attempt: {member!r}")
zf.extractall(dest) # noqa: S202

0
lib/cli/src/crewai_cli/templates/crew/skills/.gitkeep Normal file
View File

0
lib/cli/src/crewai_cli/templates/flow/skills/.gitkeep Normal file
View File

0
lib/cli/tests/skills/__init__.py Normal file
View File

205
lib/cli/tests/skills/test_main.py Normal file
View File

@@ -0,0 +1,205 @@
"""Tests for SkillCommand CLI."""
from __future__ import annotations
import io
import os
import tempfile
import zipfile
from contextlib import contextmanager
from datetime import datetime, timedelta
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from crewai_cli.shared.token_manager import TokenManager
@contextmanager
def in_temp_dir():
original = os.getcwd()
with tempfile.TemporaryDirectory() as td:
os.chdir(td)
try:
yield td
finally:
os.chdir(original)
@pytest.fixture
def skill_command():
with tempfile.TemporaryDirectory() as temp_dir:
with patch.object(
TokenManager, "_get_secure_storage_path", return_value=Path(temp_dir)
):
TokenManager().save_tokens(
"test-token", (datetime.now() + timedelta(seconds=36000)).timestamp()
)
from crewai_cli.skills.main import SkillCommand
cmd = SkillCommand()
yield cmd
# ---------------------------------------------------------------------------
# create
# ---------------------------------------------------------------------------
class TestSkillCreate:
def test_create_in_project(self, skill_command, tmp_path):
with in_temp_dir():
# Simulate being inside a project
Path("pyproject.toml").write_text("[tool.poetry]\nname = 'test'\n")
skill_command.create("my-skill")
assert Path("skills/my-skill/SKILL.md").exists()
assert Path("skills/my-skill/scripts").is_dir()
assert Path("skills/my-skill/references").is_dir()
assert Path("skills/my-skill/assets").is_dir()
def test_create_outside_project(self, skill_command, tmp_path):
with in_temp_dir():
skill_command.create("standalone-skill", in_project=False)
assert Path("standalone-skill/SKILL.md").exists()
def test_create_adds_name_to_skill_md(self, skill_command):
with in_temp_dir():
skill_command.create("hello-world", in_project=False)
content = Path("hello-world/SKILL.md").read_text()
assert "name: hello-world" in content
assert "version: 0.1.0" in content
def test_create_fails_if_dir_exists(self, skill_command):
with in_temp_dir():
Path("existing-skill").mkdir()
with pytest.raises(SystemExit):
skill_command.create("existing-skill", in_project=False)
# ---------------------------------------------------------------------------
# install
# ---------------------------------------------------------------------------
class TestSkillInstall:
def _zip_skill(self, name: str) -> bytes:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("SKILL.md", f"---\nname: {name}\ndescription: Test.\n---\nInstructions.")
return buf.getvalue()
def test_install_invalid_ref_no_at(self, skill_command):
with pytest.raises(SystemExit):
skill_command.install("acme/my-skill")
def test_install_invalid_ref_no_slash(self, skill_command):
with pytest.raises(SystemExit):
skill_command.install("@acmeskill")
def test_install_404(self, skill_command):
mock_resp = MagicMock()
mock_resp.status_code = 404
skill_command.plus_api_client.get_skill = MagicMock(return_value=mock_resp)
with pytest.raises(SystemExit):
skill_command.install("@acme/ghost")
def test_install_in_project(self, skill_command):
import base64
archive = self._zip_skill("my-skill")
encoded = "data:application/zip;base64," + base64.b64encode(archive).decode()
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"file": encoded, "version": "1.0.0"}
skill_command.plus_api_client.get_skill = MagicMock(return_value=mock_resp)
with in_temp_dir():
Path("pyproject.toml").write_text("[tool]\n")
skill_command.install("@acme/my-skill")
assert Path("skills/my-skill/SKILL.md").exists()
# ---------------------------------------------------------------------------
# publish
# ---------------------------------------------------------------------------
class TestSkillPublish:
def test_publish_no_skill_md(self, skill_command):
with in_temp_dir():
with pytest.raises(SystemExit):
skill_command.publish(is_public=True, org="acme")
def test_publish_missing_version(self, skill_command):
with in_temp_dir():
Path("SKILL.md").write_text(
"---\nname: my-skill\ndescription: Test.\n---\nInstructions."
)
with pytest.raises(SystemExit):
skill_command.publish(is_public=True, org="acme")
def test_publish_missing_name(self, skill_command):
with in_temp_dir():
Path("SKILL.md").write_text(
"---\ndescription: Test.\nversion: 1.0.0\n---\nInstructions."
)
with pytest.raises(SystemExit):
skill_command.publish(is_public=True, org="acme")
def test_publish_no_org(self, skill_command):
with in_temp_dir():
Path("SKILL.md").write_text(
"---\nname: my-skill\nversion: 1.0.0\ndescription: Test.\n---\nInstructions."
)
with patch.object(skill_command, "plus_api_client") as mock_client:
mock_resp = MagicMock()
mock_resp.is_success = True
mock_resp.status_code = 200
mock_resp.json.return_value = {}
mock_client.publish_skill.return_value = mock_resp
# No org set → should SystemExit (no org_name in settings)
with patch("crewai_cli.skills.main.Settings") as mock_settings_cls:
mock_settings_cls.return_value.org_name = None
mock_settings_cls.return_value.enterprise_base_url = None
with pytest.raises(SystemExit):
skill_command.publish(is_public=True, org=None)
def test_publish_calls_api(self, skill_command):
with in_temp_dir():
Path("SKILL.md").write_text(
"---\nname: my-skill\nversion: 1.0.0\ndescription: A test skill.\n---\nInstructions."
)
mock_resp = MagicMock()
mock_resp.is_success = True
mock_resp.status_code = 200
mock_resp.json.return_value = {}
skill_command.plus_api_client.publish_skill = MagicMock(return_value=mock_resp)
with patch("crewai_cli.skills.main.Settings") as mock_settings_cls:
mock_settings_cls.return_value.org_name = "acme"
mock_settings_cls.return_value.enterprise_base_url = None
skill_command.publish(is_public=False, org="acme")
skill_command.plus_api_client.publish_skill.assert_called_once()
call_kwargs = skill_command.plus_api_client.publish_skill.call_args
assert call_kwargs.kwargs["name"] == "my-skill"
assert call_kwargs.kwargs["version"] == "1.0.0"
# ---------------------------------------------------------------------------
# list_cached
# ---------------------------------------------------------------------------
class TestSkillListCached:
def test_list_cached_empty(self, skill_command, capsys):
with in_temp_dir():
skill_command.list_cached()
# Should not raise
def test_list_cached_shows_project_skills(self, skill_command, capsys):
with in_temp_dir():
skill_dir = Path("skills/my-skill")
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text(
"---\nname: my-skill\nversion: 0.5.0\ndescription: A skill.\n---\nBody."
)
skill_command.list_cached()
# Should complete without error

42
lib/crewai-core/src/crewai_core/plus_api.py
View File

@@ -140,6 +140,7 @@ class PlusAPI:
"""Client for working with the CrewAI+ API."""
TOOLS_RESOURCE: Final = "/crewai_plus/api/v1/tools"
SKILLS_RESOURCE: Final = "/crewai_plus/api/v1/skills"
ORGANIZATIONS_RESOURCE: Final = "/crewai_plus/api/v1/me/organizations"
CREWS_RESOURCE: Final = "/crewai_plus/api/v1/crews"
AGENTS_RESOURCE: Final = "/crewai_plus/api/v1/agents"
@@ -228,6 +229,47 @@ class PlusAPI:
}
return self._make_request("POST", f"{self.TOOLS_RESOURCE}", json=params)
def get_skill(
self, org: str, name: str, version: str | None = None
) -> httpx.Response:
params: dict[str, str] = {}
if version is not None:
params["version"] = version
return self._make_request(
"GET",
f"{self.SKILLS_RESOURCE}/{org}/{name}",
params=params or None,
)
def publish_skill(
self,
org: str,
name: str,
version: str,
is_public: bool,
description: str | None,
encoded_file: str,
) -> httpx.Response:
payload = {
"org": org,
"name": name,
"version": version,
"public": is_public,
"description": description,
"file": encoded_file,
}
return self._make_request("POST", self.SKILLS_RESOURCE, json=payload)
def list_skills(self, org: str | None = None) -> httpx.Response:
params: dict[str, str] = {}
if org is not None:
params["org"] = org
return self._make_request(
"GET",
self.SKILLS_RESOURCE,
params=params or None,
)
def deploy_by_name(self, project_name: str) -> httpx.Response:
return self._make_request(
"POST", f"{self.CREWS_RESOURCE}/by-name/{project_name}/deploy"

24
lib/crewai/src/crewai/agent/core.py
View File

@@ -434,7 +434,7 @@ class Agent(BaseAgent):
from crewai.crew import Crew
if resolved_crew_skills is None:
crew_skills: list[Path | SkillModel] | None = (
crew_skills: list[Path | SkillModel | str] | None = (
self.crew.skills
if isinstance(self.crew, Crew) and isinstance(self.crew.skills, list)
else None
@@ -446,7 +446,7 @@ class Agent(BaseAgent):
return
needs_work = self.skills and any(
isinstance(s, Path)
isinstance(s, (Path, str))
or (isinstance(s, SkillModel) and s.disclosure_level < INSTRUCTIONS)
for s in self.skills
)
@@ -454,14 +454,28 @@ class Agent(BaseAgent):
return
seen: set[str] = set()
resolved: list[Path | SkillModel] = []
items: list[Path | SkillModel] = list(self.skills) if self.skills else []
resolved: list[Path | SkillModel | str] = []
items: list[Path | SkillModel | str] = list(self.skills) if self.skills else []
if crew_skills:
items.extend(crew_skills)
for item in items:
if isinstance(item, Path):
if isinstance(item, str):
from crewai.skills.registry import (
is_registry_ref,
parse_registry_ref,
resolve_registry_ref,
)
if is_registry_ref(item):
skill = resolve_registry_ref(item, source=self)
org, _ = parse_registry_ref(item)
dedup_key = f"{org}/{skill.name}"
if dedup_key not in seen:
seen.add(dedup_key)
resolved.append(skill)
elif isinstance(item, Path):
discovered = discover_skills(item, source=self)
for skill in discovered:
if skill.name not in seen:

18
lib/crewai/src/crewai/agents/agent_builder/base_agent.py
View File

@@ -334,9 +334,9 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
"If not set, falls back to crew memory."
),
)
skills: list[Path | Skill] | None = Field(
skills: list[Path | Skill | str] | None = Field(
default=None,
description="Agent Skills. Accepts paths for discovery or pre-loaded Skill objects.",
description="Agent Skills. Accepts paths for discovery, pre-loaded Skill objects, or '@org/name' registry refs.",
min_length=1,
)
execution_context: ExecutionContext | None = Field(default=None)
@@ -429,6 +429,20 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
def process_model_config(cls, values: Any) -> dict[str, Any]:
return process_config(values, cls)
@field_validator("skills", mode="before")
@classmethod
def coerce_skill_strings(cls, skills: Any) -> Any:
"""Coerce plain path strings to Path objects; keep @-prefixed refs as str."""
if not isinstance(skills, list):
return skills
result = []
for item in skills:
if isinstance(item, str) and not item.startswith("@"):
result.append(Path(item))
else:
result.append(item)
return result
@field_validator("tools")
@classmethod
def validate_tools(cls, tools: list[Any]) -> list[BaseTool]:

18
lib/crewai/src/crewai/crew.py
View File

@@ -341,9 +341,9 @@ class Crew(FlowTrackable, BaseModel):
default=None,
description="Knowledge for the crew.",
)
skills: list[Path | Skill] | None = Field(
skills: list[Path | Skill | str] | None = Field(
default=None,
description="Skill search paths or pre-loaded Skill objects applied to all agents in the crew.",
description="Skill search paths, pre-loaded Skill objects, or '@org/name' registry refs applied to all agents in the crew.",
)
security_config: SecurityConfig = Field(
@@ -526,6 +526,20 @@ class Crew(FlowTrackable, BaseModel):
if max_seq > 0:
set_emission_counter(max_seq)
@field_validator("skills", mode="before")
@classmethod
def coerce_skill_strings(cls, skills: Any) -> Any:
"""Coerce plain path strings to Path objects; keep @-prefixed refs as str."""
if not isinstance(skills, list):
return skills
result = []
for item in skills:
if isinstance(item, str) and not item.startswith("@"):
result.append(Path(item))
else:
result.append(item)
return result
@field_validator("id", mode="before")
@classmethod
def _deny_user_set_id(cls, v: UUID4 | None, info: Any) -> UUID4 | None:

17
lib/crewai/src/crewai/events/types/skill_events.py
View File

@@ -60,3 +60,20 @@ class SkillLoadFailedEvent(SkillEvent):
type: Literal["skill_load_failed"] = "skill_load_failed"
error: str
class SkillDownloadStartedEvent(SkillEvent):
"""Event emitted when a registry skill download begins."""
type: Literal["skill_download_started"] = "skill_download_started"
registry_ref: str
version: str | None = None
class SkillDownloadCompletedEvent(SkillEvent):
"""Event emitted when a registry skill download completes."""
type: Literal["skill_download_completed"] = "skill_download_completed"
registry_ref: str
version: str | None = None
cache_path: Path | None = None

5
lib/crewai/src/crewai/skills/__init__.py
View File

@@ -3,15 +3,20 @@
Provides filesystem-based skill packaging with progressive disclosure.
"""
from crewai.skills.cache import SkillCacheManager
from crewai.skills.loader import activate_skill, discover_skills
from crewai.skills.models import Skill, SkillFrontmatter
from crewai.skills.parser import SkillParseError
from crewai.skills.registry import is_registry_ref, resolve_registry_ref
__all__ = [
"Skill",
"SkillCacheManager",
"SkillFrontmatter",
"SkillParseError",
"activate_skill",
"discover_skills",
"is_registry_ref",
"resolve_registry_ref",
]

148
lib/crewai/src/crewai/skills/cache.py Normal file
View File

@@ -0,0 +1,148 @@
"""Cache manager for registry-downloaded skills.
Manages ~/.crewai/skills/{org}/{name}/ as the global skill cache.
One version is stored per skill (last install wins).
"""
from __future__ import annotations
from datetime import datetime, timezone
import json
import logging
from pathlib import Path
import tarfile
from typing import TypedDict
import zipfile
_logger = logging.getLogger(__name__)
_CACHE_ROOT = Path.home() / ".crewai" / "skills"
_META_FILENAME = ".crewai_meta.json"
class SkillMetadata(TypedDict):
org: str
name: str
version: str | None
installed_at: str
class SkillCacheManager:
"""Manages the global skill cache at ~/.crewai/skills/."""
def __init__(self, cache_root: Path | None = None) -> None:
self._root = cache_root or _CACHE_ROOT
def _skill_dir(self, org: str, name: str) -> Path:
return self._root / org / name
def get_cached_path(self, org: str, name: str) -> Path | None:
"""Return the cached skill directory path if it exists, else None."""
skill_dir = self._skill_dir(org, name)
meta_file = skill_dir / _META_FILENAME
if skill_dir.is_dir() and meta_file.exists():
return skill_dir
return None
def store(
self, org: str, name: str, version: str | None, archive_bytes: bytes
) -> Path:
"""Unpack an archive into the cache and write metadata.
Uses tarfile with filter='data' for path-traversal protection.
Args:
org: Organisation slug.
name: Skill name.
version: Semantic version string, or None if unknown.
archive_bytes: Raw bytes of a .tar.gz archive.
Returns:
Path to the stored skill directory.
"""
skill_dir = self._skill_dir(org, name)
# Wipe any previous version
if skill_dir.exists():
import shutil
shutil.rmtree(skill_dir)
skill_dir.mkdir(parents=True, exist_ok=True)
import io
# Try tar.gz first, fall back to zip
try:
with tarfile.open(fileobj=io.BytesIO(archive_bytes), mode="r:gz") as tf:
try:
tf.extractall(skill_dir, filter="data")
except TypeError:
_safe_extractall(tf, skill_dir)
except tarfile.TarError:
with zipfile.ZipFile(io.BytesIO(archive_bytes)) as zf:
_safe_extract_zip(zf, skill_dir)
meta: SkillMetadata = {
"org": org,
"name": name,
"version": version,
"installed_at": datetime.now(tz=timezone.utc).isoformat(),
}
(skill_dir / _META_FILENAME).write_text(json.dumps(meta, indent=2))
return skill_dir
def list_cached(self) -> list[SkillMetadata]:
"""Return metadata for every cached skill."""
results: list[SkillMetadata] = []
if not self._root.exists():
return results
for org_dir in sorted(self._root.iterdir()):
if not org_dir.is_dir():
continue
for skill_dir in sorted(org_dir.iterdir()):
meta_file = skill_dir / _META_FILENAME
if meta_file.exists():
try:
results.append(json.loads(meta_file.read_text()))
except (json.JSONDecodeError, KeyError):
_logger.debug(
"Skipping malformed cache entry: %s",
meta_file,
exc_info=True,
)
return results
def invalidate(self, org: str, name: str) -> bool:
"""Remove a cached skill.
Returns:
True if the cache entry existed and was removed, False otherwise.
"""
skill_dir = self._skill_dir(org, name)
if skill_dir.exists():
import shutil
shutil.rmtree(skill_dir)
return True
return False
def _safe_extractall(tf: tarfile.TarFile, dest: Path) -> None:
"""Path-traversal-safe extraction for Python < 3.12."""
dest_resolved = dest.resolve()
for member in tf.getmembers():
member_path = (dest / member.name).resolve()
if not member_path.is_relative_to(dest_resolved):
raise ValueError(f"Blocked path traversal attempt: {member.name!r}")
tf.extractall(dest) # noqa: S202
def _safe_extract_zip(zf: zipfile.ZipFile, dest: Path) -> None:
"""Path-traversal-safe ZIP extraction."""
dest_resolved = dest.resolve()
for member in zf.namelist():
member_path = (dest / member).resolve()
if not member_path.is_relative_to(dest_resolved):
raise ValueError(f"Blocked path traversal attempt: {member!r}")
zf.extractall(dest) # noqa: S202

4
lib/crewai/src/crewai/skills/models.py
View File

@@ -78,6 +78,10 @@ class SkillFrontmatter(BaseModel):
alias="allowed-tools",
description="Pre-approved tool names the skill may use, parsed from a space-delimited string in frontmatter.",
)
version: str | None = Field(
default=None,
description="Semantic version of the skill, e.g. '1.0.0'. Optional for local skills.",
)
@model_validator(mode="before")
@classmethod

223
lib/crewai/src/crewai/skills/registry.py Normal file
View File

@@ -0,0 +1,223 @@
"""Registry reference resolution for the Agent Skills standard.
Handles @org/skill-name references, local-first resolution, and downloads
via the CrewAI+ API with a global cache at ~/.crewai/skills/.
"""
from __future__ import annotations
import logging
from pathlib import Path
import sys
from typing import Any
from crewai.skills.cache import SkillCacheManager
_logger = logging.getLogger(__name__)
class SkillNotCachedError(Exception):
"""Raised when a registry skill is not cached and the environment is non-interactive."""
def __init__(self, ref: str) -> None:
super().__init__(
f"Skill {ref!r} is not cached locally. "
f"Run `crewai skill install {ref}` to install it first."
)
self.ref = ref
def is_registry_ref(value: Any) -> bool:
"""Return True if *value* looks like a registry reference (@org/name)."""
return isinstance(value, str) and value.startswith("@")
def parse_registry_ref(ref: str) -> tuple[str, str]:
"""Parse '@org/skill-name' into (org, name).
Args:
ref: A registry reference, e.g. '@acme/my-skill'.
Returns:
A (org, name) tuple.
Raises:
ValueError: If the reference format is invalid.
"""
if not ref.startswith("@"):
raise ValueError(f"Registry reference must start with '@', got: {ref!r}")
without_at = ref[1:]
if without_at.count("/") != 1:
raise ValueError(
f"Registry reference must be in '@org/name' format, got: {ref!r}"
)
org, name = without_at.split("/", 1)
if (
not org
or not name
or org.startswith(".")
or name.startswith(".")
or "/" in org
or "/" in name
):
raise ValueError(
f"Registry reference org and name must be single, non-empty path "
f"segments (no '..' or leading dots), got: {ref!r}"
)
return org, name
def _is_noninteractive() -> bool:
"""Return True in CI or explicitly non-interactive environments."""
import os
return (
os.environ.get("CI") == "1"
or os.environ.get("CREWAI_NONINTERACTIVE") == "1"
or not sys.stdin.isatty()
)
def resolve_registry_ref(
ref: str,
source: Any = None,
) -> Skill: # type: ignore[name-defined] # noqa: F821
"""Resolve a registry reference to a Skill object.
Resolution order:
1. ./skills/{name}/ in the current working directory (project-local)
2. ~/.crewai/skills/{org}/{name}/ (global cache)
3. Download from registry (interactive only; raises SkillNotCachedError in CI)
Args:
ref: A registry reference, e.g. '@acme/my-skill'.
source: Optional source object passed through to skill loaders (for events).
Returns:
A Skill loaded at INSTRUCTIONS disclosure level.
Raises:
SkillNotCachedError: When not cached and running in non-interactive mode.
"""
from crewai.skills.loader import activate_skill
from crewai.skills.parser import load_skill_metadata
org, name = parse_registry_ref(ref)
# 1. Project-local: ./skills/{name}/
local_path = Path.cwd() / "skills" / name
if local_path.is_dir() and (local_path / "SKILL.md").exists():
try:
skill = load_skill_metadata(local_path)
return activate_skill(skill, source=source)
except Exception:
_logger.debug("Failed to load local skill at %s", local_path, exc_info=True)
# 2. Global cache
cache = SkillCacheManager()
cached_path = cache.get_cached_path(org, name)
if cached_path is not None and (cached_path / "SKILL.md").exists():
try:
skill = load_skill_metadata(cached_path)
return activate_skill(skill, source=source)
except Exception:
_logger.debug(
"Failed to load cached skill at %s", cached_path, exc_info=True
)
# 3. Download
if _is_noninteractive():
raise SkillNotCachedError(ref)
return download_skill(org, name, source=source)
def download_skill(
org: str,
name: str,
source: Any = None,
) -> Skill: # type: ignore[name-defined] # noqa: F821
"""Download a skill from the registry and store it in the cache.
Args:
org: Organisation slug.
name: Skill name.
source: Optional source for event emission.
Returns:
The downloaded Skill at INSTRUCTIONS level.
"""
from crewai.skills.loader import activate_skill
from crewai.skills.parser import load_skill_metadata
ref = f"@{org}/{name}"
try:
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.skill_events import (
SkillDownloadCompletedEvent,
SkillDownloadStartedEvent,
)
_has_events = True
except ImportError:
_has_events = False
if _has_events:
crewai_event_bus.emit(
source,
event=SkillDownloadStartedEvent(
registry_ref=ref,
),
)
try:
from crewai_core.plus_api import PlusAPI
api = PlusAPI()
response = api.get_skill(org, name)
response.raise_for_status()
data = response.json()
except Exception as exc:
raise RuntimeError(
f"Failed to download skill {ref!r} from registry: {exc}"
) from exc
import base64
import httpx
version = data.get("latest_version") or data.get("version")
download_url = data.get("download_url")
if download_url:
dl_response = httpx.get(download_url, follow_redirects=True)
dl_response.raise_for_status()
archive_bytes = dl_response.content
else:
encoded = data.get("file", "")
# Strip data URI prefix if present
if "," in encoded:
encoded = encoded.split(",", 1)[1]
archive_bytes = base64.b64decode(encoded)
cache = SkillCacheManager()
skill_dir = cache.store(org, name, version, archive_bytes)
if _has_events:
crewai_event_bus.emit(
source,
event=SkillDownloadCompletedEvent(
registry_ref=ref,
version=version,
cache_path=skill_dir,
),
)
if not (skill_dir / "SKILL.md").exists():
raise RuntimeError(
f"Skill archive for {ref!r} downloaded but no SKILL.md found in {skill_dir}"
)
skill = load_skill_metadata(skill_dir)
return activate_skill(skill, source=source)

116
lib/crewai/tests/skills/test_cache.py Normal file
View File

@@ -0,0 +1,116 @@
"""Tests for SkillCacheManager."""
from __future__ import annotations
import gzip
import io
import json
import tarfile
from pathlib import Path
from crewai.skills.cache import SkillCacheManager
def _make_tar_gz(files: dict[str, str]) -> bytes:
"""Build an in-memory .tar.gz containing the given filename → content mapping."""
buf = io.BytesIO()
with gzip.GzipFile(fileobj=buf, mode="wb") as gz:
gz_buf = io.BytesIO()
with tarfile.open(fileobj=gz_buf, mode="w") as tf:
for name, content in files.items():
data = content.encode()
info = tarfile.TarInfo(name=name)
info.size = len(data)
tf.addfile(info, io.BytesIO(data))
gz.write(gz_buf.getvalue())
buf.seek(0)
# Re-create properly: gzip wrapping a tar stream
out = io.BytesIO()
with tarfile.open(fileobj=out, mode="w:gz") as tf:
for name, content in files.items():
data = content.encode()
info = tarfile.TarInfo(name=name)
info.size = len(data)
tf.addfile(info, io.BytesIO(data))
return out.getvalue()
class TestSkillCacheManager:
def test_get_cached_path_missing(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
assert cache.get_cached_path("acme", "my-skill") is None
def test_store_and_retrieve(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
archive = _make_tar_gz({"SKILL.md": "---\nname: my-skill\n---\nHello"})
dest = cache.store("acme", "my-skill", "1.0.0", archive)
assert dest.is_dir()
assert (dest / "SKILL.md").exists()
retrieved = cache.get_cached_path("acme", "my-skill")
assert retrieved == dest
def test_store_writes_metadata(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
archive = _make_tar_gz({"SKILL.md": "content"})
dest = cache.store("acme", "my-skill", "2.3.4", archive)
meta_file = dest / ".crewai_meta.json"
assert meta_file.exists()
meta = json.loads(meta_file.read_text())
assert meta["org"] == "acme"
assert meta["name"] == "my-skill"
assert meta["version"] == "2.3.4"
assert "installed_at" in meta
def test_store_overwrites_previous_version(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
archive_v1 = _make_tar_gz({"SKILL.md": "v1", "extra.txt": "old"})
cache.store("acme", "my-skill", "1.0.0", archive_v1)
archive_v2 = _make_tar_gz({"SKILL.md": "v2"})
dest = cache.store("acme", "my-skill", "2.0.0", archive_v2)
# Old file should be gone
assert not (dest / "extra.txt").exists()
assert (dest / "SKILL.md").read_text() == "v2"
meta = json.loads((dest / ".crewai_meta.json").read_text())
assert meta["version"] == "2.0.0"
def test_list_cached_empty(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
assert cache.list_cached() == []
def test_list_cached(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
archive = _make_tar_gz({"SKILL.md": "x"})
cache.store("acme", "skill-a", "1.0.0", archive)
cache.store("acme", "skill-b", "0.1.0", archive)
cache.store("other-org", "skill-c", None, archive)
entries = cache.list_cached()
names = {e["name"] for e in entries}
assert names == {"skill-a", "skill-b", "skill-c"}
def test_invalidate_existing(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
archive = _make_tar_gz({"SKILL.md": "x"})
cache.store("acme", "my-skill", "1.0.0", archive)
removed = cache.invalidate("acme", "my-skill")
assert removed is True
assert cache.get_cached_path("acme", "my-skill") is None
def test_invalidate_missing(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
removed = cache.invalidate("acme", "ghost-skill")
assert removed is False
def test_store_version_none(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
archive = _make_tar_gz({"SKILL.md": "x"})
dest = cache.store("acme", "my-skill", None, archive)
meta = json.loads((dest / ".crewai_meta.json").read_text())
assert meta["version"] is None

32
lib/crewai/tests/skills/test_models_version.py Normal file
View File

@@ -0,0 +1,32 @@
"""Tests for the version field added to SkillFrontmatter."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from crewai.skills.models import SkillFrontmatter
class TestSkillFrontmatterVersion:
def test_version_defaults_to_none(self) -> None:
fm = SkillFrontmatter(name="my-skill", description="A skill.")
assert fm.version is None
def test_version_can_be_set(self) -> None:
fm = SkillFrontmatter(name="my-skill", description="A skill.", version="1.2.3")
assert fm.version == "1.2.3"
def test_existing_frontmatter_without_version_still_valid(self) -> None:
"""Backward compat: existing SKILL.md files without version must still parse."""
fm = SkillFrontmatter(name="old-skill", description="Old skill without version.")
assert fm.version is None
def test_version_is_optional_string(self) -> None:
fm = SkillFrontmatter(name="my-skill", description="Desc.", version=None)
assert fm.version is None
def test_frontmatter_is_frozen(self) -> None:
fm = SkillFrontmatter(name="my-skill", description="A skill.", version="1.0.0")
with pytest.raises(ValidationError):
fm.version = "2.0.0" # type: ignore[misc]

129
lib/crewai/tests/skills/test_registry.py Normal file
View File

@@ -0,0 +1,129 @@
"""Tests for SkillRegistry."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from crewai.skills.registry import (
SkillNotCachedError,
is_registry_ref,
parse_registry_ref,
)
class TestIsRegistryRef:
def test_at_prefixed(self) -> None:
assert is_registry_ref("@acme/my-skill") is True
def test_plain_string(self) -> None:
assert is_registry_ref("my-skill") is False
def test_path_like_string(self) -> None:
assert is_registry_ref("./skills/my-skill") is False
def test_non_string(self) -> None:
assert is_registry_ref(None) is False
assert is_registry_ref(42) is False
assert is_registry_ref(Path("something")) is False
class TestParseRegistryRef:
def test_valid(self) -> None:
assert parse_registry_ref("@acme/my-skill") == ("acme", "my-skill")
def test_valid_with_dashes(self) -> None:
assert parse_registry_ref("@my-org/cool-skill") == ("my-org", "cool-skill")
def test_missing_at(self) -> None:
with pytest.raises(ValueError, match="must start with '@'"):
parse_registry_ref("acme/my-skill")
def test_missing_slash(self) -> None:
with pytest.raises(ValueError, match="'@org/name' format"):
parse_registry_ref("@acme-skill")
def test_empty_org(self) -> None:
with pytest.raises(ValueError, match="non-empty"):
parse_registry_ref("@/my-skill")
def test_empty_name(self) -> None:
with pytest.raises(ValueError, match="non-empty"):
parse_registry_ref("@acme/")
class TestResolveRegistryRef:
"""Test resolution order and CI mode behaviour."""
def _make_skill_dir(self, base: Path, name: str) -> Path:
"""Write a minimal SKILL.md into base/name/."""
skill_dir = base / name
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: Test skill.\n---\n\nInstructions."
)
return skill_dir
def test_resolves_project_local(self, tmp_path: Path) -> None:
"""Local ./skills/{name}/ takes priority over cache."""
skills_dir = tmp_path / "skills"
skills_dir.mkdir()
self._make_skill_dir(skills_dir, "my-skill")
# Mock SkillCacheManager to return None (not cached) so only local is hit
mock_cache = MagicMock()
mock_cache.get_cached_path.return_value = None
with (
patch("crewai.skills.registry._is_noninteractive", return_value=False),
patch.object(Path, "cwd", return_value=tmp_path),
patch("crewai.skills.registry.SkillCacheManager", return_value=mock_cache),
):
from crewai.skills.registry import resolve_registry_ref
skill = resolve_registry_ref("@acme/my-skill")
assert skill.name == "my-skill"
def test_raises_in_ci_when_not_cached(self, tmp_path: Path) -> None:
"""In CI mode, raise SkillNotCachedError if no local or cached copy."""
mock_cache = MagicMock()
mock_cache.get_cached_path.return_value = None
with (
patch("crewai.skills.registry._is_noninteractive", return_value=True),
patch.object(Path, "cwd", return_value=tmp_path),
patch("crewai.skills.registry.SkillCacheManager", return_value=mock_cache),
):
from crewai.skills.registry import resolve_registry_ref
with pytest.raises(SkillNotCachedError) as exc_info:
resolve_registry_ref("@acme/ghost-skill")
assert "@acme/ghost-skill" in str(exc_info.value)
def test_resolves_from_cache(self, tmp_path: Path) -> None:
"""Falls back to global cache when no project-local skill exists."""
cache_dir = tmp_path / "acme" / "cached-skill"
cache_dir.mkdir(parents=True)
(cache_dir / "SKILL.md").write_text(
"---\nname: cached-skill\ndescription: Cached.\n---\n\nCached instructions."
)
mock_cache = MagicMock()
mock_cache.get_cached_path.return_value = cache_dir
# tmp_path has no ./skills/ directory
with (
patch("crewai.skills.registry._is_noninteractive", return_value=False),
patch.object(Path, "cwd", return_value=tmp_path),
patch("crewai.skills.registry.SkillCacheManager", return_value=mock_cache),
):
from crewai.skills.registry import resolve_registry_ref
skill = resolve_registry_ref("@acme/cached-skill")
assert skill.name == "cached-skill"
def test_skill_not_cached_error_contains_ref(self) -> None:
err = SkillNotCachedError("@foo/bar")
assert "@foo/bar" in str(err)
assert err.ref == "@foo/bar"