diff --git a/docs/edge/en/concepts/skills.mdx b/docs/edge/en/concepts/skills.mdx
index 23da69088..05d8aef2d 100644
--- a/docs/edge/en/concepts/skills.mdx
+++ b/docs/edge/en/concepts/skills.mdx
@@ -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
diff --git a/lib/crewai/src/crewai/experimental/skills/__init__.py b/lib/crewai/src/crewai/experimental/skills/__init__.py
index 3974f8b9a..1d2bd6daa 100644
--- a/lib/crewai/src/crewai/experimental/skills/__init__.py
+++ b/lib/crewai/src/crewai/experimental/skills/__init__.py
@@ -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",
 ]
diff --git a/lib/crewai/src/crewai/skills/__init__.py b/lib/crewai/src/crewai/skills/__init__.py
index 8d72f8af6..77ccd892a 100644
--- a/lib/crewai/src/crewai/skills/__init__.py
+++ b/lib/crewai/src/crewai/skills/__init__.py
@@ -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",
 ]
diff --git a/lib/crewai/src/crewai/skills/cache.py b/lib/crewai/src/crewai/skills/cache.py
index 64d1b7bff..7e35fbc6b 100644
--- a/lib/crewai/src/crewai/skills/cache.py
+++ b/lib/crewai/src/crewai/skills/cache.py
@@ -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
diff --git a/lib/crewai/src/crewai/skills/registry.py b/lib/crewai/src/crewai/skills/registry.py
index d6b9fa98f..a20600bac 100644
--- a/lib/crewai/src/crewai/skills/registry.py
+++ b/lib/crewai/src/crewai/skills/registry.py
@@ -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,
             ),
         )
diff --git a/lib/crewai/src/crewai/skills/validation.py b/lib/crewai/src/crewai/skills/validation.py
index 78acd7b76..e46721485 100644
--- a/lib/crewai/src/crewai/skills/validation.py
+++ b/lib/crewai/src/crewai/skills/validation.py
@@ -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.
 
diff --git a/lib/crewai/src/crewai/utilities/agent_utils.py b/lib/crewai/src/crewai/utilities/agent_utils.py
index 48e0fa7c9..6a1d2fcf5 100644
--- a/lib/crewai/src/crewai/utilities/agent_utils.py
+++ b/lib/crewai/src/crewai/utilities/agent_utils.py
@@ -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"),
diff --git a/lib/crewai/tests/agents/test_agent.py b/lib/crewai/tests/agents/test_agent.py
index 4bcb4d404..98d220bd1 100644
--- a/lib/crewai/tests/agents/test_agent.py
+++ b/lib/crewai/tests/agents/test_agent.py
@@ -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()
diff --git a/lib/crewai/tests/skills/test_cache.py b/lib/crewai/tests/skills/test_cache.py
index 25ffce39b..ba052094a 100644
--- a/lib/crewai/tests/skills/test_cache.py
+++ b/lib/crewai/tests/skills/test_cache.py
@@ -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"})
diff --git a/lib/crewai/tests/skills/test_registry.py b/lib/crewai/tests/skills/test_registry.py
index 25afe50c1..a6b9de976 100644
--- a/lib/crewai/tests/skills/test_registry.py
+++ b/lib/crewai/tests/skills/test_registry.py
@@ -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
diff --git a/lib/crewai/tests/utilities/test_agent_utils.py b/lib/crewai/tests/utilities/test_agent_utils.py
index 9cf4a2d2a..0910fb86e 100644
--- a/lib/crewai/tests/utilities/test_agent_utils.py
+++ b/lib/crewai/tests/utilities/test_agent_utils.py
@@ -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())