Add Skills-over-MCP SEP support to host

Implements the io.modelcontextprotocol/skills extension (pre-submission
SEP at modelcontextprotocol/experimental-ext-skills#69): skills served
by connected MCP servers under the skill:// URI scheme are discovered
via the well-known skill://index.json resource and surfaced through the
existing skills pipeline so they behave identically to filesystem skills.

- SkillManifest gains optional uri/server_name; path becomes optional
- format_skills_for_prompt renders skill:// URIs as <location>/<directory>
  for MCP-backed skills and adds an MCP-specific preamble note
- New mcp_skills_loader module: fetches skill://index.json, parses
  concrete skill-md entries, logs SEP capability declaration when servers
  advertise it (template entries deferred)
- SkillReader accepts skill:// URIs, dispatches via aggregator.get_resource,
  enforces a URI-root trust boundary mirroring the filesystem allowed-dirs
- McpAgent.initialize fetches MCP skills post-connect and merges with
  filesystem manifests; filesystem wins on name collision
- MCPServerSettings gains mcp_skills bool for per-server opt-out

16 new unit tests; 80/80 skills suite passing.

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

Author: olaservo

src/fast_agent/agents/mcp_agent.py

@@ -316,6 +316,11 @@ async def initialize(self) -> None:
         """
         await self.__aenter__()
 
+        # Discover Skills-over-MCP skills from connected servers and merge
+        # them with any filesystem manifests before the instruction template
+        # is built (so the frontmatter lands in {{agentSkills}}).
+        await self._load_mcp_skill_manifests()
+
         # Apply template substitution to the instruction with server instructions
         await self._apply_instruction_templates()
 
@@ -583,11 +588,76 @@ def _warn_if_invalid_shell_working_directory(self, working_directory: Path | Non
                 surface="startup_once",
             )
 
+    async def _load_mcp_skill_manifests(self) -> None:
+        """Fetch skills served by connected MCP servers per the Skills-over-MCP SEP.
+
+        Merges discovered MCP manifests with any pre-existing filesystem
+        manifests and updates the skill reader. On name collision, the
+        filesystem manifest wins (consistent with SkillRegistry dedup).
+        Disabled per-server via MCPServerSettings.mcp_skills.
+        """
+        try:
+            from fast_agent.mcp.mcp_skills_loader import load_mcp_skill_manifests
+        except Exception:  # pragma: no cover - import guard
+            return
+
+        server_names = tuple(self._aggregator.server_names or ())
+        if not server_names:
+            return
+
+        # Collect per-server opt-out from config (default: enabled).
+        enabled_servers: set[str] | None = None
+        if self._context and self._context.config and self._context.config.mcp:
+            server_settings = self._context.config.mcp.servers or {}
+            enabled_servers = {
+                name
+                for name in server_names
+                if getattr(server_settings.get(name), "mcp_skills", True)
+            }
+            if not enabled_servers:
+                return
+
+        try:
+            mcp_manifests = await load_mcp_skill_manifests(
+                self._aggregator,
+                server_names,
+                enabled_servers=enabled_servers,
+            )
+        except Exception as exc:
+            self.logger.warning(f"Failed to load MCP skills: {exc}")
+            return
+
+        if not mcp_manifests:
+            return
+
+        filesystem_names = {m.name.lower() for m in self._skill_manifests}
+        merged = list(self._skill_manifests)
+        for mcp_manifest in mcp_manifests:
+            key = mcp_manifest.name.lower()
+            if key in filesystem_names:
+                self._record_warning(
+                    f"[dim]MCP-served skill '{mcp_manifest.name}' from server "
+                    f"'{mcp_manifest.server_name}' hidden by local filesystem skill.[/dim]",
+                    surface="startup_once",
+                )
+                continue
+            merged.append(mcp_manifest)
+            filesystem_names.add(key)
+
+        self.set_skill_manifests(merged)
+
     def set_skill_manifests(self, manifests: Sequence[SkillManifest]) -> None:
         self._skill_manifests = list(manifests)
         self._skill_map = {manifest.name: manifest for manifest in self._skill_manifests}
         if self._skill_manifests:
-            self._skill_reader = SkillReader(self._skill_manifests, self.logger)
+            # The aggregator is only needed when any manifest is URI-backed
+            # (Skills-over-MCP), but passing it unconditionally is cheap and
+            # keeps the reader uniform.
+            self._skill_reader = SkillReader(
+                self._skill_manifests,
+                self.logger,
+                aggregator=self._aggregator,
+            )
             self._ensure_shell_runtime_for_skills()
         else:
             self._skill_reader = None

src/fast_agent/config.py

@@ -380,6 +380,11 @@ class MCPServerSettings(BaseModel):
     include_instructions: bool = True
     """Whether to include this server's instructions in the system prompt (default: True)."""
 
+    mcp_skills: bool = True
+    """Whether to discover and load Skills-over-MCP (io.modelcontextprotocol/skills)
+    skills from this server (default: True). Set False to suppress reading
+    `skill://index.json` and its entries from this server."""
+
     reconnect_on_disconnect: bool = True
     """Whether to automatically reconnect when the server session is terminated (e.g., 404).
 

src/fast_agent/mcp/mcp_skills_loader.py

@@ -0,0 +1,258 @@
+"""Load skills served by connected MCP servers per the Skills-over-MCP SEP.
+
+Implements the `io.modelcontextprotocol/skills` extension: fetches the
+well-known `skill://index.json` resource from each connected server, parses
+its entries, and builds `SkillManifest` objects backed by `skill://` URIs.
+
+SEP: https://github.com/modelcontextprotocol/experimental-ext-skills/pull/69
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Iterable, Sequence
+
+from mcp.types import TextResourceContents
+
+from fast_agent.core.logging.logger import get_logger
+from fast_agent.skills.registry import SkillManifest, SkillRegistry
+
+if TYPE_CHECKING:
+    from fast_agent.mcp.mcp_aggregator import MCPAggregator
+
+logger = get_logger(__name__)
+
+INDEX_URI = "skill://index.json"
+SKILLS_EXTENSION_ID = "io.modelcontextprotocol/skills"
+
+
+async def load_mcp_skill_manifests(
+    aggregator: "MCPAggregator",
+    server_names: Sequence[str],
+    *,
+    enabled_servers: set[str] | None = None,
+) -> list[SkillManifest]:
+    """Fetch and parse skill manifests from connected MCP servers.
+
+    For each server, reads `skill://index.json` (optional; missing index is
+    silently skipped), then fetches each listed `SKILL.md` resource and parses
+    its frontmatter. Returns one `SkillManifest` per concrete `skill-md` index
+    entry. `mcp-resource-template` entries are logged and skipped (future work).
+
+    Errors from a single server or entry are logged as warnings; a failure
+    never aborts the whole batch.
+    """
+
+    manifests: list[SkillManifest] = []
+    for server_name in server_names:
+        if enabled_servers is not None and server_name not in enabled_servers:
+            logger.debug(
+                "MCP skill discovery disabled for server",
+                data={"server": server_name},
+            )
+            continue
+
+        await _log_capability_declaration(aggregator, server_name)
+
+        entries = await _read_index(aggregator, server_name)
+        if not entries:
+            continue
+
+        for entry in entries:
+            entry_type = entry.get("type")
+            if entry_type == "mcp-resource-template":
+                logger.debug(
+                    "Skipping MCP skill template entry (not yet supported)",
+                    data={
+                        "server": server_name,
+                        "url": entry.get("url"),
+                    },
+                )
+                continue
+            if entry_type != "skill-md":
+                logger.debug(
+                    "Skipping MCP skill entry with unrecognized type",
+                    data={"server": server_name, "type": entry_type},
+                )
+                continue
+            manifest = await _load_concrete_entry(aggregator, server_name, entry)
+            if manifest is not None:
+                manifests.append(manifest)
+
+    return manifests
+
+
+async def _log_capability_declaration(
+    aggregator: "MCPAggregator", server_name: str
+) -> None:
+    """Observability: note when a server declares the skills extension capability.
+
+    Per SEP, servers declare `capabilities.extensions["io.modelcontextprotocol/skills"]`
+    in their `initialize` response. Hosts don't need the declaration to load
+    skills (direct resource read works either way), but logging when present
+    helps diagnose discovery issues.
+    """
+    try:
+        caps = await aggregator.get_capabilities(server_name)
+    except Exception as exc:  # pragma: no cover - observability only
+        logger.debug(
+            "Could not read server capabilities for skills extension check",
+            data={"server": server_name, "error": str(exc)},
+        )
+        return
+
+    if caps is None:
+        return
+
+    # ServerCapabilities is a pydantic model; `extensions` may live there or
+    # in model_extra depending on the MCP SDK version.
+    extensions = getattr(caps, "extensions", None)
+    if extensions is None:
+        model_extra = getattr(caps, "model_extra", None)
+        if isinstance(model_extra, dict):
+            extensions = model_extra.get("extensions")
+    if not extensions:
+        return
+
+    declared = (
+        extensions.get(SKILLS_EXTENSION_ID)
+        if isinstance(extensions, dict)
+        else getattr(extensions, SKILLS_EXTENSION_ID, None)
+    )
+    if declared is not None:
+        logger.info(
+            "Server declares Skills-over-MCP extension",
+            data={"server": server_name, "extension": SKILLS_EXTENSION_ID},
+        )
+
+
+async def _read_index(
+    aggregator: "MCPAggregator", server_name: str
+) -> list[dict] | None:
+    """Read and parse `skill://index.json` from a server; returns None if absent."""
+    try:
+        result = await aggregator.get_resource(INDEX_URI, server_name=server_name)
+    except Exception as exc:
+        # The SEP treats the index as optional. Absence / lack of resources
+        # support / network error all fall through to "no indexed skills".
+        logger.debug(
+            "No skill index from server",
+            data={"server": server_name, "error": str(exc)},
+        )
+        return None
+
+    text = _first_text(result.contents)
+    if not text:
+        logger.warning(
+            "Skill index has no text content",
+            data={"server": server_name},
+        )
+        return None
+
+    try:
+        parsed = json.loads(text)
+    except Exception as exc:
+        logger.warning(
+            "Failed to parse skill index JSON",
+            data={"server": server_name, "error": str(exc)},
+        )
+        return None
+
+    skills = parsed.get("skills") if isinstance(parsed, dict) else None
+    if not isinstance(skills, list):
+        logger.warning(
+            "Skill index missing `skills` array",
+            data={"server": server_name},
+        )
+        return None
+    return [entry for entry in skills if isinstance(entry, dict)]
+
+
+async def _load_concrete_entry(
+    aggregator: "MCPAggregator",
+    server_name: str,
+    entry: dict,
+) -> SkillManifest | None:
+    url = entry.get("url")
+    if not isinstance(url, str) or not url:
+        logger.warning(
+            "Skill entry missing `url`",
+            data={"server": server_name, "entry": entry},
+        )
+        return None
+
+    try:
+        result = await aggregator.get_resource(url, server_name=server_name)
+    except Exception as exc:
+        logger.warning(
+            "Failed to read MCP skill SKILL.md",
+            data={"server": server_name, "url": url, "error": str(exc)},
+        )
+        return None
+
+    text = _first_text(result.contents)
+    if not text:
+        logger.warning(
+            "MCP skill SKILL.md has no text content",
+            data={"server": server_name, "url": url},
+        )
+        return None
+
+    manifest, parse_error = SkillRegistry.parse_manifest_text(text)
+    if manifest is None:
+        logger.warning(
+            "Failed to parse MCP skill frontmatter",
+            data={"server": server_name, "url": url, "error": parse_error},
+        )
+        return None
+
+    # SEP: the final segment of <skill-path> MUST equal the frontmatter name.
+    # If index or URI disagree with the frontmatter, frontmatter wins — the
+    # spec says the skill's identity is its `name` field — but log it.
+    url_name = _skill_name_from_uri(url)
+    if url_name and url_name != manifest.name:
+        logger.warning(
+            "MCP skill URI final segment differs from frontmatter name",
+            data={
+                "server": server_name,
+                "url": url,
+                "url_name": url_name,
+                "frontmatter_name": manifest.name,
+            },
+        )
+
+    return SkillManifest(
+        name=manifest.name,
+        description=manifest.description,
+        body=manifest.body,
+        path=None,
+        license=manifest.license,
+        compatibility=manifest.compatibility,
+        metadata=manifest.metadata,
+        allowed_tools=manifest.allowed_tools,
+        uri=url,
+        server_name=server_name,
+    )
+
+
+def _first_text(contents: Iterable) -> str | None:
+    for item in contents:
+        if isinstance(item, TextResourceContents):
+            return item.text
+    return None
+
+
+def _skill_name_from_uri(url: str) -> str | None:
+    """Return the final `<skill-path>` segment from a `skill://.../SKILL.md` URI."""
+    suffix = "/SKILL.md"
+    if not url.endswith(suffix):
+        return None
+    stem = url[: -len(suffix)]
+    # Strip scheme: skill://<path...>
+    scheme_sep = stem.find("://")
+    if scheme_sep != -1:
+        stem = stem[scheme_sep + 3 :]
+    # Last path segment
+    if "/" in stem:
+        return stem.rsplit("/", 1)[-1]
+    return stem or None