Python: Add MCP-based skills discovery (McpSkillsSource) (#6169)

* Add MCP-based skills discovery (McpSkill, McpSkillsSource, McpSkillResource)

Implement Agent Skills discovery over MCP following the SEP-2640 convention:
- McpSkillsSource: reads skill://index.json to discover skills served by an MCP server
- McpSkill: lazily fetches SKILL.md content via resources/read on demand
- McpSkillResource: wraps MCP resource results (text and binary)
- Path traversal protection in get_resource for defense in depth
- Samples for Foundry Toolbox and standalone MCP skills server
- Comprehensive unit tests (514 lines)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Address PR review comments: rename to MCP* convention, fix error handling and samples

- Rename McpSkill/McpSkillResource/McpSkillsSource to MCPSkill/MCPSkillResource/MCPSkillsSource
- Add data-URI prefix stripping for blob resource decoding
- Let non-McpError exceptions propagate from get_resource()
- Fix contradictory test comment
- Use interactive input() in mcp_based_skill sample
- Remove misleading sample output block

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Restore debug logging for McpError in get_resource()

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Use AzureCliCredential in Foundry toolbox skills sample for consistency

Replace DefaultAzureCredential with AzureCliCredential to match the
credential convention used in all other samples.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Use MCPStreamableHTTPTool in MCP skills sample

Replace raw mcp library imports (ClientSession, streamable_http_client)
with the framework's MCPStreamableHTTPTool to keep MCP server connections
consistent regardless of whether skills are enabled.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Branch on McpError.error.code so only not-found errors return empty

Previously _try_read_index() and get_resource() swallowed every McpError
as 'no skills available', making auth failures, server crashes, and
connection drops indistinguishable from a server that simply has no
skills.

Now only two codes are treated as not-found:
- -32002 (MCP-spec Resource not found)
- -32601 (METHOD_NOT_FOUND — server lacks resources/read)

All other McpError codes and non-McpError exceptions propagate with a
warning log, surfacing real failures visibly.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Add tests for non-McpError and non-not-found error propagation in MCP skills

Cover the re-raise branch in MCPSkill.get_resource for plain
ConnectionError/TimeoutError, the generic McpError (code 0) propagation
on get_resource, and TimeoutError propagation in _try_read_index.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Revert "Use MCPStreamableHTTPTool in MCP skills sample"

This reverts commit f31ed0ded9.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Introduce MCP_SKILLS experimental feature for MCP skill classes

Add a separate MCP_SKILLS feature ID to ExperimentalFeature enum and
use it for MCPSkillResource, MCPSkill, and MCPSkillsSource, since their
promotion timeline is partly outside of our control.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

python/packages/core/agent_framework/__init__.py

@@ -168,6 +168,9 @@ from ._skills import (
     InlineSkillResource,
     InlineSkillScript,
     InMemorySkillsSource,
+    MCPSkill,
+    MCPSkillResource,
+    MCPSkillsSource,
     Skill,
     SkillFrontmatter,
     SkillResource,
@@ -444,6 +447,9 @@ __all__ = [
     "MCPStdioTool",
     "MCPStreamableHTTPTool",
     "MCPWebsocketTool",
+    "MCPSkill",
+    "MCPSkillResource",
+    "MCPSkillsSource",
     "MemoryContextProvider",
     "MemoryFileStore",
     "MemoryIndexEntry",

python/packages/core/agent_framework/_feature_stage.py

@@ -58,6 +58,7 @@ class ExperimentalFeature(str, Enum):
     FOUNDRY_PREVIEW_TOOLS = "FOUNDRY_PREVIEW_TOOLS"
     FUNCTIONAL_WORKFLOWS = "FUNCTIONAL_WORKFLOWS"
     HARNESS = "HARNESS"
+    MCP_SKILLS = "MCP_SKILLS"
     PROGRESSIVE_TOOLS = "PROGRESSIVE_TOOLS"
     SKILLS = "SKILLS"
     TO_PROMPT_AGENT = "TO_PROMPT_AGENT"

python/packages/core/agent_framework/_skills.py

@@ -44,6 +44,7 @@ Only use skills from trusted sources.
 from __future__ import annotations
 
 import asyncio
+import base64
 import inspect
 import json
 import logging
@@ -60,6 +61,10 @@ from ._sessions import ContextProvider
 from ._tools import FunctionTool
 
 if TYPE_CHECKING:
+    from mcp.client.session import ClientSession
+    from mcp.types import ReadResourceResult
+    from pydantic import AnyUrl
+
     from ._agents import SupportsAgentRun
     from ._sessions import AgentSession, SessionContext
 
@@ -3285,4 +3290,443 @@ class AggregatingSkillsSource(SkillsSource):
         return result
 
 
+# region MCP Skills
+
+
+def _mcp_any_url(uri: str) -> AnyUrl:
+    """Convert a string URI to a :class:`pydantic.AnyUrl` for MCP client calls."""
+    from pydantic import AnyUrl as _AnyUrl
+
+    return _AnyUrl(uri)
+
+
+def _is_mcp_resource_not_found(ex: Exception) -> bool:
+    """Return ``True`` when *ex* is an :class:`McpError` indicating a missing resource.
+
+    Two codes are treated as "not found":
+
+    * ``-32002`` — the MCP-spec "Resource not found" code returned by a
+      compliant server when the URI does not exist. Not exported as a
+      constant from ``mcp.types`` but defined by the resources subprotocol.
+    * ``METHOD_NOT_FOUND`` (``-32601``) — the server does not implement
+      ``resources/read`` at all, which for the skills source is functionally
+      equivalent to "no skills available."
+
+    All other codes — ``INVALID_PARAMS``, ``INTERNAL_ERROR``, ``PARSE_ERROR``,
+    ``CONNECTION_CLOSED``, auth rejections, and generic handler errors
+    (code ``0``) — are treated as real failures so that a misconfigured
+    token or crashing server is not silently mistaken for "the server has no
+    skills."
+    """
+    from mcp.shared.exceptions import McpError as _McpError
+
+    if not isinstance(ex, _McpError):
+        return False
+    from mcp.types import METHOD_NOT_FOUND as _METHOD_NOT_FOUND
+
+    return ex.error.code in {-32002, _METHOD_NOT_FOUND}
+
+
+def _mcp_join_text(result: ReadResourceResult) -> str:
+    """Join all :class:`TextResourceContents` items in a result into a single string."""
+    from mcp.types import TextResourceContents as _TextResourceContents
+
+    return "\n".join(c.text for c in result.contents if isinstance(c, _TextResourceContents))
+
+
+class _McpSkillIndexEntry:  # noqa: B903
+    """A single entry in the ``skill://index.json`` discovery document.
+
+    All fields are optional to support lenient deserialization; callers
+    validate required fields before use.
+    """
+
+    def __init__(
+        self,
+        *,
+        name: str | None = None,
+        type: str | None = None,
+        description: str | None = None,
+        url: str | None = None,
+        digest: str | None = None,
+    ) -> None:
+        self.name = name
+        self.type = type
+        self.description = description
+        self.url = url
+        self.digest = digest
+
+
+class _McpSkillIndex:
+    """DTO for the ``skill://index.json`` discovery document.
+
+    Represents the Agent Skills Discovery v0.2.0 schema as bound to MCP
+    by SEP-2640.
+    """
+
+    def __init__(
+        self,
+        *,
+        schema: str | None = None,
+        skills: list[_McpSkillIndexEntry] | None = None,
+    ) -> None:
+        self.schema = schema
+        self.skills: list[_McpSkillIndexEntry] = skills if skills is not None else []
+
+
+def _parse_mcp_skill_index(text: str) -> _McpSkillIndex:
+    """Parse a JSON string into a :class:`_McpSkillIndex`.
+
+    Args:
+        text: Raw JSON text from ``skill://index.json``.
+
+    Returns:
+        A populated :class:`_McpSkillIndex` instance.
+
+    Raises:
+        json.JSONDecodeError: If the text is not valid JSON.
+        ValueError: If the top-level value is not a JSON object.
+    """
+    raw: dict[str, Any] = json.loads(text)
+
+    if not isinstance(raw, dict):
+        raise ValueError("skill://index.json must be a JSON object")
+
+    entries: list[_McpSkillIndexEntry] = []
+
+    raw_skills: list[Any] = raw.get("skills") or []
+
+    for item in raw_skills:
+        if isinstance(item, dict):
+            d = cast(dict[str, Any], item)
+
+            entries.append(
+                _McpSkillIndexEntry(
+                    name=d.get("name"),
+                    type=d.get("type"),
+                    description=d.get("description"),
+                    url=d.get("url"),
+                    digest=d.get("digest"),
+                )
+            )
+
+    return _McpSkillIndex(schema=raw.get("$schema"), skills=entries)
+
+
+@experimental(feature_id=ExperimentalFeature.MCP_SKILLS)
+class MCPSkillResource(SkillResource):
+    """A :class:`SkillResource` backed by content fetched from an MCP server.
+
+    The :class:`~mcp.types.ReadResourceResult` is fetched eagerly by
+    :meth:`MCPSkill.get_resource` at construction time; :meth:`read`
+    extracts text or binary content from the result.
+    """
+
+    def __init__(self, *, name: str, result: ReadResourceResult) -> None:
+        """Initialize an MCPSkillResource.
+
+        Args:
+            name: The resource name (e.g. a relative path or identifier).
+            result: The result returned by the MCP server's ``resources/read`` request.
+        """
+        super().__init__(name=name)
+        self._result = result
+
+    async def read(self, **kwargs: Any) -> Any:
+        """Read the resource content.
+
+        Returns:
+            A ``bytes`` object when the resource contains binary content,
+            a ``str`` when it contains text, or ``None`` when the server
+            returned no content blocks.
+        """
+        from mcp.types import BlobResourceContents, TextResourceContents
+
+        for content in self._result.contents:
+            if isinstance(content, BlobResourceContents):
+                blob = content.blob
+                # Strip data-URI prefix if present (some MCP servers send
+                # full data URIs instead of raw base64).
+                if blob.startswith("data:"):
+                    blob = blob.split(",", 1)[-1]
+                return base64.b64decode(blob)
+
+        text = "\n".join(c.text for c in self._result.contents if isinstance(c, TextResourceContents))
+        return text if text else None
+
+
+@experimental(feature_id=ExperimentalFeature.MCP_SKILLS)
+class MCPSkill(Skill):
+    """A :class:`Skill` discovered from an MCP server exposing the Agent Skills convention.
+
+    The skill is constructed from ``skill://index.json`` discovery metadata;
+    :meth:`get_content` fetches the full ``SKILL.md`` content from the MCP
+    server on demand via ``resources/read``.
+
+    Per SEP-2640, resources referenced inside SKILL.md are fetched on demand
+     via the originating MCP server: :meth:`get_resource` resolves a relative
+    resource name against the skill's root URI, issues a ``resources/read``
+     request, and returns an :class:`MCPSkillResource` with pre-fetched content.
+    """
+
+    _SKILL_MD_SUFFIX: Final[str] = "SKILL.md"
+
+    def __init__(
+        self,
+        frontmatter: SkillFrontmatter,
+        skill_md_uri: str,
+        client: ClientSession,
+    ) -> None:
+        """Initialize an MCPSkill.
+
+        Args:
+            frontmatter: The parsed frontmatter metadata for this skill.
+            skill_md_uri: The full MCP resource URI of the ``SKILL.md`` resource
+                (e.g. ``skill://unit-converter/SKILL.md``). The skill's root URI
+                is derived by stripping the trailing ``SKILL.md`` segment.
+            client: The MCP client session used to fetch resources on demand.
+        """
+        self._frontmatter = frontmatter
+        self._skill_md_uri = skill_md_uri
+        self._skill_root_uri = self._compute_skill_root_uri(skill_md_uri)
+        self._client = client
+        self._content: str | None = None
+
+    @property
+    def frontmatter(self) -> SkillFrontmatter:
+        """The L1 discovery metadata for this skill."""
+        return self._frontmatter
+
+    async def get_content(self) -> str:
+        """Get the full SKILL.md content from the MCP server.
+
+        Fetches the content via ``resources/read`` on the first call and
+        caches the result for subsequent calls.
+
+        Returns:
+            The SKILL.md content string.
+
+        Raises:
+            ValueError: If the MCP server returned no text content for the
+                SKILL.md resource.
+        """
+        if self._content is not None:
+            return self._content
+
+        result = await self._client.read_resource(_mcp_any_url(self._skill_md_uri))
+        text = _mcp_join_text(result)
+        if not text:
+            raise ValueError(
+                f"The MCP server returned no text content for SKILL.md resource '{self._skill_md_uri}'."
+            )
+        self._content = text
+        return text
+
+    async def get_resource(self, name: str) -> SkillResource | None:
+        """Get a sibling resource by name from the MCP server.
+
+        Resolves *name* as a relative path against the skill's root URI,
+        issues a ``resources/read`` request to the MCP server, and returns
+        an :class:`MCPSkillResource` with the pre-fetched content.
+
+        Args:
+            name: The resource name (e.g. ``references/checklist.md``).
+
+        Returns:
+            An :class:`MCPSkillResource`, or ``None`` when the name is empty
+            or the resource does not exist on the server.
+        """
+        if not name or not name.strip():
+            return None
+
+        normalized = self._validate_resource_name(name)
+        if normalized is None:
+            return None
+
+        uri = self._skill_root_uri + normalized
+        try:
+            result = await self._client.read_resource(_mcp_any_url(uri))
+        except Exception as ex:
+            if _is_mcp_resource_not_found(ex):
+                logger.debug("MCP resource '%s' not available: %s", uri, ex)
+                return None
+            raise
+
+        return MCPSkillResource(name=name, result=result)
+
+    @staticmethod
+    def _validate_resource_name(name: str) -> str | None:
+        """Validate a resource name and return the normalized form.
+
+        Defense in depth: refuses names that could escape the skill root
+        (absolute paths, embedded URI schemes, parent-traversal segments).
+        The MCP server is the authority on URI resolution, but rejecting
+        obviously unsafe shapes client-side avoids leaking escape attempts
+        upstream.
+
+        Args:
+            name: The raw resource name to validate.
+
+        Returns:
+            The normalized name with backslashes replaced by forward slashes,
+            or ``None`` if the name is unsafe.
+        """
+        normalized = name.replace("\\", "/")
+        if (
+            normalized.startswith("/")
+            or "://" in normalized
+            or any(seg == ".." for seg in normalized.split("/"))
+        ):
+            logger.debug("Rejecting resource name with unsafe path components: %r", name)
+            return None
+        return normalized
+
+    @staticmethod
+    def _compute_skill_root_uri(skill_md_uri: str) -> str:
+        """Strip the trailing ``SKILL.md`` from the URI to produce the skill root.
+
+        If the URI doesn't end with ``SKILL.md``, ensures it ends with a
+        trailing slash.
+        """
+        if skill_md_uri.endswith(MCPSkill._SKILL_MD_SUFFIX):
+            return skill_md_uri[: -len(MCPSkill._SKILL_MD_SUFFIX)]
+        if skill_md_uri.endswith("/"):
+            return skill_md_uri
+        return skill_md_uri + "/"
+
+
+@experimental(feature_id=ExperimentalFeature.MCP_SKILLS)
+class MCPSkillsSource(SkillsSource):
+    """A :class:`SkillsSource` that discovers Agent Skills served over MCP.
+
+    Discovery follows the SEP-2640 recommended approach: the source reads
+    the well-known ``skill://index.json`` resource and constructs one
+    :class:`MCPSkill` per ``skill-md`` entry directly from the entry's
+    ``name``, ``description``, and ``url`` fields.
+
+    The referenced ``SKILL.md`` resource is **not** read during discovery;
+    the host fetches its body on demand via ``resources/read`` when the
+    skill content is needed.
+
+    Only index entries of type ``skill-md`` are supported; entries of any
+    other type are silently skipped.
+
+    If ``skill://index.json`` is absent, unreadable, empty, or fails to
+    parse, this source returns an empty list.
+
+    Examples:
+        .. code-block:: python
+
+            from mcp.client.session import ClientSession
+
+            source = MCPSkillsSource(client=session)
+            skills = await source.get_skills()
+    """
+
+    _INDEX_URI: Final[str] = "skill://index.json"
+    _SKILL_MD_TYPE: Final[str] = "skill-md"
+
+    def __init__(self, client: ClientSession) -> None:
+        """Initialize an MCPSkillsSource.
+
+        Args:
+            client: An MCP client session connected to a server that
+                exposes Agent Skills resources.
+        """
+        self._client = client
+
+    async def get_skills(self) -> list[Skill]:
+        """Discover and return skills from the MCP server.
+
+        Reads ``skill://index.json``, parses it, and creates an
+        :class:`MCPSkill` for each valid ``skill-md`` entry.
+
+        Returns:
+            A list of discovered :class:`MCPSkill` instances.
+        """
+        index = await self._try_read_index()
+        if index is None:
+            return []
+
+        skills: list[Skill] = []
+        for entry in index.skills:
+            result = self._try_create_skill(entry)
+            if result is not None:
+                skills.append(result)
+                logger.info("Loaded MCP skill: %s", result.frontmatter.name)
+            else:
+                logger.debug(
+                    "Skipping skill index entry '%s'",
+                    entry.name or "(unnamed)",
+                )
+
+        logger.info("Successfully loaded %d skills from MCP server", len(skills))
+        return skills
+
+    async def _try_read_index(self) -> _McpSkillIndex | None:
+        """Attempt to read and parse ``skill://index.json`` from the MCP server.
+
+        Returns:
+            A parsed :class:`_McpSkillIndex`, or ``None`` if the index is
+            absent, empty, or malformed.
+        """
+        try:
+            result = await self._client.read_resource(_mcp_any_url(self._INDEX_URI))
+        except Exception as ex:
+            if _is_mcp_resource_not_found(ex):
+                logger.debug("No skill://index.json resource available on MCP server: %s", ex)
+                return None
+            logger.warning("Failed to read skill://index.json from MCP server.", exc_info=True)
+            raise
+
+        index_text = _mcp_join_text(result)
+        if not index_text:
+            logger.debug("skill://index.json on MCP server returned empty/non-text contents")
+            return None
+
+        try:
+            return _parse_mcp_skill_index(index_text)
+        except (json.JSONDecodeError, ValueError):
+            logger.warning("Failed to parse skill://index.json JSON document.", exc_info=True)
+            return None
+
+    def _try_create_skill(self, entry: _McpSkillIndexEntry) -> MCPSkill | None:
+        """Attempt to create an :class:`MCPSkill` from an index entry.
+
+        Args:
+            entry: A single entry from the skill index.
+
+        Returns:
+            An :class:`MCPSkill` if the entry is valid, or ``None`` if the
+            entry should be skipped.
+        """
+        if entry.type != self._SKILL_MD_TYPE:
+            logger.debug(
+                "Skipping entry '%s': unsupported type '%s'",
+                entry.name or "(unnamed)",
+                entry.type or "(none)",
+            )
+            return None
+
+        if not entry.name or not entry.name.strip():
+            logger.debug("Skipping entry: missing required 'name' field")
+            return None
+
+        if not entry.description or not entry.description.strip():
+            logger.debug("Skipping entry '%s': missing required 'description' field", entry.name)
+            return None
+
+        if not entry.url or not entry.url.strip():
+            logger.debug("Skipping entry '%s': missing required 'url' field", entry.name)
+            return None
+
+        try:
+            fm = SkillFrontmatter(name=entry.name, description=entry.description)
+        except ValueError as ex:
+            logger.debug("Skipping entry '%s': invalid metadata: %s", entry.name, ex)
+            return None
+
+        return MCPSkill(frontmatter=fm, skill_md_uri=entry.url, client=self._client)
+
+
 # endregion

python/packages/core/tests/core/test_mcp_skills.py

@@ -0,0 +1,667 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for MCP-based skills (MCPSkillsSource, MCPSkill, MCPSkillResource)."""
+
+from __future__ import annotations
+
+import base64
+import json
+from unittest.mock import AsyncMock
+
+import pytest
+from mcp.shared.exceptions import McpError
+from mcp.types import (
+    BlobResourceContents,
+    ErrorData,
+    ReadResourceResult,
+    TextResourceContents,
+)
+from pydantic import AnyUrl
+
+from agent_framework import MCPSkill, MCPSkillResource, MCPSkillsSource
+from agent_framework._skills import _parse_mcp_skill_index
+
+# ---------------------------------------------------------------------------
+# Fixtures & helpers
+# ---------------------------------------------------------------------------
+
+SAMPLE_SKILL_MD = """\
+---
+name: unit-converter
+description: Convert between common units.
+---
+# Unit Converter
+
+Body content here.
+"""
+
+SAMPLE_SKILL_INDEX = json.dumps(
+    {
+        "$schema": "https://schemas.agentskills.io/discovery/0.2.0/schema.json",
+        "skills": [
+            {
+                "name": "unit-converter",
+                "type": "skill-md",
+                "description": "Convert between common units.",
+                "url": "skill://unit-converter/SKILL.md",
+            }
+        ],
+    }
+)
+
+
+def _make_text_result(text: str, uri: str = "skill://test") -> ReadResourceResult:
+    """Create a ReadResourceResult with a single TextResourceContents."""
+    return ReadResourceResult(
+        contents=[TextResourceContents(uri=AnyUrl(uri), text=text, mimeType="text/markdown")]
+    )
+
+
+def _make_blob_result(
+    data: bytes,
+    uri: str = "skill://test",
+    mime_type: str = "application/octet-stream",
+) -> ReadResourceResult:
+    """Create a ReadResourceResult with a single BlobResourceContents."""
+    return ReadResourceResult(
+        contents=[BlobResourceContents(uri=AnyUrl(uri), blob=base64.b64encode(data).decode(), mimeType=mime_type)]
+    )
+
+
+def _make_empty_result() -> ReadResourceResult:
+    """Create a ReadResourceResult with no contents."""
+    return ReadResourceResult(contents=[])
+
+
+def _make_client(**read_resource_responses: ReadResourceResult) -> AsyncMock:
+    """Create a mock ClientSession whose read_resource returns different results per URI.
+
+    Args:
+        **read_resource_responses: Mapping of URI string to ReadResourceResult.
+            Any URI not in this mapping raises McpError with the MCP-spec
+            "Resource not found" code (-32002).
+    """
+    client = AsyncMock()
+
+    async def _read_resource(uri: AnyUrl) -> ReadResourceResult:
+        uri_str = str(uri)
+        if uri_str in read_resource_responses:
+            return read_resource_responses[uri_str]
+        raise McpError(error=ErrorData(code=-32002, message=f"Resource not found: {uri_str}"))
+
+    client.read_resource = AsyncMock(side_effect=_read_resource)
+    return client
+
+
+# ---------------------------------------------------------------------------
+# _parse_mcp_skill_index tests
+# ---------------------------------------------------------------------------
+
+
+class TestParseMCPSkillIndex:
+    """Tests for the _parse_mcp_skill_index helper."""
+
+    def test_parses_valid_index(self) -> None:
+        index = _parse_mcp_skill_index(SAMPLE_SKILL_INDEX)
+        assert index.schema == "https://schemas.agentskills.io/discovery/0.2.0/schema.json"
+        assert len(index.skills) == 1
+        assert index.skills[0].name == "unit-converter"
+        assert index.skills[0].type == "skill-md"
+        assert index.skills[0].url == "skill://unit-converter/SKILL.md"
+
+    def test_parses_empty_skills_array(self) -> None:
+        index = _parse_mcp_skill_index('{"$schema": "test", "skills": []}')
+        assert index.skills == []
+
+    def test_parses_missing_skills_key(self) -> None:
+        index = _parse_mcp_skill_index('{"$schema": "test"}')
+        assert index.skills == []
+
+    def test_raises_on_non_object(self) -> None:
+        with pytest.raises(ValueError, match="must be a JSON object"):
+            _parse_mcp_skill_index("[]")
+
+    def test_raises_on_invalid_json(self) -> None:
+        with pytest.raises(json.JSONDecodeError):
+            _parse_mcp_skill_index("not json")
+
+    def test_skips_non_dict_entries(self) -> None:
+        index = _parse_mcp_skill_index('{"skills": ["not-a-dict", {"name": "ok", "type": "skill-md"}]}')
+        assert len(index.skills) == 1
+        assert index.skills[0].name == "ok"
+
+
+# ---------------------------------------------------------------------------
+# MCPSkillResource tests
+# ---------------------------------------------------------------------------
+
+
+class TestMCPSkillResource:
+    """Tests for MCPSkillResource."""
+
+    @pytest.mark.asyncio
+    async def test_read_text_content(self) -> None:
+        result = _make_text_result("hello world")
+        resource = MCPSkillResource(name="test.md", result=result)
+        content = await resource.read()
+        assert content == "hello world"
+
+    @pytest.mark.asyncio
+    async def test_read_binary_content(self) -> None:
+        data = bytes([0x01, 0x02, 0x03, 0x04])
+        result = _make_blob_result(data)
+        resource = MCPSkillResource(name="icon.bin", result=result)
+        content = await resource.read()
+        assert content == data
+
+    @pytest.mark.asyncio
+    async def test_read_empty_returns_none(self) -> None:
+        result = _make_empty_result()
+        resource = MCPSkillResource(name="empty", result=result)
+        content = await resource.read()
+        assert content is None
+
+    @pytest.mark.asyncio
+    async def test_read_multiple_text_contents_joined(self) -> None:
+        result = ReadResourceResult(
+            contents=[
+                TextResourceContents(uri=AnyUrl("skill://a"), text="line1", mimeType="text/plain"),
+                TextResourceContents(uri=AnyUrl("skill://b"), text="line2", mimeType="text/plain"),
+            ]
+        )
+        resource = MCPSkillResource(name="multi", result=result)
+        content = await resource.read()
+        assert content == "line1\nline2"
+
+    @pytest.mark.asyncio
+    async def test_binary_takes_precedence_over_text(self) -> None:
+        data = b"\xff\xfe"
+        result = ReadResourceResult(
+            contents=[
+                TextResourceContents(uri=AnyUrl("skill://a"), text="text", mimeType="text/plain"),
+                BlobResourceContents(
+                    uri=AnyUrl("skill://b"),
+                    blob=base64.b64encode(data).decode(),
+                    mimeType="application/octet-stream",
+                ),
+            ]
+        )
+        resource = MCPSkillResource(name="mixed", result=result)
+        content = await resource.read()
+        # The implementation iterates all contents checking for BlobResourceContents
+        # first, so when both text and binary are present, binary is returned.
+        assert content == data
+
+
+# ---------------------------------------------------------------------------
+# MCPSkill tests
+# ---------------------------------------------------------------------------
+
+
+class TestMCPSkill:
+    """Tests for MCPSkill."""
+
+    @pytest.mark.asyncio
+    async def test_get_content_fetches_and_caches(self) -> None:
+        client = _make_client(**{"skill://unit-converter/SKILL.md": _make_text_result(SAMPLE_SKILL_MD)})
+        from agent_framework import SkillFrontmatter
+
+        fm = SkillFrontmatter(name="unit-converter", description="Convert between common units.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://unit-converter/SKILL.md", client=client)
+
+        content1 = await skill.get_content()
+        content2 = await skill.get_content()
+
+        assert "Body content here." in content1
+        assert content1 == content2
+        # Only one MCP call should be made (cached)
+        assert client.read_resource.call_count == 1
+
+    @pytest.mark.asyncio
+    async def test_get_content_raises_on_empty(self) -> None:
+        client = _make_client(**{"skill://empty/SKILL.md": _make_empty_result()})
+        from agent_framework import SkillFrontmatter
+
+        fm = SkillFrontmatter(name="empty-skill", description="Empty skill.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://empty/SKILL.md", client=client)
+
+        with pytest.raises(ValueError, match="no text content"):
+            await skill.get_content()
+
+    @pytest.mark.asyncio
+    async def test_get_resource_text(self) -> None:
+        client = _make_client(
+            **{
+                "skill://unit-converter/SKILL.md": _make_text_result(SAMPLE_SKILL_MD),
+                "skill://unit-converter/references/checklist.md": _make_text_result("- check thing 1\n- check thing 2"),
+            }
+        )
+        from agent_framework import SkillFrontmatter
+
+        fm = SkillFrontmatter(name="unit-converter", description="Convert between common units.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://unit-converter/SKILL.md", client=client)
+
+        resource = await skill.get_resource("references/checklist.md")
+        assert resource is not None
+        content = await resource.read()
+        assert content == "- check thing 1\n- check thing 2"
+
+    @pytest.mark.asyncio
+    async def test_get_resource_binary(self) -> None:
+        data = bytes([0x01, 0x02, 0x03, 0x04])
+        client = _make_client(
+            **{
+                "skill://unit-converter/SKILL.md": _make_text_result(SAMPLE_SKILL_MD),
+                "skill://unit-converter/assets/icon.bin": _make_blob_result(data),
+            }
+        )
+        from agent_framework import SkillFrontmatter
+
+        fm = SkillFrontmatter(name="unit-converter", description="Convert between common units.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://unit-converter/SKILL.md", client=client)
+
+        resource = await skill.get_resource("assets/icon.bin")
+        assert resource is not None
+        content = await resource.read()
+        assert content == data
+
+    @pytest.mark.asyncio
+    async def test_get_resource_unknown_returns_none(self) -> None:
+        client = _make_client(**{"skill://unit-converter/SKILL.md": _make_text_result(SAMPLE_SKILL_MD)})
+        from agent_framework import SkillFrontmatter
+
+        fm = SkillFrontmatter(name="unit-converter", description="Convert between common units.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://unit-converter/SKILL.md", client=client)
+
+        resource = await skill.get_resource("references/does-not-exist.md")
+        assert resource is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "name",
+        [
+            "../escape.md",
+            "references/../../escape.md",
+            "..",
+            "..\\escape.md",
+            "/etc/passwd",
+            "http://attacker.example.com/payload",
+        ],
+    )
+    async def test_get_resource_path_traversal_returns_none(self, name: str) -> None:
+        # Register a permissive mock that would happily return content for any URI,
+        # so the test fails unless the client-side validation rejects the name
+        # before issuing the read.
+        client = AsyncMock()
+        client.read_resource = AsyncMock(return_value=_make_text_result("should never be returned"))
+
+        from agent_framework import SkillFrontmatter
+
+        fm = SkillFrontmatter(name="unit-converter", description="Convert between common units.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://unit-converter/SKILL.md", client=client)
+
+        resource = await skill.get_resource(name)
+        assert resource is None
+        client.read_resource.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_get_resource_empty_name_returns_none(self) -> None:
+        client = _make_client()
+        from agent_framework import SkillFrontmatter
+
+        fm = SkillFrontmatter(name="test-skill", description="Test.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://test/SKILL.md", client=client)
+
+        assert await skill.get_resource("") is None
+        assert await skill.get_resource("   ") is None
+
+    @pytest.mark.asyncio
+    async def test_get_script_returns_none(self) -> None:
+        client = _make_client()
+        from agent_framework import SkillFrontmatter
+
+        fm = SkillFrontmatter(name="test-skill", description="Test.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://test/SKILL.md", client=client)
+
+        assert await skill.get_script("anything") is None
+
+    def test_compute_skill_root_uri_strips_suffix(self) -> None:
+        assert MCPSkill._compute_skill_root_uri("skill://unit-converter/SKILL.md") == "skill://unit-converter/"
+
+    def test_compute_skill_root_uri_trailing_slash(self) -> None:
+        assert MCPSkill._compute_skill_root_uri("skill://unit-converter/") == "skill://unit-converter/"
+
+    def test_compute_skill_root_uri_no_suffix_adds_slash(self) -> None:
+        assert MCPSkill._compute_skill_root_uri("skill://unit-converter") == "skill://unit-converter/"
+
+
+# ---------------------------------------------------------------------------
+# MCPSkillsSource tests
+# ---------------------------------------------------------------------------
+
+
+class TestMCPSkillsSource:
+    """Tests for MCPSkillsSource."""
+
+    @pytest.mark.asyncio
+    async def test_index_based_discovery_returns_skill(self) -> None:
+        client = _make_client(
+            **{
+                "skill://index.json": _make_text_result(SAMPLE_SKILL_INDEX, uri="skill://index.json"),
+                "skill://unit-converter/SKILL.md": _make_text_result(SAMPLE_SKILL_MD),
+            }
+        )
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+
+        assert len(skills) == 1
+        assert skills[0].frontmatter.name == "unit-converter"
+        assert skills[0].frontmatter.description == "Convert between common units."
+
+        # Content is fetched on demand, not during discovery
+        content = await skills[0].get_content()
+        assert "Body content here." in content
+
+    @pytest.mark.asyncio
+    async def test_no_index_returns_empty(self) -> None:
+        client = _make_client()  # No resources at all
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+        assert skills == []
+
+    @pytest.mark.asyncio
+    async def test_does_not_read_skill_md_during_discovery(self) -> None:
+        # Index points to a skill, but SKILL.md is not registered on the server.
+        # Discovery should succeed because it only reads the index.
+        client = _make_client(
+            **{"skill://index.json": _make_text_result(SAMPLE_SKILL_INDEX, uri="skill://index.json")}
+        )
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+
+        assert len(skills) == 1
+        assert skills[0].frontmatter.name == "unit-converter"
+
+    @pytest.mark.asyncio
+    async def test_invalid_name_is_skipped(self) -> None:
+        index_json = json.dumps(
+            {
+                "$schema": "https://schemas.agentskills.io/discovery/0.2.0/schema.json",
+                "skills": [
+                    {
+                        "name": "UnitConverter",  # Invalid: uppercase
+                        "type": "skill-md",
+                        "description": "Convert between common units.",
+                        "url": "skill://UnitConverter/SKILL.md",
+                    }
+                ],
+            }
+        )
+        client = _make_client(**{"skill://index.json": _make_text_result(index_json, uri="skill://index.json")})
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+        assert skills == []
+
+    @pytest.mark.asyncio
+    async def test_missing_required_fields_is_skipped(self) -> None:
+        index_json = json.dumps(
+            {
+                "$schema": "https://schemas.agentskills.io/discovery/0.2.0/schema.json",
+                "skills": [
+                    {
+                        "name": "unit-converter",
+                        "type": "skill-md",
+                        # Missing description and url
+                    }
+                ],
+            }
+        )
+        client = _make_client(**{"skill://index.json": _make_text_result(index_json, uri="skill://index.json")})
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+        assert skills == []
+
+    @pytest.mark.asyncio
+    async def test_unsupported_type_is_skipped(self) -> None:
+        index_json = json.dumps(
+            {
+                "$schema": "https://schemas.agentskills.io/discovery/0.2.0/schema.json",
+                "skills": [
+                    {
+                        "name": "some-skill",
+                        "type": "archive",
+                        "description": "Packaged skill.",
+                        "url": "skill://some-skill.tar.gz",
+                    }
+                ],
+            }
+        )
+        client = _make_client(**{"skill://index.json": _make_text_result(index_json, uri="skill://index.json")})
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+        assert skills == []
+
+    @pytest.mark.asyncio
+    async def test_template_type_is_skipped(self) -> None:
+        index_json = json.dumps(
+            {
+                "$schema": "https://schemas.agentskills.io/discovery/0.2.0/schema.json",
+                "skills": [
+                    {
+                        "type": "mcp-resource-template",
+                        "description": "Per-product documentation skill",
+                        "url": "skill://docs/{product}/SKILL.md",
+                    }
+                ],
+            }
+        )
+        client = _make_client(**{"skill://index.json": _make_text_result(index_json, uri="skill://index.json")})
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+        assert skills == []
+
+    @pytest.mark.asyncio
+    async def test_empty_index_returns_empty(self) -> None:
+        client = _make_client(
+            **{"skill://index.json": _make_text_result('{"skills": []}', uri="skill://index.json")}
+        )
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+        assert skills == []
+
+    @pytest.mark.asyncio
+    async def test_malformed_index_json_returns_empty(self) -> None:
+        client = _make_client(
+            **{"skill://index.json": _make_text_result("not valid json", uri="skill://index.json")}
+        )
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+        assert skills == []
+
+    @pytest.mark.asyncio
+    async def test_sibling_text_resource(self) -> None:
+        client = _make_client(
+            **{
+                "skill://index.json": _make_text_result(SAMPLE_SKILL_INDEX, uri="skill://index.json"),
+                "skill://unit-converter/SKILL.md": _make_text_result(SAMPLE_SKILL_MD),
+                "skill://unit-converter/references/checklist.md": _make_text_result("- check thing 1\n- check thing 2"),
+            }
+        )
+        source = MCPSkillsSource(client=client)
+        skill = (await source.get_skills())[0]
+        resource = await skill.get_resource("references/checklist.md")
+        assert resource is not None
+        content = await resource.read()
+        assert content == "- check thing 1\n- check thing 2"
+
+    @pytest.mark.asyncio
+    async def test_sibling_binary_resource(self) -> None:
+        data = bytes([0x01, 0x02, 0x03, 0x04])
+        client = _make_client(
+            **{
+                "skill://index.json": _make_text_result(SAMPLE_SKILL_INDEX, uri="skill://index.json"),
+                "skill://unit-converter/SKILL.md": _make_text_result(SAMPLE_SKILL_MD),
+                "skill://unit-converter/assets/icon.bin": _make_blob_result(data),
+            }
+        )
+        source = MCPSkillsSource(client=client)
+        skill = (await source.get_skills())[0]
+        resource = await skill.get_resource("assets/icon.bin")
+        assert resource is not None
+        content = await resource.read()
+        assert content == data
+
+
+# ---------------------------------------------------------------------------
+# McpError code branching tests
+# ---------------------------------------------------------------------------
+
+
+class TestMCPSkillsSourceErrorCodeBranching:
+    """Tests that MCPSkillsSource and MCPSkill branch on McpError.error.code.
+
+    Only "not found" codes (RESOURCE_NOT_FOUND -32002, METHOD_NOT_FOUND -32601)
+    should be silently swallowed as "no skills available." Other McpError codes
+    and non-McpError exceptions must propagate so that auth failures, server
+    crashes, and connection drops are visible.
+    """
+
+    @pytest.mark.asyncio
+    async def test_index_method_not_found_returns_empty(self) -> None:
+        """METHOD_NOT_FOUND (-32601) -> server doesn't support resources/read."""
+        client = AsyncMock()
+        client.read_resource = AsyncMock(side_effect=McpError(error=ErrorData(code=-32601, message="Method not found")))
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+        assert skills == []
+
+    @pytest.mark.asyncio
+    async def test_index_resource_not_found_returns_empty(self) -> None:
+        """MCP-spec "Resource not found" (-32002) -> server has no index."""
+        client = AsyncMock()
+        client.read_resource = AsyncMock(
+            side_effect=McpError(error=ErrorData(code=-32002, message="Resource not found"))
+        )
+        source = MCPSkillsSource(client=client)
+        skills = await source.get_skills()
+        assert skills == []
+
+    @pytest.mark.asyncio
+    async def test_index_invalid_params_propagates(self) -> None:
+        """INVALID_PARAMS (-32602) is a real bug, must propagate (not "not found")."""
+        client = AsyncMock()
+        client.read_resource = AsyncMock(side_effect=McpError(error=ErrorData(code=-32602, message="Invalid params")))
+        source = MCPSkillsSource(client=client)
+        with pytest.raises(McpError):
+            await source.get_skills()
+
+    @pytest.mark.asyncio
+    async def test_index_internal_error_propagates(self) -> None:
+        """INTERNAL_ERROR (-32603) must propagate, not silently return empty."""
+        client = AsyncMock()
+        client.read_resource = AsyncMock(side_effect=McpError(error=ErrorData(code=-32603, message="Internal error")))
+        source = MCPSkillsSource(client=client)
+        with pytest.raises(McpError):
+            await source.get_skills()
+
+    @pytest.mark.asyncio
+    async def test_index_connection_closed_propagates(self) -> None:
+        """CONNECTION_CLOSED (-32000) must propagate."""
+        client = AsyncMock()
+        client.read_resource = AsyncMock(
+            side_effect=McpError(error=ErrorData(code=-32000, message="Connection closed"))
+        )
+        source = MCPSkillsSource(client=client)
+        with pytest.raises(McpError):
+            await source.get_skills()
+
+    @pytest.mark.asyncio
+    async def test_index_generic_error_code_propagates(self) -> None:
+        """Generic handler error (code 0) must propagate."""
+        client = AsyncMock()
+        client.read_resource = AsyncMock(side_effect=McpError(error=ErrorData(code=0, message="Some handler error")))
+        source = MCPSkillsSource(client=client)
+        with pytest.raises(McpError):
+            await source.get_skills()
+
+    @pytest.mark.asyncio
+    async def test_index_non_mcp_error_propagates(self) -> None:
+        """Non-McpError exceptions (connection drop, timeout) must propagate."""
+        client = AsyncMock()
+        client.read_resource = AsyncMock(side_effect=ConnectionError("connection lost"))
+        source = MCPSkillsSource(client=client)
+        with pytest.raises(ConnectionError):
+            await source.get_skills()
+
+    @pytest.mark.asyncio
+    async def test_get_resource_internal_error_propagates(self) -> None:
+        """McpError with INTERNAL_ERROR on get_resource must propagate."""
+        from agent_framework import SkillFrontmatter
+
+        client = AsyncMock()
+        client.read_resource = AsyncMock(side_effect=McpError(error=ErrorData(code=-32603, message="Server crashed")))
+        fm = SkillFrontmatter(name="test-skill", description="Test.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://test/SKILL.md", client=client)
+        with pytest.raises(McpError):
+            await skill.get_resource("references/file.md")
+
+    @pytest.mark.asyncio
+    async def test_get_resource_not_found_returns_none(self) -> None:
+        """McpError with RESOURCE_NOT_FOUND (-32002) on get_resource returns None."""
+        from agent_framework import SkillFrontmatter
+
+        client = AsyncMock()
+        client.read_resource = AsyncMock(
+            side_effect=McpError(error=ErrorData(code=-32002, message="Resource not found"))
+        )
+        fm = SkillFrontmatter(name="test-skill", description="Test.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://test/SKILL.md", client=client)
+        result = await skill.get_resource("references/file.md")
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_get_resource_connection_error_propagates(self) -> None:
+        """A plain ConnectionError on get_resource must propagate, not return None."""
+        from agent_framework import SkillFrontmatter
+
+        client = AsyncMock()
+        client.read_resource = AsyncMock(side_effect=ConnectionError("connection lost"))
+        fm = SkillFrontmatter(name="test-skill", description="Test.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://test/SKILL.md", client=client)
+        with pytest.raises(ConnectionError):
+            await skill.get_resource("references/file.md")
+
+    @pytest.mark.asyncio
+    async def test_get_resource_timeout_error_propagates(self) -> None:
+        """A TimeoutError on get_resource must propagate, not return None."""
+        from agent_framework import SkillFrontmatter
+
+        client = AsyncMock()
+        client.read_resource = AsyncMock(side_effect=TimeoutError("read timed out"))
+        fm = SkillFrontmatter(name="test-skill", description="Test.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://test/SKILL.md", client=client)
+        with pytest.raises(TimeoutError):
+            await skill.get_resource("references/file.md")
+
+    @pytest.mark.asyncio
+    async def test_get_resource_generic_mcp_error_propagates(self) -> None:
+        """McpError with a generic code (0) on get_resource must propagate."""
+        from agent_framework import SkillFrontmatter
+
+        client = AsyncMock()
+        client.read_resource = AsyncMock(
+            side_effect=McpError(error=ErrorData(code=0, message="Handler error"))
+        )
+        fm = SkillFrontmatter(name="test-skill", description="Test.")
+        skill = MCPSkill(frontmatter=fm, skill_md_uri="skill://test/SKILL.md", client=client)
+        with pytest.raises(McpError):
+            await skill.get_resource("references/file.md")
+
+    @pytest.mark.asyncio
+    async def test_index_timeout_error_propagates(self) -> None:
+        """A TimeoutError reading skill://index.json must propagate."""
+        client = AsyncMock()
+        client.read_resource = AsyncMock(side_effect=TimeoutError("read timed out"))
+        source = MCPSkillsSource(client=client)
+        with pytest.raises(TimeoutError):
+            await source.get_skills()

python/samples/02-agents/providers/foundry/README.md

@@ -27,6 +27,7 @@ This folder contains Azure AI Foundry and Foundry Local samples for Agent Framew
 | [`foundry_chat_client_with_local_mcp.py`](foundry_chat_client_with_local_mcp.py) | Foundry Chat Client with local MCP |
 | [`foundry_chat_client_with_session.py`](foundry_chat_client_with_session.py) | Foundry Chat Client with session management |
 | [`foundry_chat_client_with_toolbox.py`](foundry_chat_client_with_toolbox.py) | Foundry Chat Client connected to a toolbox via its MCP endpoint using `MCPStreamableHTTPTool` |
+| [`foundry_chat_client_with_toolbox_skills.py`](foundry_chat_client_with_toolbox_skills.py) | Foundry Chat Client that discovers MCP-based skills from a Foundry Toolbox endpoint via `MCPSkillsSource` (uses an Azure AD bearer token and the toolbox preview header) |
 
 ## FoundryLocalClient Samples
 

python/samples/02-agents/providers/foundry/foundry_chat_client_with_toolbox_skills.py

@@ -0,0 +1,87 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import os
+from collections.abc import Generator
+
+import httpx
+from agent_framework import Agent, MCPSkillsSource, SkillsProvider
+from agent_framework.foundry import FoundryChatClient
+from azure.core.credentials import TokenCredential
+from azure.identity import AzureCliCredential, get_bearer_token_provider
+from dotenv import load_dotenv
+from mcp.client.session import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+
+# Load environment variables from .env file
+load_dotenv()
+
+"""
+Foundry Chat Client with Toolbox-Hosted Skills
+
+Discover Agent Skills served by a Microsoft Foundry Toolbox MCP endpoint
+and inject them into a ``FoundryChatClient`` agent via ``MCPSkillsSource``.
+The toolbox's discovery document (``skill://index.json``) is read once at
+startup; SKILL.md bodies are fetched on demand as the agent uses them.
+
+Prerequisites:
+- A Microsoft Foundry project with a toolbox that exposes
+  ``skill://index.json`` with ``skill-md`` entries
+- FOUNDRY_PROJECT_ENDPOINT and FOUNDRY_MODEL environment variables set
+- FOUNDRY_TOOLBOX_MCP_SERVER_URL: the toolbox's MCP endpoint URL, e.g.
+  ``https://<account>.services.ai.azure.com/api/projects/<project>/toolboxes/<name>/mcp?api-version=v1``
+- Azure CLI authentication (``az login``)
+"""
+
+
+class _BearerAuth(httpx.Auth):
+    """Attach a fresh Foundry bearer token to every request."""
+
+    def __init__(self, credential: TokenCredential) -> None:
+        self._get_token = get_bearer_token_provider(credential, "https://ai.azure.com/.default")
+
+    def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, httpx.Response, None]:
+        request.headers["Authorization"] = f"Bearer {self._get_token()}"
+        yield request
+
+
+async def main() -> None:
+    """Example showing toolbox-hosted MCP skills for a Foundry Chat Client agent."""
+    credential = AzureCliCredential()
+
+    # HTTP client that signs every request with a fresh Foundry bearer token
+    # and advertises the toolbox preview feature flag, plus the MCP streamable
+    # HTTP transport that uses it.
+    async with (
+        httpx.AsyncClient(
+            auth=_BearerAuth(credential),
+            headers={"Foundry-Features": "Toolboxes=V1Preview"},
+            timeout=httpx.Timeout(30.0, read=300.0),
+            follow_redirects=True,
+        ) as http_client,
+        streamable_http_client(
+            url=os.environ["FOUNDRY_TOOLBOX_MCP_SERVER_URL"],
+            http_client=http_client,
+        ) as (read, write, _),
+        ClientSession(read, write) as session,
+    ):
+        await session.initialize()
+
+        # Discover skills served by the toolbox and inject them as a context provider.
+        skills_provider = SkillsProvider(MCPSkillsSource(client=session))
+
+        async with Agent(
+            client=FoundryChatClient(credential=credential),
+            name="ToolboxMCPSkillsAgent",
+            instructions="You are a helpful assistant. Use available skills to answer the user.",
+            context_providers=[skills_provider],
+        ) as agent:
+            query = input("User: ").strip()  # noqa: ASYNC250
+            if not query:
+                return
+            response = await agent.run(query)
+            print(f"Assistant: {response.text}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/02-agents/skills/README.md

@@ -12,6 +12,7 @@ Start with file-based or code-defined skills, then explore combining them and ad
 | [**code_defined_skill**](code_defined_skill/) | Define skills entirely in Python code using `Skill`, `@skill.resource`, and `@skill.script` decorators. Uses a code-defined unit-converter skill. |
 | [**class_based_skill**](class_based_skill/) | Define skills as Python classes using `ClassSkill` with `@ClassSkill.resource` and `@ClassSkill.script` decorators for auto-discovery. Uses a class-based unit-converter skill. |
 | [**mixed_skills**](mixed_skills/) | Combine code-defined, class-based, and file-based skills in a single agent. Uses a code-defined volume-converter, a class-based temperature-converter, and a file-based unit-converter. |
+| [**mcp_based_skill**](mcp_based_skill/) | Discover skills served over the [Model Context Protocol (MCP)](https://modelcontextprotocol.io) via `MCPSkillsSource`. Connects to a remote MCP server that exposes skills as `skill://...` resources following the SEP-2640 convention. |
 | [**script_approval**](script_approval/) | Require human-in-the-loop approval before executing skill scripts |
 
 ## Key Concepts

python/samples/02-agents/skills/mcp_based_skill/README.md

@@ -0,0 +1,51 @@
+# MCP-Based Agent Skills Sample
+
+This sample demonstrates how to discover **Agent Skills served over MCP** with an `Agent`.
+
+## What it demonstrates
+
+- Connecting to a remote MCP server (over streamable HTTP) that exposes skill
+  resources following the SEP-2640 convention.
+- Building a `SkillsProvider` from an `MCPSkillsSource`, which reads
+  `skill://index.json` (SEP-2640 canonical discovery) and constructs skills from
+  the index entries.
+- The progressive disclosure pattern across MCP: advertise → load → read
+  resources, exactly as for filesystem-backed skills.
+
+## Running the Sample
+
+### Prerequisites
+
+- Python 3.10+
+- An [Azure AI Foundry](https://ai.azure.com/) project with a deployed model
+- Azure CLI authentication (`az login`)
+- A running MCP server that hosts SEP-2640 skill resources (see "Providing
+  an MCP server" below)
+
+### Setup
+
+Set the following environment variables (in a `.env` file or your shell):
+
+```powershell
+$env:FOUNDRY_PROJECT_ENDPOINT="https://your-endpoint.services.ai.azure.com/api/projects/your-project"
+$env:FOUNDRY_MODEL="gpt-4o-mini"
+$env:MCP_SKILLS_SERVER_URL="https://your-mcp-server.example.com/mcp"
+```
+
+### Run
+
+```powershell
+python mcp_based_skill.py
+```
+
+## Providing an MCP server
+
+This sample is a **consumer**: it does not host an MCP server itself. To try
+it end-to-end you need an MCP server that exposes the SEP-2640 skill
+resources (`skill://index.json` plus per-skill `SKILL.md`).
+
+- See [`samples/02-agents/mcp/agent_as_mcp_server.py`](../../mcp/agent_as_mcp_server.py)
+  for an example of hosting an MCP server via the Agent Framework.
+- The Model Context Protocol working group maintains reference MCP-skills
+  servers at
+  [`modelcontextprotocol/experimental-ext-skills`](https://github.com/modelcontextprotocol/experimental-ext-skills).

python/samples/02-agents/skills/mcp_based_skill/mcp_based_skill.py

@@ -0,0 +1,75 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import os
+
+# Uncomment this filter to suppress the experimental Skills warning before
+# using the sample's Skills APIs.
+# import warnings
+# warnings.filterwarnings("ignore", message=r"\[SKILLS\].*", category=FutureWarning)
+from agent_framework import Agent, MCPSkillsSource, SkillsProvider
+from agent_framework.foundry import FoundryChatClient
+from azure.identity import AzureCliCredential
+from dotenv import load_dotenv
+from mcp.client.session import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+
+"""
+MCP-Based Agent Skills
+
+This sample demonstrates how to discover Agent Skills served over the
+Model Context Protocol (MCP) using :class:`MCPSkillsSource`.
+
+The sample connects to a remote MCP server that exposes skill resources
+under the ``skill://`` URI scheme:
+
+* ``skill://index.json`` — discovery document listing all skills
+* ``skill://<skill-name>/SKILL.md`` — the skill instructions
+
+To run, set ``MCP_SKILLS_SERVER_URL`` to the streamable HTTP endpoint of an
+MCP server that hosts the skill resources.
+"""
+
+
+async def main() -> None:
+    """Connect to a remote MCP skills server and run the agent."""
+    load_dotenv()
+
+    endpoint = os.environ["FOUNDRY_PROJECT_ENDPOINT"]
+    deployment = os.environ.get("FOUNDRY_MODEL", "gpt-4o-mini")
+    mcp_url = os.environ["MCP_SKILLS_SERVER_URL"]
+
+    print("Discovering MCP-based skills")
+    print("-" * 60)
+
+    # 1. Connect to the MCP server over streamable HTTP.
+    async with streamable_http_client(url=mcp_url) as (read, write, _), ClientSession(read, write) as session:
+        await session.initialize()
+
+        # 2. Build a SkillsProvider that discovers skills over MCP.
+        #    MCPSkillsSource reads skill://index.json and creates one
+        #    MCPSkill per skill-md entry; SKILL.md bodies are fetched
+        #    on demand via resources/read.
+        skills_provider = SkillsProvider(MCPSkillsSource(client=session))
+
+        # 3. Run the agent.
+        client = FoundryChatClient(
+            project_endpoint=endpoint,
+            model=deployment,
+            credential=AzureCliCredential(),
+        )
+
+        async with Agent(
+            client=client,
+            instructions="You are a helpful assistant. Use available skills to answer the user.",
+            context_providers=[skills_provider],
+        ) as agent:
+            query = input("User: ").strip()  # noqa: ASYNC250
+            if not query:
+                return
+            response = await agent.run(query)
+            print(f"Agent: {response}\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())