feat(mcp): model-facing resource tools with paginated list/read/templates (RFC-0058)

changelog/unreleased/2026-08-20-mcp-resource-integration.md

@@ -0,0 +1,12 @@
+# Unified MCP Resource integration
+
+AgentPool now exposes exactly three model-facing MCP Resource tools:
+`list_mcp_resources`, `list_mcp_resource_templates`, and `read_mcp_resource`.
+Listings preserve MCP metadata and use opaque host cursors; reads require an
+explicit server and URI, with structured errors, text truncation, and bounded
+binary attachment handling.
+
+Resource-capable MCP providers are negotiated and registered independently of
+the `resources.enabled` model-tool gate. Host catalog enumeration and
+`ResourceSource` injection use the configured server display name, preserving
+same-URI isolation across providers.

changelog/unreleased/README.md

@@ -0,0 +1,3 @@
+# Unreleased
+
+- [2026-08-20] Unify MCP Resource discovery, pagination, server-qualified reads, and Host ResourceSource routing. ([details](2026-08-20-mcp-resource-integration.md))

src/wolfharness/agents/native_agent/agent.py

@@ -1158,7 +1158,7 @@ async def get_agentlet[AgentOutputType](  # noqa: PLR0915
             #    Per-agent opt-out via ``resources.enabled: false`` in YAML.
             if self.config is not None and self.config.resources.enabled:
                 resource_cap = pool.resource_capability
-                if resource_cap is not None:
+                if resource_cap is not None and resource_cap not in self._external_capabilities:
                     tool_capabilities.append(resource_cap)
 
         # Register per-session capabilities (MCP, SkillManagerCap)
@@ -1178,9 +1178,20 @@ async def get_agentlet[AgentOutputType](  # noqa: PLR0915
                         ScopeLevel,
                     )
 
-                    session_scope = Scope(level=ScopeLevel.SESSION, session_id=session_id)
+                    session_scope = Scope(
+                        level=ScopeLevel.SESSION,
+                        agent_name=self.name,
+                        session_id=session_id,
+                    )
                     for cap in mcp_capabilities:
                         registry.register(cap, session_scope)
+                    # Agent/session MCP managers may own additional resource
+                    # providers. Pool-shared providers are already registered
+                    # at POOL scope by AgentFactory and must not be duplicated.
+                    if self.host_context is None or self.mcp is not self.host_context.mcp:
+                        for provider in self.mcp.get_mcp_providers():
+                            if provider.resources_supported is not False:
+                                registry.register(provider, session_scope)
                     if pool is not None:
                         pool_caps = pool.skill_capabilities
                         if pool_caps:

src/wolfharness/capabilities/extension_registry.py

@@ -24,6 +24,7 @@
 from typing import TYPE_CHECKING, Any
 import warnings
 
+from wolfharness.capabilities.uri_scheme_registry import UriSchemeRegistry
 from wolfharness.log import get_logger
 
 
@@ -36,6 +37,7 @@
     from wolfharness.capabilities.resource_protocols import (
         ChangeObservable,
         CommandResource,
+        McpResourceProvider,
         ResourceAccess,
         ResourceTemplateAccess,
         SkillResource,
@@ -126,15 +128,20 @@ class ExtensionRegistry:
     def __init__(
         self,
         max_composition_depth: int = DEFAULT_MAX_COMPOSITION_DEPTH,
+        scheme_registry: UriSchemeRegistry | None = None,
     ) -> None:
         """Initialize the registry with empty scope storage.
 
         Args:
             max_composition_depth: Maximum composition depth (root-inclusive).
                 When depth exceeds this limit, a warning is logged but
                 registration is NOT blocked. Default: 3.
+            scheme_registry: Optional ``UriSchemeRegistry`` for scheme-based
+                resource routing. When ``None``, routing falls back to the
+                legacy iteration over all providers.
         """
         self._max_composition_depth = max_composition_depth
+        self._scheme_registry = scheme_registry or UriSchemeRegistry()
 
         # 4-level scope storage: POOL > AGENT > SESSION > TURN
         self._pool: list[AbstractCapability[Any]] = []
@@ -315,6 +322,20 @@ def clear_session(self, session_id: str) -> None:
         self._session.pop(session_id, None)
         self._turn.pop(session_id, None)
 
+    # ------------------------------------------------------------------
+    # Properties
+    # ------------------------------------------------------------------
+
+    @property
+    def scheme_registry(self) -> UriSchemeRegistry:
+        """Return the URI scheme registry for resource routing.
+
+        Returns:
+            The ``UriSchemeRegistry`` instance, which is created
+            automatically during ``__init__`` if one was not provided.
+        """
+        return self._scheme_registry
+
     # ------------------------------------------------------------------
     # Query
     # ------------------------------------------------------------------
@@ -387,7 +408,24 @@ def get_resource_access(
         from wolfharness.capabilities.resource_protocols import ResourceAccess
 
         return [
-            cap for cap in self.get_visible_capabilities(scope) if isinstance(cap, ResourceAccess)
+            cap
+            for cap in self.get_visible_capabilities(scope)
+            if isinstance(cap, ResourceAccess)
+            and getattr(cap, "resources_supported", None) is not False
+        ]
+
+    def get_mcp_resource_providers(
+        self,
+        scope: Scope,
+    ) -> list[McpResourceProvider]:
+        """Get visible providers implementing the paged MCP Resource contract."""
+        from wolfharness.capabilities.resource_protocols import McpResourceProvider
+
+        return [
+            cap
+            for cap in self.get_visible_capabilities(scope)
+            if isinstance(cap, McpResourceProvider)
+            and getattr(cap, "resources_supported", None) is not False
         ]
 
     def get_tool_access(

src/wolfharness/capabilities/mcp_server_cap.py

@@ -30,6 +30,9 @@
     CommandResource,
     CompletionArgument,
     CompletionResult,
+    McpResourceListPage,
+    McpResourceProvider,
+    McpResourceTemplateListPage,
     ResourceAccess,
     ResourceEntry,
     ResourceTemplateAccess,
@@ -40,6 +43,7 @@
     ToolAccess,
     ToolEntry,
     ToolResult,
+    normalize_mcp_json_object,
 )
 
 
@@ -69,6 +73,7 @@ class McpServerCap(
     AbstractCapability[AgentDepsT],
     ToolAccess,
     ResourceAccess,
+    McpResourceProvider,
     ResourceTemplateAccess,
     SkillResource,
     CommandResource,
@@ -135,6 +140,10 @@ def __init__(
         self._name = name or self._config.client_id
         self._tool_prefix = tool_prefix
         self._client: MCPClient | None = client
+        # ``None`` means that the MCP initialize handshake has not yet been
+        # queried.  The registry treats an explicit ``False`` as a tools-only
+        # server and keeps it out of the Resource provider view.
+        self._resources_supported: bool | None = None
         self._change_queues: set[asyncio.Queue[ChangeEvent]] = set()
         self._resources_cache: list[ResourceEntry] | None = None
         self._resource_templates_cache: list[ResourceTemplateEntry] | None = None
@@ -162,6 +171,11 @@ def client(self) -> MCPClient | None:
         """Return the wrapped MCP client, or None if not yet initialized."""
         return self._client
 
+    @property
+    def resources_supported(self) -> bool | None:
+        """Return the cached MCP Resource capability state, when known."""
+        return self._resources_supported
+
     # ---- Lazy client initialization ----
 
     async def _ensure_client(self) -> MCPClient:
@@ -321,6 +335,12 @@ async def _build_toolset(
             tools = await client.list_tools()
             if not tools:
                 return None
+            # Apply enabled_tools/disabled_tools filtering from config.
+            # _config is a BaseMCPServerConfig (or subclass) which provides
+            # is_tool_allowed() and needs_tool_filtering().
+            config = self._config
+            if config.needs_tool_filtering():
+                tools = [t for t in tools if config.is_tool_allowed(t.name)]
             from pydantic_ai.toolsets import CombinedToolset, FunctionToolset, PrefixedToolset
 
             from wolfharness.capabilities.tool_schema_overlap_config import (
@@ -433,6 +453,89 @@ async def call_tool(self, name: str, args: dict[str, Any]) -> ToolResult:
 
     # ---- ResourceAccess ----
 
+    @property
+    def client_name(self) -> str:
+        """Return the stable configured MCP client/server identifier."""
+        display_name = getattr(self._config, "display_name", None)
+        if isinstance(display_name, str) and display_name.strip():
+            return display_name.strip()
+        return self._config.client_id
+
+    @property
+    def server_name(self) -> str:
+        """Return the stable configured server identifier."""
+        return self.client_name
+
+    async def supports_resources(self) -> bool:
+        """Return whether this upstream MCP server declared resources."""
+        client = await self._ensure_client()
+        try:
+            supported = await client.supports_resources()
+        except (OSError, RuntimeError, TimeoutError, ValueError):
+            # A failed handshake must not leave an ambiguous provider in the
+            # Resource registry; the MCP tool connection remains available.
+            self._resources_supported = False
+            raise
+        self._resources_supported = supported
+        return supported
+
+    async def list_resources_page(self, cursor: str | None = None) -> McpResourceListPage:
+        """Return one upstream resource page with complete MCP metadata."""
+        client = await self._ensure_client()
+        return await client.list_resources_mcp(cursor)
+
+    async def list_resource_templates_page(
+        self, cursor: str | None = None
+    ) -> McpResourceTemplateListPage:
+        """Return one upstream resource-template page."""
+        client = await self._ensure_client()
+        return await client.list_resource_templates_mcp(cursor)
+
+    async def read_mcp_resource(
+        self, uri: str
+    ) -> list[TextResourceContent | BlobResourceContent] | None:
+        """Read a resource through the MCP Resource contract."""
+        client = await self._ensure_client()
+        contents = await client.read_resource(uri)
+        return self._convert_resource_contents(uri, contents)
+
+    @staticmethod
+    def _convert_resource_contents(
+        uri: str,
+        contents: Sequence[Any],
+    ) -> list[TextResourceContent | BlobResourceContent] | None:
+        """Convert MCP content blocks to shared protocol content types."""
+        if not contents:
+            return None
+        result: list[TextResourceContent | BlobResourceContent] = []
+        for content in contents:
+            text_value: str | None = getattr(content, "text", None)
+            if text_value is not None:
+                result.append(
+                    TextResourceContent(
+                        uri=uri,
+                        mime_type=getattr(content, "mimeType", None),
+                        meta=normalize_mcp_json_object(
+                            getattr(content, "meta", getattr(content, "_meta", None))
+                        ),
+                        text=text_value,
+                    )
+                )
+                continue
+            blob_value: str | None = getattr(content, "blob", None)
+            if blob_value is not None:
+                result.append(
+                    BlobResourceContent(
+                        uri=uri,
+                        mime_type=getattr(content, "mimeType", None),
+                        meta=normalize_mcp_json_object(
+                            getattr(content, "meta", getattr(content, "_meta", None))
+                        ),
+                        blob=blob_value,
+                    )
+                )
+        return result or None
+
     async def list_resources(self) -> Sequence[ResourceEntry]:
         """List available MCP resources.
 
@@ -448,9 +551,14 @@ async def list_resources(self) -> Sequence[ResourceEntry]:
         self._resources_cache = [
             ResourceEntry(
                 uri=str(r.uri),
+                server=self.server_name,
                 name=r.title or r.name,
+                title=getattr(r, "title", "") or "",
                 description=r.description or "",
                 mime_type=r.mimeType if r.mimeType else "",
+                size=getattr(r, "size", None),
+                annotations=normalize_mcp_json_object(getattr(r, "annotations", None)),
+                meta=normalize_mcp_json_object(getattr(r, "meta", getattr(r, "_meta", None))),
             )
             for r in resources
         ]
@@ -474,33 +582,7 @@ async def read_resource(
         except Exception:
             logger.warning("Failed to read resource %r", uri, exc_info=True)
             return None
-        if not contents:
-            return None
-        result: list[TextResourceContent | BlobResourceContent] = []
-        for c in contents:
-            # MCP TextResourceContents has .text, BlobResourceContents has .blob
-            text_val: str | None = getattr(c, "text", None)
-            if text_val is not None:
-                result.append(
-                    TextResourceContent(
-                        uri=uri,
-                        mime_type=getattr(c, "mimeType", None),
-                        meta=getattr(c, "meta", None),
-                        text=text_val,
-                    )
-                )
-            else:
-                blob_val: str | None = getattr(c, "blob", None)
-                if blob_val is not None:
-                    result.append(
-                        BlobResourceContent(
-                            uri=uri,
-                            mime_type=getattr(c, "mimeType", None),
-                            meta=getattr(c, "meta", None),
-                            blob=blob_val,
-                        )
-                    )
-        return result if result else None
+        return self._convert_resource_contents(uri, contents)
 
     async def resource_exists(self, uri: str) -> bool:
         """Check if an MCP resource exists.
@@ -532,11 +614,13 @@ async def list_resource_templates(self) -> Sequence[ResourceTemplateEntry]:
         self._resource_templates_cache = [
             ResourceTemplateEntry(
                 uri_template=str(t.uriTemplate),
+                server=self.server_name,
                 name=t.name or "",
                 title=getattr(t, "title", "") or "",
                 description=t.description or "",
                 mime_type=t.mimeType if t.mimeType else "",
-                annotations=getattr(t, "annotations", None),
+                annotations=normalize_mcp_json_object(getattr(t, "annotations", None)),
+                meta=normalize_mcp_json_object(getattr(t, "meta", getattr(t, "_meta", None))),
             )
             for t in templates
         ]