andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-27 01:35:12 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

fix(skills): resolve registry skills through the runtime's CrewAI+ client (#6658)

Browse Source 
* fix(skills): resolve registry skills through the installed AMP client

Skill downloads built their own `PlusAPI` and authenticated it from
`CREWAI_USER_PAT`, the platform integration token, or the saved CLI login.
Managed runtimes have no user credential to offer: they install a client of
their own, which `load_agent_from_repository` already resolves through, so
Agent Repository lookups worked while the skill downloads beside them failed
with 401.

Skills now resolve their client the same way, via `resolve_plus_client()` next
to the hook it reads. A client that can't fetch skills falls back to
environment credentials and warns, so older runtimes behave as they do today.
`resolve_plus_response()` shares the sync/async bridging both lookups need,
since `PlusAPI` is synchronous while managed clients are not.

Version pinning, which the same bug was hiding:

- Registry refs accept `@org/name@version`, and `@org/name@v1.2.0` since people
  write it both ways. `parse_skill_ref()` returns a `SkillRef(org, name,
  version)`; `parse_registry_ref()` keeps its `(org, name)` shape and drops the
  pin, so existing callers are unaffected
- Agent Repository agents record a version per skill, which was parsed off the
  response and dropped. Those pins now travel with the refs, so publishing a
  new version of a skill no longer changes every agent that uses it
- A pinned ref only accepts a project-local copy declaring that version in its
  `metadata.version` frontmatter, and the cache reports a miss when the version
  it recorded differs — so a pin re-resolves rather than loading another
  version. Unpinned refs keep hitting the cache as before
- An unknown pin fails instead of quietly falling back to the newest version

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(skills): reject a blank version pin instead of floating to latest

A blank `version` passed to `download_skill` read as "unpinned" and quietly
resolved the latest version, which is not what a caller supplying one asked
for — and it disagreed with `parse_skill_ref`, which already rejects empty
pins. Not reachable through `resolve_registry_ref` or the Agent Repository
auto-pinning, both of which only ever pass a non-empty version.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(skills): carry the caller's context into the worker thread

When resolve_plus_response bridges an async client from inside a running loop
it runs the coroutine on a worker thread, which starts with empty ContextVars.
A client reading runtime state there — the platform integration token, flow
context — would see defaults rather than the caller's values, which is hard to
diagnose from the resulting auth or routing failure.

Copy the context across, matching how the parallel-summarization bridge in this
module already does it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
João Moura
2026-07-26 13:27:29 -03:00
committed by GitHub
parent bd2cb0f23e
commit cc0759854d
11 changed files with 935 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
docs/edge/en/concepts/skills.mdx
View File

@@ -221,6 +221,34 @@ agent = Agent(
)
```
### Pin a Version
An unpinned reference resolves to the newest published version, so publishing a
new version changes every agent that references it. Append `@<version>` to pin
one instead:
```python
agent = Agent(
role="Senior Code Reviewer",
goal="Review pull requests for quality and security issues",
backstory="Staff engineer with expertise in secure coding practices.",
skills=["@acme/code-review@1.2.0"], # pinned; a leading "v" also works
)
```
A pinned reference re-downloads unless the copy it finds is that exact version —
a pin asks for a specific version rather than hinting at one. A cached skill is
matched on the version recorded when it was installed, so it needs nothing in
its frontmatter; a project-local copy under `skills/` has no such record, so it
is matched on `metadata.version` in its `SKILL.md` frontmatter. Pinning an
unpublished version fails rather than falling back to the latest.
<Note>
Agents from the **Agent Repository** are pinned automatically: the repository
records a version alongside each skill it assigns, and the runtime applies those
pins when it loads the agent.
</Note>
### List
```shell Terminal

4
lib/crewai/src/crewai/experimental/skills/__init__.py
View File

@@ -12,8 +12,10 @@ import sys
from crewai.skills import cache, events, registry
from crewai.skills.cache import SkillCacheManager
from crewai.skills.registry import (
SkillRef,
is_registry_ref,
parse_registry_ref,
parse_skill_ref,
resolve_registry_ref,
)
@@ -26,10 +28,12 @@ sys.modules[__name__ + ".registry"] = registry
__all__ = [
"SkillCacheManager",
"SkillRef",
"cache",
"events",
"is_registry_ref",
"parse_registry_ref",
"parse_skill_ref",
"registry",
"resolve_registry_ref",
]

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

@@ -15,8 +15,10 @@ from crewai.skills.loader import (
from crewai.skills.models import Skill, SkillFrontmatter
from crewai.skills.parser import SkillParseError
from crewai.skills.registry import (
SkillRef,
is_registry_ref,
parse_registry_ref,
parse_skill_ref,
resolve_registry_ref,
)
@@ -26,11 +28,13 @@ __all__ = [
"SkillCacheManager",
"SkillFrontmatter",
"SkillParseError",
"SkillRef",
"activate_skill",
"discover_skills",
"is_registry_ref",
"load_skill",
"load_skills",
"parse_registry_ref",
"parse_skill_ref",
"resolve_registry_ref",
]

44
lib/crewai/src/crewai/skills/cache.py
View File

@@ -15,6 +15,8 @@ import tarfile
from typing import TypedDict
import zipfile
from crewai.skills.validation import versions_match
_logger = logging.getLogger(__name__)
@@ -39,13 +41,45 @@ class SkillCacheManager:
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."""
def get_cached_path(
self, org: str, name: str, version: str | None = None
) -> Path | None:
"""Return the cached skill directory path if usable, else None.
Args:
org: Organisation slug.
name: Skill name.
version: When given, the cached entry must record this version.
The cache holds one version per skill, so a pinned lookup for a
different version reports a miss and the caller re-downloads
rather than loading the wrong version.
Returns:
The cached skill directory, or None on a miss.
"""
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
if not (skill_dir.is_dir() and meta_file.exists()):
return None
if version is not None and not self._records_version(meta_file, version):
return None
return skill_dir
def _records_version(self, meta_file: Path, version: str) -> bool:
"""Return True when the cache metadata records *version*."""
try:
meta = json.loads(meta_file.read_text(encoding="utf-8"))
except (OSError, ValueError):
# ValueError covers both JSONDecodeError and the UnicodeDecodeError
# a non-UTF-8 file raises, so a corrupted entry reads as a miss.
_logger.debug("Unreadable cache entry: %s", meta_file, exc_info=True)
return False
if not isinstance(meta, dict):
_logger.debug("Malformed cache entry: %s", meta_file)
return False
# versions_match() treats a non-string version as no match, so a
# corrupted entry reads as a miss rather than raising.
return versions_match(version, meta.get("version"))
def store(
self, org: str, name: str, version: str | None, archive_bytes: bytes

260
lib/crewai/src/crewai/skills/registry.py
View File

@@ -1,35 +1,59 @@
"""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/.
Handles @org/skill-name references (optionally pinned to a version with
@org/skill-name@version), local-first resolution, and downloads via the CrewAI
AMP API with a global cache at ~/.crewai/skills/.
"""
from __future__ import annotations
import inspect
import logging
import os
from pathlib import Path
from typing import Any
from typing import Any, NamedTuple
from crewai.skills.cache import SkillCacheManager
from crewai.skills.validation import versions_match
_logger = logging.getLogger(__name__)
class SkillRef(NamedTuple):
"""A parsed registry reference.
Attributes:
org: Organisation slug or uuid.
name: Skill name.
version: Pinned version, or None to resolve the latest.
"""
org: str
name: str
version: str | None = None
def __str__(self) -> str:
ref = f"@{self.org}/{self.name}"
return f"{ref}@{self.version}" if self.version else 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).
def parse_skill_ref(ref: str) -> SkillRef:
"""Parse a registry reference into its organisation, name and version.
Accepts '@org/skill-name' and the version-pinned '@org/skill-name@1.2.0'
(a leading 'v' on the version is allowed: '@org/skill-name@v1.2.0').
Args:
ref: A registry reference, e.g. '@acme/my-skill'.
ref: A registry reference.
Returns:
A (org, name) tuple.
The parsed :class:`SkillRef`.
Raises:
ValueError: If the reference format is invalid.
@@ -41,7 +65,10 @@ def parse_registry_ref(ref: str) -> tuple[str, str]:
raise ValueError(
f"Registry reference must be in '@org/name' format, got: {ref!r}"
)
org, name = without_at.split("/", 1)
org, remainder = without_at.split("/", 1)
# Skill names cannot contain '@', so the first one starts the version pin.
name, separator, version = remainder.partition("@")
version = version.strip()
if (
not org
or not name
@@ -54,7 +81,34 @@ def parse_registry_ref(ref: str) -> tuple[str, str]:
f"Registry reference org and name must be single, non-empty path "
f"segments (no '..' or leading dots), got: {ref!r}"
)
return org, name
if separator and not version:
raise ValueError(
f"Registry reference version must be non-empty when pinned, got: {ref!r}"
)
if "@" in version:
raise ValueError(
f"Registry reference must carry a single version pin, got: {ref!r}"
)
return SkillRef(org=org, name=name, version=version or None)
def parse_registry_ref(ref: str) -> tuple[str, str]:
"""Parse '@org/skill-name' into (org, name), ignoring any version pin.
Retained for backwards compatibility. Use :func:`parse_skill_ref` to read
the pinned version too.
Args:
ref: A registry reference, e.g. '@acme/my-skill'.
Returns:
An (org, name) tuple.
Raises:
ValueError: If the reference format is invalid.
"""
parsed = parse_skill_ref(ref)
return parsed.org, parsed.name
def resolve_registry_ref(
@@ -68,44 +122,152 @@ def resolve_registry_ref(
2. ~/.crewai/skills/{org}/{name}/ (global cache)
3. Download from registry
A version-pinned reference only accepts a local or cached copy that reports
the pinned version, and downloads otherwise — a pin is a request for a
specific version, not a hint.
Args:
ref: A registry reference, e.g. '@acme/my-skill'.
ref: A registry reference, e.g. '@acme/my-skill' or '@acme/my-skill@1.2.0'.
source: Optional source object passed through to skill loaders (for events).
Returns:
A Skill loaded at INSTRUCTIONS disclosure level.
"""
from crewai.skills.loader import activate_skill
from crewai.skills.parser import load_skill_metadata
org, name = parse_registry_ref(ref)
org, name, version = parse_skill_ref(ref)
local_path = Path.cwd() / "skills" / name
if local_path.is_dir() and (local_path / "SKILL.md").exists():
try:
skill = load_skill_metadata(local_path)
# A project-local copy has no installation metadata, so its frontmatter
# is the only thing a pin can be checked against.
skill = _load_matching_skill(local_path, version)
if skill is not None:
return activate_skill(skill, source=source)
except Exception:
_logger.debug("Failed to load local skill at %s", local_path, exc_info=True)
cache = SkillCacheManager()
cached_path = cache.get_cached_path(org, name)
cached_path = cache.get_cached_path(org, name, version=version)
if cached_path is not None and (cached_path / "SKILL.md").exists():
try:
skill = load_skill_metadata(cached_path)
# get_cached_path already matched the pin against the version the cache
# recorded when it stored the archive, which is authoritative. Checking
# the frontmatter as well would miss on every skill that doesn't declare
# metadata.version and re-download it on each resolution.
skill = _load_skill(cached_path)
if skill is not None:
return activate_skill(skill, source=source)
except Exception:
_logger.debug(
"Failed to load cached skill at %s", cached_path, exc_info=True
)
return download_skill(org, name, source=source)
return download_skill(org, name, source=source, version=version)
def _load_skill(path: Path) -> Skill | None: # type: ignore[name-defined] # noqa: F821
"""Load the skill at *path*, or None when it can't be read."""
from crewai.skills.parser import load_skill_metadata
try:
return load_skill_metadata(path)
except Exception:
_logger.debug("Failed to load skill at %s", path, exc_info=True)
return None
def _load_matching_skill(
path: Path,
version: str | None,
) -> Skill | None: # type: ignore[name-defined] # noqa: F821
"""Load the skill at *path*, or None when it can't satisfy *version*.
Skills record their version in SKILL.md frontmatter under
``metadata.version``; a skill that declares no version can't satisfy a pin.
"""
skill = _load_skill(path)
if skill is None or version is None:
return skill
declared = (skill.frontmatter.metadata or {}).get("version")
if versions_match(version, declared):
return skill
_logger.debug(
"Skill at %s declares version %r, which does not satisfy the pinned %r",
path,
declared,
version,
)
return None
def _accepts_version(get_skill: Any) -> bool:
"""Return True when *get_skill* takes a ``version`` argument.
Assumes it does when the callable can't be introspected, so an unusual but
working client isn't pushed onto the fallback path.
"""
try:
parameters = inspect.signature(get_skill).parameters
except (TypeError, ValueError):
return True
return "version" in parameters or any(
parameter.kind is inspect.Parameter.VAR_KEYWORD
for parameter in parameters.values()
)
def _plus_client(ref: str, pinned: bool) -> Any:
"""Return the CrewAI AMP client to fetch skills with.
Hosted runtimes install a client authenticated for the environment they run
in, and have no user credential to fall back on — so skills resolve through
the same client the Agent Repository uses rather than building a ``PlusAPI``
of their own.
Args:
ref: The reference being resolved, for logging.
pinned: Whether this lookup carries a version pin, which the installed
client has to be able to forward.
"""
from crewai.utilities.agent_utils import resolve_plus_client
def build_default_client() -> Any:
# Deliberately wider than the Agent Repository's default, which reads the
# CLI login only: skills kept their existing precedence so this fix
# doesn't change which credential either lookup picks.
from crewai_core.plus_api import PlusAPI
from crewai.auth.token import get_auth_token
from crewai.context import get_platform_integration_token
return PlusAPI(
api_key=os.getenv("CREWAI_USER_PAT")
or get_platform_integration_token()
or get_auth_token(),
organization_id=os.getenv("CREWAI_ORGANIZATION_UUID"),
)
client = resolve_plus_client(build_default_client)
get_skill = getattr(client, "get_skill", None)
# A runtime predating the Skills Repository installs a client that can't
# fetch skills at all; one predating pins has a get_skill that would reject
# the version argument. Repository agents pin automatically, so both cases
# fall back to a client that can serve the request rather than raising.
if not callable(get_skill) or (pinned and not _accepts_version(get_skill)):
_logger.warning(
"The configured CrewAI AMP client cannot fetch %s, falling back to "
"credentials from the environment.",
"pinned skills" if pinned else "skills",
)
return build_default_client()
return client
def download_skill(
org: str,
name: str,
source: Any = None,
*,
version: str | None = None,
) -> Skill: # type: ignore[name-defined] # noqa: F821
"""Download a skill from the registry and store it in the cache.
@@ -113,14 +275,28 @@ def download_skill(
org: Organisation slug.
name: Skill name.
source: Optional source for event emission.
version: Optional pinned version; the latest is fetched when omitted.
Returns:
The downloaded Skill at INSTRUCTIONS level.
Raises:
ValueError: If *version* is given but blank.
"""
from crewai.skills.loader import activate_skill
from crewai.skills.parser import load_skill_metadata
from crewai.utilities.agent_utils import resolve_plus_response
ref = f"@{org}/{name}"
if version is not None:
# A blank pin would otherwise read as "unpinned" and quietly float to
# the latest version, which is not what a caller passing one asked for.
version = version.strip()
if not version:
raise ValueError(
"A pinned skill version must be non-empty; omit it to fetch the latest."
)
ref = str(SkillRef(org=org, name=name, version=version))
try:
from crewai.events.event_bus import crewai_event_bus
@@ -142,18 +318,12 @@ def download_skill(
)
try:
from crewai_core.plus_api import PlusAPI
from crewai.auth.token import get_auth_token
from crewai.context import get_platform_integration_token
api = PlusAPI(
api_key=os.getenv("CREWAI_USER_PAT")
or get_platform_integration_token()
or get_auth_token(),
organization_id=os.getenv("CREWAI_ORGANIZATION_UUID"),
)
response = api.get_skill(org, name)
api = _plus_client(ref, pinned=bool(version))
# Only pass the version when pinned, so clients predating version
# support keep working for unpinned refs. The pin goes over verbatim and
# the registry decides how to match it, including a leading "v".
pin = {"version": version} if version else {}
response = resolve_plus_response(api.get_skill(org, name, **pin))
response.raise_for_status()
data = response.json()
except Exception as exc:
@@ -165,7 +335,17 @@ def download_skill(
import httpx
version = data.get("latest_version") or data.get("version")
served_version = data.get("version") or data.get("latest_version")
if version and not versions_match(version, served_version):
# Registries that predate pinning ignore the parameter and serve their
# newest version. Caching that under the pinned version would poison
# every later lookup for the pin, so refuse it instead.
raise RuntimeError(
f"Failed to download skill {ref!r} from registry: it served version "
f"{served_version!r} rather than the pinned version {version!r}."
)
resolved_version = version or served_version
download_url = data.get("download_url")
if download_url:
@@ -179,14 +359,14 @@ def download_skill(
archive_bytes = base64.b64decode(encoded)
cache = SkillCacheManager()
skill_dir = cache.store(org, name, version, archive_bytes)
skill_dir = cache.store(org, name, resolved_version, archive_bytes)
if _has_events:
crewai_event_bus.emit(
source,
event=SkillDownloadCompletedEvent(
registry_ref=ref,
version=version,
version=resolved_version,
cache_path=skill_dir,
),
)

27
lib/crewai/src/crewai/skills/validation.py
View File

@@ -7,7 +7,7 @@ from __future__ import annotations
from pathlib import Path
import re
from typing import Final
from typing import Any, Final
MAX_SKILL_NAME_LENGTH: Final[int] = 64
@@ -15,6 +15,31 @@ MIN_SKILL_NAME_LENGTH: Final[int] = 1
SKILL_NAME_PATTERN: Final[re.Pattern[str]] = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
def versions_match(left: Any, right: Any) -> bool:
"""Return True when two skill versions refer to the same version.
Versions arrive as ``1.2.0`` or ``v1.2.0`` depending on whether they came
from a registry ref, SKILL.md frontmatter or a cache entry, so comparison
ignores case and an optional leading ``v``.
Anything that isn't a version string — None, or a corrupted cache entry
holding a number — never matches: a pinned reference should re-resolve
rather than accept a version it can't confirm.
"""
normalized_left = _normalize_version(left)
normalized_right = _normalize_version(right)
if normalized_left is None or normalized_right is None:
return False
return normalized_left == normalized_right
def _normalize_version(value: Any) -> str | None:
"""Return a comparable form of a version, or None when it isn't one."""
if not isinstance(value, str):
return None
return value.strip().lower().removeprefix("v") or None
def validate_directory_name(skill_dir: Path, skill_name: str) -> None:
"""Validate that a directory name matches the skill name.

136
lib/crewai/src/crewai/utilities/agent_utils.py
View File

@@ -55,6 +55,71 @@ if TYPE_CHECKING:
_create_plus_client_hook: Callable[[], Any] | None = None
def resolve_plus_client(default: Callable[[], Any]) -> Any:
"""Return the CrewAI AMP client to use, or build ``default``.
Hosted runtimes install a client of their own through
``_create_plus_client_hook``, authenticated for the environment they run in.
Every registry lookup resolves through here so they all share that client.
Args:
default: Builds the client to use when no hook is installed. A callable
rather than a value so credential lookups that raise when absent are
only evaluated when they're actually needed.
Returns:
A CrewAI AMP client.
"""
if callable(_create_plus_client_hook):
return _create_plus_client_hook()
return default()
def resolve_plus_response(response: Any) -> Any:
"""Return ``response``, awaiting it first when the client is async.
Args:
response: A response, or an awaitable resolving to one.
Returns:
The resolved response.
Raises:
TypeError: If the awaitable is already bound to a loop (a Task or
Future), which can't be resolved synchronously.
"""
if not inspect.isawaitable(response):
return response
if isinstance(response, asyncio.Future):
raise TypeError(
"CrewAI AMP clients must return a coroutine, not a Task or Future "
"already bound to a running loop."
)
# asyncio.run() takes a coroutine specifically, while isawaitable() also
# covers custom __await__ objects, so wrap rather than passing it through.
async def await_response() -> Any:
return await response
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = None
# asyncio.run() refuses to nest inside a running loop, so hand the coroutine
# to a worker thread with a loop of its own — carrying a copy of the caller's
# context, since a fresh thread would otherwise start with empty ContextVars
# and a client reading runtime state (the platform token, flow context) would
# see defaults.
if loop and loop.is_running():
ctx = contextvars.copy_context()
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
return pool.submit(ctx.run, asyncio.run, await_response()).result()
return asyncio.run(await_response())
class SummaryContent(TypedDict):
"""Structure for summary content entries.
@@ -1127,27 +1192,15 @@ def load_agent_from_repository(from_repository: str) -> dict[str, Any]:
if from_repository:
import importlib
if callable(_create_plus_client_hook):
client = _create_plus_client_hook()
else:
def build_default_client() -> Any:
from crewai.auth.token import get_auth_token
from crewai.plus_api import PlusAPI
client = PlusAPI(api_key=get_auth_token())
_print_current_organization()
response = client.get_agent(from_repository)
if inspect.isawaitable(response):
coro = response
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = None
return PlusAPI(api_key=get_auth_token())
if loop and loop.is_running():
with concurrent.futures.ThreadPoolExecutor() as pool:
response = pool.submit(asyncio.run, coro).result() # type: ignore[arg-type]
else:
response = asyncio.run(coro) # type: ignore[arg-type]
client = resolve_plus_client(build_default_client)
_print_current_organization()
response = resolve_plus_response(client.get_agent(from_repository))
if response.status_code == 404:
raise AgentRepositoryError(
f"Agent {from_repository} does not exist, make sure the name is correct or the agent is available on your organization."
@@ -1161,10 +1214,18 @@ def load_agent_from_repository(from_repository: str) -> dict[str, Any]:
)
agent = response.json()
skill_versions = agent.get("skill_versions") or []
for key, value in agent.items():
if value is None:
continue
if key == "tools":
if key == "skill_versions":
# Folded into the skill refs below, and not an Agent field.
continue
if key == "skills":
if not value:
continue
attributes[key] = _pin_skill_refs(value, skill_versions)
elif key == "tools":
attributes[key] = []
for tool in value:
try:
@@ -1182,13 +1243,48 @@ def load_agent_from_repository(from_repository: str) -> dict[str, Any]:
raise AgentRepositoryError(
f"Tool {tool['name']} could not be loaded: {e}"
) from e
elif key == "skills" and value == []:
continue
else:
attributes[key] = value
return attributes
def _pin_skill_refs(refs: list[Any], skill_versions: list[dict[str, Any]]) -> list[Any]:
"""Pin repository skill refs to the versions the repository recorded.
An agent's ``skills`` are plain ``@org/name`` refs while ``skill_versions``
carries the version pinned against each one. Without folding the two
together the runtime resolves whatever version is newest, so publishing a
new version of a skill would silently change every agent using it.
Args:
refs: Skill references as returned by the repository.
skill_versions: Repository version records, each with a ``registry_ref``
and ``version``.
Returns:
The refs, version-pinned where the repository recorded one.
"""
pinned_versions = {
entry["registry_ref"]: entry["version"]
for entry in skill_versions
if isinstance(entry, dict)
and entry.get("registry_ref")
and entry.get("version")
}
if not pinned_versions:
return refs
pinned_refs: list[Any] = []
for ref in refs:
version = pinned_versions.get(ref) if isinstance(ref, str) else None
# Leave anything already pinned (a second '@') exactly as it is.
already_pinned = isinstance(ref, str) and "@" in ref[1:]
pinned_refs.append(
f"{ref}@{version}" if version and not already_pinned else ref
)
return pinned_refs
DELEGATION_TOOL_NAMES: Final[frozenset[str]] = frozenset(
[
sanitize_tool_name("Delegate work to coworker"),

36
lib/crewai/tests/agents/test_agent.py
View File

@@ -2375,6 +2375,42 @@ def test_agent_from_repository_ignores_empty_skills(
assert agent.skills is None
@patch("crewai.plus_api.PlusAPI.get_agent")
def test_agent_from_repository_pins_skills_to_recorded_versions(
mock_get_agent, mock_get_auth_token
):
"""The repository records a version per skill; without the pin the runtime
resolves whatever is newest, so publishing a skill would silently change
every agent using it."""
from crewai.utilities.agent_utils import load_agent_from_repository
mock_get_response = MagicMock()
mock_get_response.status_code = 200
mock_get_response.json.return_value = {
"role": "test role",
"skills": [
"@acme/crewai-brand",
"@acme/already-pinned@3.0.0",
"@acme/unrecorded",
],
"skill_versions": [
{"registry_ref": "@acme/crewai-brand", "version": "2.1.0"},
{"registry_ref": "@acme/already-pinned", "version": "1.0.0"},
],
}
mock_get_agent.return_value = mock_get_response
attributes = load_agent_from_repository("test_agent")
assert attributes["skills"] == [
"@acme/crewai-brand@2.1.0",
"@acme/already-pinned@3.0.0", # keeps the pin it already carried
"@acme/unrecorded", # no recorded version to apply
]
# Not an Agent field — it only exists to carry the pins.
assert "skill_versions" not in attributes
@patch("crewai.plus_api.PlusAPI.get_agent")
def test_agent_from_repository_with_invalid_tools(mock_get_agent, mock_get_auth_token):
mock_get_response = MagicMock()

61
lib/crewai/tests/skills/test_cache.py
View File

@@ -62,6 +62,67 @@ class TestSkillCacheManager:
retrieved = cache.get_cached_path("acme", "my-skill")
assert retrieved == dest
def test_get_cached_path_matches_a_requested_version(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 cache.get_cached_path("acme", "my-skill", version="1.0.0") == dest
# A leading "v" on either side describes the same version.
assert cache.get_cached_path("acme", "my-skill", version="v1.0.0") == dest
def test_get_cached_path_misses_a_different_version(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
archive = _make_tar_gz({"SKILL.md": "---\nname: my-skill\n---\nHello"})
cache.store("acme", "my-skill", "1.0.0", archive)
assert cache.get_cached_path("acme", "my-skill", version="2.0.0") is None
def test_get_cached_path_misses_when_the_recorded_version_is_not_a_string(
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)
meta_file = dest / ".crewai_meta.json"
meta = json.loads(meta_file.read_text(encoding="utf-8"))
meta["version"] = 1.0
meta_file.write_text(json.dumps(meta), encoding="utf-8")
assert cache.get_cached_path("acme", "my-skill", version="1.0.0") is None
def test_get_cached_path_misses_when_metadata_is_not_valid_utf8(
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)
(dest / ".crewai_meta.json").write_bytes(b"\x80\x81")
assert cache.get_cached_path("acme", "my-skill", version="1.0.0") is None
def test_get_cached_path_misses_when_metadata_is_not_an_object(
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)
(dest / ".crewai_meta.json").write_text("[]", encoding="utf-8")
assert cache.get_cached_path("acme", "my-skill", version="1.0.0") is None
def test_get_cached_path_misses_when_the_cached_version_is_unknown(
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", None, archive)
# Unversioned entries still satisfy unpinned lookups.
assert cache.get_cached_path("acme", "my-skill") == dest
# ...but can't confirm a pin, so a pinned lookup re-resolves.
assert cache.get_cached_path("acme", "my-skill", version="1.0.0") is None
def test_store_writes_metadata(self, tmp_path: Path) -> None:
cache = SkillCacheManager(cache_root=tmp_path)
archive = _make_tar_gz({"SKILL.md": "content"})

339
lib/crewai/tests/skills/test_registry.py
View File

@@ -1,6 +1,8 @@
from __future__ import annotations
import base64
from collections.abc import Iterator
from contextlib import contextmanager
from io import BytesIO
from pathlib import Path
from unittest.mock import MagicMock, patch
@@ -9,32 +11,95 @@ from zipfile import ZipFile
from crewai.context import platform_context
from crewai.skills.cache import SkillCacheManager
from crewai.skills.registry import (
SkillRef,
download_skill,
is_registry_ref,
parse_registry_ref,
parse_skill_ref,
resolve_registry_ref,
)
import pytest
def _skill_archive(name: str) -> bytes:
def _skill_archive(name: str, version: str | None = None) -> bytes:
metadata = f"metadata:\n version: {version}\n" if version else ""
archive = BytesIO()
with ZipFile(archive, "w") as zip_file:
zip_file.writestr(
"SKILL.md",
f"---\nname: {name}\ndescription: Test skill.\n---\n\nInstructions.",
f"---\nname: {name}\ndescription: Test skill.\n{metadata}---\n\nInstructions.",
)
return archive.getvalue()
def _mock_skill_response(name: str) -> MagicMock:
def _skill_response(name: str, version: str | None = None) -> MagicMock:
response = MagicMock()
response.json.return_value = {
payload: dict[str, object] = {
"latest_version": "1.0.0",
"file": base64.b64encode(_skill_archive(name)).decode(),
"file": base64.b64encode(_skill_archive(name, version)).decode(),
}
if version is not None:
payload["version"] = version
response.json.return_value = payload
return response
# Retained under its original name for the pre-existing tests below.
_mock_skill_response = _skill_response
def _stub_api(name: str, version: str | None = None) -> MagicMock:
"""A CrewAI AMP client stub serving one skill."""
api = MagicMock()
api.get_skill.return_value = _skill_response(name, version)
return api
def _cache(tmp_path: Path) -> SkillCacheManager:
return SkillCacheManager(cache_root=tmp_path / "cache")
def _write_local_skill(tmp_path: Path, name: str, version: str | None = None) -> Path:
"""Write a project-local ./skills/<name>/SKILL.md under *tmp_path*."""
skill_dir = tmp_path / "skills" / name
skill_dir.mkdir(parents=True)
metadata = f"metadata:\n version: {version}\n" if version else ""
(skill_dir / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: Test skill.\n{metadata}---\n\nInstructions."
)
return skill_dir
def _install_client(monkeypatch: pytest.MonkeyPatch, client: object) -> None:
"""Install *client* the way a hosted runtime does."""
from crewai.utilities import agent_utils
monkeypatch.setattr(agent_utils, "_create_plus_client_hook", lambda: client)
@contextmanager
def _resolving(
tmp_path: Path, api: MagicMock, cache: SkillCacheManager | None = None
) -> Iterator[SkillCacheManager]:
"""Resolve refs against a temp cwd and cache, with *api* as the registry."""
cache = cache or _cache(tmp_path)
with (
patch.object(Path, "cwd", return_value=tmp_path),
patch("crewai.auth.token.get_auth_token", return_value="saved-login"),
patch("crewai.skills.registry.SkillCacheManager", return_value=cache),
patch("crewai_core.plus_api.PlusAPI", return_value=api),
):
yield cache
@pytest.fixture(autouse=True)
def _no_installed_client(monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep a client installed by one test from leaking into the next."""
from crewai.utilities import agent_utils
monkeypatch.setattr(agent_utils, "_create_plus_client_hook", None, raising=False)
class TestIsRegistryRef:
def test_at_prefixed(self) -> None:
assert is_registry_ref("@acme/my-skill") is True
@@ -74,22 +139,52 @@ class TestParseRegistryRef:
with pytest.raises(ValueError, match="non-empty"):
parse_registry_ref("@acme/")
def test_drops_the_version_pin(self) -> None:
assert parse_registry_ref("@acme/my-skill@1.2.0") == ("acme", "my-skill")
class TestParseSkillRef:
def test_without_version(self) -> None:
assert parse_skill_ref("@acme/my-skill") == SkillRef("acme", "my-skill", None)
def test_with_version(self) -> None:
assert parse_skill_ref("@acme/my-skill@1.2.0") == SkillRef(
"acme", "my-skill", "1.2.0"
)
def test_with_v_prefixed_version(self) -> None:
assert parse_skill_ref("@acme/my-skill@v0.1").version == "v0.1"
def test_with_uuid_org(self) -> None:
assert parse_skill_ref(
"@548e8ab5-f806-4c44-9dea-087ea05880f1/crewai-brand@2.0"
) == SkillRef("548e8ab5-f806-4c44-9dea-087ea05880f1", "crewai-brand", "2.0")
def test_empty_version(self) -> None:
with pytest.raises(ValueError, match="version must be non-empty"):
parse_skill_ref("@acme/my-skill@")
def test_whitespace_only_version(self) -> None:
with pytest.raises(ValueError, match="version must be non-empty"):
parse_skill_ref("@acme/my-skill@ ")
def test_rejects_more_than_one_version_pin(self) -> None:
with pytest.raises(ValueError, match="single version pin"):
parse_skill_ref("@acme/my-skill@1.2.0@extra")
def test_strips_surrounding_whitespace_from_the_version(self) -> None:
assert parse_skill_ref("@acme/my-skill@ 1.2.0 ").version == "1.2.0"
def test_round_trips_through_str(self) -> None:
assert str(parse_skill_ref("@acme/my-skill@1.2.0")) == "@acme/my-skill@1.2.0"
assert str(parse_skill_ref("@acme/my-skill")) == "@acme/my-skill"
class TestResolveRegistryRef:
def _make_skill_dir(self, base: Path, name: str) -> Path:
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_prefers_project_local_skill_over_cached_skill(
self, tmp_path: Path
) -> None:
skills_dir = tmp_path / "skills"
skills_dir.mkdir()
self._make_skill_dir(skills_dir, "my-skill")
_write_local_skill(tmp_path, "my-skill")
cache_dir = tmp_path / "cache" / "acme" / "my-skill"
cache_dir.mkdir(parents=True)
@@ -156,6 +251,218 @@ class TestResolveRegistryRef:
assert skill.name == "cached-skill"
class TestResolveVersionPinnedRef:
def test_requests_the_pinned_version_from_the_registry(self, tmp_path: Path) -> None:
api = _stub_api("pinned-skill", "1.0.0")
with _resolving(tmp_path, api):
skill = resolve_registry_ref("@acme/pinned-skill@1.0.0")
assert skill.name == "pinned-skill"
api.get_skill.assert_called_once_with("acme", "pinned-skill", version="1.0.0")
def test_unpinned_ref_omits_the_version_argument(self, tmp_path: Path) -> None:
api = _stub_api("floating-skill")
with _resolving(tmp_path, api):
resolve_registry_ref("@acme/floating-skill")
api.get_skill.assert_called_once_with("acme", "floating-skill")
def test_resolves_a_pinned_skill_from_the_cache(self, tmp_path: Path) -> None:
"""Skills need not declare metadata.version; the cache records what it
stored, so a second resolution must not download again."""
api = _stub_api("cached-skill")
with _resolving(tmp_path, api):
resolve_registry_ref("@acme/cached-skill@1.0.0")
skill = resolve_registry_ref("@acme/cached-skill@1.0.0")
assert skill.name == "cached-skill"
assert api.get_skill.call_count == 1
def test_refuses_a_version_the_registry_served_instead_of_the_pin(
self, tmp_path: Path
) -> None:
"""A registry predating pinning ignores the parameter and serves its
newest version; caching that under the pin would poison every later
lookup for it."""
cache = _cache(tmp_path)
api = _stub_api("drifting-skill", "2.0.0")
with _resolving(tmp_path, api, cache=cache), pytest.raises(
RuntimeError, match="served version '2.0.0' rather than the pinned"
):
resolve_registry_ref("@acme/drifting-skill@1.0.0")
assert cache.get_cached_path("acme", "drifting-skill") is None
def test_redownloads_when_the_cached_version_is_not_the_pinned_one(
self, tmp_path: Path
) -> None:
cache = SkillCacheManager(cache_root=tmp_path / "cache")
cache.store("acme", "drifting-skill", "1.0.0", _skill_archive("drifting-skill"))
api = _stub_api("drifting-skill", "2.0.0")
with _resolving(tmp_path, api, cache=cache):
resolve_registry_ref("@acme/drifting-skill@2.0.0")
api.get_skill.assert_called_once_with("acme", "drifting-skill", version="2.0.0")
def test_uses_a_project_local_skill_that_declares_the_pinned_version(
self, tmp_path: Path
) -> None:
_write_local_skill(tmp_path, "local-skill", "1.0.0")
api = _stub_api("local-skill")
with _resolving(tmp_path, api):
skill = resolve_registry_ref("@acme/local-skill@v1.0.0")
assert skill.name == "local-skill"
api.get_skill.assert_not_called()
def test_downloads_when_a_project_local_skill_declares_another_version(
self, tmp_path: Path
) -> None:
_write_local_skill(tmp_path, "stale-skill", "1.0.0")
api = _stub_api("stale-skill", "2.0.0")
with _resolving(tmp_path, api):
resolve_registry_ref("@acme/stale-skill@2.0.0")
api.get_skill.assert_called_once_with("acme", "stale-skill", version="2.0.0")
class TestDownloadSkillClient:
"""A hosted runtime has no user credential, so its own client must be used."""
def test_uses_the_client_the_runtime_installed(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
installed = _stub_api("my-skill")
monkeypatch.delenv("CREWAI_USER_PAT", raising=False)
_install_client(monkeypatch, installed)
with (
patch("crewai.skills.registry.SkillCacheManager", return_value=_cache(tmp_path)),
patch("crewai.auth.token.get_auth_token") as get_auth_token,
patch("crewai_core.plus_api.PlusAPI") as plus_api,
):
download_skill("acme", "my-skill")
installed.get_skill.assert_called_once_with("acme", "my-skill")
# No user credential is read at all when a client is installed.
plus_api.assert_not_called()
get_auth_token.assert_not_called()
def test_awaits_an_async_client(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
response = _skill_response("my-skill")
async def get_skill(org: str, name: str, **kwargs: object) -> MagicMock:
return response
installed = MagicMock()
installed.get_skill = get_skill
_install_client(monkeypatch, installed)
with patch(
"crewai.skills.registry.SkillCacheManager", return_value=_cache(tmp_path)
):
skill = download_skill("acme", "my-skill")
assert skill.name == "my-skill"
@pytest.mark.parametrize(
"get_skill", [None, "not-callable"], ids=["missing", "not-callable"]
)
def test_falls_back_when_the_client_cannot_fetch_skills(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, get_skill: object
) -> None:
api = _stub_api("my-skill")
monkeypatch.setenv("CREWAI_USER_PAT", "user-pat")
installed = MagicMock(spec=[] if get_skill is None else ["get_skill"])
if get_skill is not None:
installed.get_skill = get_skill
_install_client(monkeypatch, installed)
with (
patch("crewai.skills.registry.SkillCacheManager", return_value=_cache(tmp_path)),
patch("crewai_core.plus_api.PlusAPI", return_value=api) as plus_api,
):
skill = download_skill("acme", "my-skill")
assert skill.name == "my-skill"
plus_api.assert_called_once_with(api_key="user-pat", organization_id=None)
def test_falls_back_when_the_client_cannot_forward_a_pin(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Repository agents pin automatically, so a client whose get_skill
predates pinning must not turn a working lookup into a TypeError."""
api = _stub_api("my-skill", "1.0.0")
monkeypatch.setenv("CREWAI_USER_PAT", "user-pat")
class VersionUnawareClient:
def get_skill(self, org: str, name: str) -> MagicMock:
raise AssertionError("must not be called with a pin")
_install_client(monkeypatch, VersionUnawareClient())
with (
patch("crewai.skills.registry.SkillCacheManager", return_value=_cache(tmp_path)),
patch("crewai_core.plus_api.PlusAPI", return_value=api) as plus_api,
):
skill = download_skill("acme", "my-skill", version="1.0.0")
assert skill.name == "my-skill"
plus_api.assert_called_once_with(api_key="user-pat", organization_id=None)
def test_uses_a_version_unaware_client_for_unpinned_refs(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
calls: list[tuple[str, str]] = []
class VersionUnawareClient:
def get_skill(self, org: str, name: str) -> MagicMock:
calls.append((org, name))
return _skill_response("my-skill")
_install_client(monkeypatch, VersionUnawareClient())
with patch(
"crewai.skills.registry.SkillCacheManager", return_value=_cache(tmp_path)
):
download_skill("acme", "my-skill")
assert calls == [("acme", "my-skill")]
@pytest.mark.parametrize("version", ["", " "], ids=["empty", "whitespace"])
def test_rejects_a_blank_pin_rather_than_floating_to_latest(
self, monkeypatch: pytest.MonkeyPatch, version: str
) -> None:
installed = _stub_api("my-skill")
_install_client(monkeypatch, installed)
with pytest.raises(ValueError, match="must be non-empty"):
download_skill("acme", "my-skill", version=version)
installed.get_skill.assert_not_called()
def test_reports_the_pinned_ref_when_the_download_fails(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
installed = MagicMock()
installed.get_skill.side_effect = RuntimeError("401 Unauthorized")
_install_client(monkeypatch, installed)
with pytest.raises(
RuntimeError, match=r"Failed to download skill '@acme/my-skill@1\.0\.0'"
):
download_skill("acme", "my-skill", version="1.0.0")
class TestDownloadSkillAuthentication:
def test_user_pat_takes_precedence_over_platform_token(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch

78
lib/crewai/tests/utilities/test_agent_utils.py
View File

@@ -1256,3 +1256,81 @@ class TestExecuteSingleNativeToolCall:
assert isinstance(result, NativeToolCallResult)
assert result.result_as_answer is False
assert "blocked by hook" in result.result
class TestResolvePlusClient:
def test_builds_the_default_when_no_client_is_installed(self) -> None:
from crewai.utilities.agent_utils import resolve_plus_client
default = MagicMock()
assert resolve_plus_client(lambda: default) is default
def test_prefers_an_installed_client(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A hosted runtime installs a client; the default must not be built,
since looking up a user credential raises when there isn't one."""
from crewai.utilities import agent_utils
installed = MagicMock()
monkeypatch.setattr(agent_utils, "_create_plus_client_hook", lambda: installed)
default = MagicMock(side_effect=AssertionError("must not be called"))
assert agent_utils.resolve_plus_client(default) is installed
default.assert_not_called()
class TestResolvePlusResponse:
def test_passes_through_a_sync_response(self) -> None:
from crewai.utilities.agent_utils import resolve_plus_response
response = MagicMock()
assert resolve_plus_response(response) is response
@pytest.mark.parametrize("inside_loop", [False, True])
def test_awaits_an_async_response(self, inside_loop: bool) -> None:
from crewai.utilities.agent_utils import resolve_plus_response
response = MagicMock()
async def call() -> Any:
return response
if not inside_loop:
assert resolve_plus_response(call()) is response
return
async def main() -> Any:
return resolve_plus_response(call())
assert asyncio.run(main()) is response
def test_carries_context_vars_into_the_worker_thread(self) -> None:
"""Inside a running loop the coroutine runs on another thread; a client
reading runtime state (the platform token, flow context) must still see
the caller's values rather than defaults."""
from crewai.context import get_platform_integration_token, platform_context
from crewai.utilities.agent_utils import resolve_plus_response
async def call() -> Any:
return get_platform_integration_token()
async def main() -> Any:
with platform_context("token-from-caller"):
return resolve_plus_response(call())
assert asyncio.run(main()) == "token-from-caller"
def test_rejects_an_awaitable_bound_to_a_loop(self) -> None:
from crewai.utilities.agent_utils import resolve_plus_response
async def main() -> None:
future: asyncio.Future[Any] = asyncio.get_running_loop().create_future()
future.set_result(MagicMock())
with pytest.raises(TypeError, match="must return a coroutine"):
resolve_plus_response(future)
asyncio.run(main())