diff --git a/changelog/unreleased/2026-08-20-mcp-resource-integration.md b/changelog/unreleased/2026-08-20-mcp-resource-integration.md new file mode 100644 index 000000000..709b758f6 --- /dev/null +++ b/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. diff --git a/changelog/unreleased/README.md b/changelog/unreleased/README.md new file mode 100644 index 000000000..c130d38be --- /dev/null +++ b/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)) diff --git a/src/wolfharness/agents/native_agent/agent.py b/src/wolfharness/agents/native_agent/agent.py index 2e73c6eda..bd9a4b5df 100644 --- a/src/wolfharness/agents/native_agent/agent.py +++ b/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: diff --git a/src/wolfharness/capabilities/extension_registry.py b/src/wolfharness/capabilities/extension_registry.py index 20152a0ab..cb1429b40 100644 --- a/src/wolfharness/capabilities/extension_registry.py +++ b/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,6 +128,7 @@ 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. @@ -133,8 +136,12 @@ def __init__( 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( diff --git a/src/wolfharness/capabilities/mcp_server_cap.py b/src/wolfharness/capabilities/mcp_server_cap.py index 9c173ccf3..3380eaa09 100644 --- a/src/wolfharness/capabilities/mcp_server_cap.py +++ b/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 ] diff --git a/src/wolfharness/capabilities/resource_capability.py b/src/wolfharness/capabilities/resource_capability.py index 959989670..7d18831d9 100644 --- a/src/wolfharness/capabilities/resource_capability.py +++ b/src/wolfharness/capabilities/resource_capability.py @@ -1,4 +1,4 @@ -"""ResourceCapability — unified resource access via 5 agent-facing tools. +"""ResourceCapability — unified MCP Resource access via three agent tools. Provides a single ``AbstractCapability`` that aggregates resource access across all visible ``ResourceAccess``, ``SkillResource``, and @@ -6,37 +6,44 @@ ``ExtensionRegistry``. The capability is stateless — it reads ``ctx.deps`` (an ``AgentContextDeps``) at runtime to resolve providers. -Tools exposed: - - ``list_resources``: Aggregate resources from MCP + skills - - ``read_resource``: Read content by URI (skill:// or other) - - ``resource_exists``: Check if a resource exists - - ``list_resource_templates``: List URI templates for dynamic discovery - - ``complete_resource_template``: Get completion suggestions for template params +Only the three ``list_mcp_*``/``read_mcp_resource`` methods are model-facing. +The older five methods remain below as internal compatibility helpers. """ from __future__ import annotations import asyncio +import base64 +import binascii +import json from typing import TYPE_CHECKING, Annotated import logfire from pydantic import Field -from pydantic_ai import ToolReturn +from pydantic_ai import BinaryContent, ToolReturn from pydantic_ai.capabilities import AbstractCapability from pydantic_ai.tools import AgentDepsT, RunContext from pydantic_ai.toolsets import AgentToolset, FunctionToolset from wolfharness.capabilities.extension_registry import Scope, ScopeLevel from wolfharness.capabilities.resource_protocols import ( + BlobResourceContent, CompletionArgument, CompletionResult, + McpResourceListResult, + McpResourceReadResult, + McpResourceTemplateListResult, ResourceEntry, + ResourceError, ResourceTemplateEntry, + TextResourceContent, ) if TYPE_CHECKING: from wolfharness.capabilities.agent_context import AgentContextDeps + from wolfharness.capabilities.resource_protocols import McpResourceProvider + from wolfharness.common_types import JsonObject # Number of header lines (header + separator) before data rows. @@ -45,7 +52,17 @@ # Default pagination limits. _DEFAULT_LIST_LIMIT = 50 _DEFAULT_READ_TEXT_LIMIT = 10_000 +_MAX_LIST_LIMIT = 100 +_CURSOR_VERSION = 2 _MAX_COMPLETION_SUGGESTIONS = 100 +_MAX_BLOB_BYTES = 10 * 1024 * 1024 +_SUPPORTED_BLOB_MIME_TYPES = frozenset({ + "application/pdf", + "image/gif", + "image/jpeg", + "image/png", + "image/webp", +}) # Max wall-clock time a single provider may take to answer a listing request. # Providers that time out are skipped with a warning instead of blocking the @@ -54,7 +71,7 @@ class ResourceCapability(AbstractCapability[AgentDepsT]): - """Unified resource access capability providing 5 agent-facing tools. + """Unified MCP Resource capability providing three agent-facing tools. Aggregates resources from all visible providers (MCP servers, local skills) via the ``ExtensionRegistry`` on ``AgentContextDeps``. The @@ -93,35 +110,24 @@ def get_instructions(self) -> str | None: management tools and supported URI schemes. """ return ( - "You have access to resource management tools:\n" - "- list_resources: List available resources from connected MCP " - "servers and local files (paginated, use offset to page through)\n" - "- read_resource: Read content from a resource URI (e.g. mcp://, " - "file://, viking://, optionally http(s) for servers that use web " - "URIs; web pages are usually better read with a web fetch tool)\n" - "- resource_exists: Check if a resource exists\n" - "- list_resource_templates: List URI templates for dynamic " - "resource discovery (paginated, use offset to page through)\n" - "- complete_resource_template: Get completion suggestions for " - "template parameters\n\n" - "URI schemes: mcp:// for MCP server resources, file:// for " - "file-based resources" + "Use MCP Resource tools progressively: list_mcp_resources or " + "list_mcp_resource_templates first, then read_mcp_resource with " + "the exact server and opaque URI returned by the provider. " + "Do not infer URI values from templates." ) @logfire.instrument("capability.resource_capability.get_toolset") def get_toolset(self) -> AgentToolset[AgentDepsT] | None: - """Return a ``FunctionToolset`` with all 5 resource tools. + """Return a ``FunctionToolset`` with the three MCP Resource tools. The tools access ``ctx.deps`` at runtime, which must be an ``AgentContextDeps`` with an ``extension_registry`` field. """ return FunctionToolset( [ - self.list_resources, - self.read_resource, - self.resource_exists, - self.list_resource_templates, - self.complete_resource_template, + self.list_mcp_resources, + self.list_mcp_resource_templates, + self.read_mcp_resource, ], id=self._toolset_id, ) @@ -185,6 +191,685 @@ def _extract_skill_name(uri: str) -> str: path = uri[len("skill://") :] return path.split("/")[0] if path else "" + @staticmethod + def _encode_cursor( + *, + server: str | None, + current_server: str | None, + provider_index: int, + upstream_cursor: str | None, + offset: int, + ) -> str: + payload = { + "version": _CURSOR_VERSION, + "server": server, + "current_server": current_server, + "provider_index": provider_index, + "upstream_cursor": upstream_cursor, + "offset": offset, + } + raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode() + return base64.urlsafe_b64encode(raw).decode().rstrip("=") + + @staticmethod + def _decode_cursor(cursor: str) -> dict[str, object]: + try: + padding = "=" * (-len(cursor) % 4) + value = json.loads(base64.urlsafe_b64decode((cursor + padding).encode())) + except ( + ValueError, + TypeError, + binascii.Error, + UnicodeDecodeError, + json.JSONDecodeError, + ) as exc: + raise ValueError("cursor is not a valid host cursor") from exc + if not isinstance(value, dict) or value.get("version") != _CURSOR_VERSION: + raise ValueError("cursor version is unsupported") + if not isinstance(value.get("provider_index"), int) or not isinstance( + value.get("offset"), int + ): + raise TypeError("cursor position is invalid") + if value["provider_index"] < 0 or value["offset"] < 0: + raise ValueError("cursor position is invalid") + if value.get("server") is not None and not isinstance(value.get("server"), str): + raise ValueError("cursor server is invalid") + if value.get("current_server") is not None and not isinstance( + value.get("current_server"), str + ): + raise ValueError("cursor current server is invalid") + if value.get("upstream_cursor") is not None and not isinstance( + value.get("upstream_cursor"), str + ): + raise ValueError("cursor upstream value is invalid") + return value + + @staticmethod + def _cursor_state(value: dict[str, object]) -> tuple[int, str | None, int]: + provider_index = value.get("provider_index") + offset = value.get("offset") + upstream_cursor = value.get("upstream_cursor") + if not isinstance(provider_index, int) or not isinstance(offset, int): + raise TypeError("cursor position is invalid") + if upstream_cursor is not None and not isinstance(upstream_cursor, str): + raise ValueError("cursor upstream value is invalid") + return provider_index, upstream_cursor, offset + + @staticmethod + def _resource_entry_dict(entry: ResourceEntry) -> dict[str, object]: + return { + "server": entry.server, + "uri": entry.uri, + "name": entry.name, + "title": entry.title, + "description": entry.description, + "mime_type": entry.mime_type, + "size": entry.size, + "annotations": entry.annotations, + "meta": entry.meta, + } + + @staticmethod + def _template_entry_dict(entry: ResourceTemplateEntry) -> dict[str, object]: + return { + "server": entry.server, + "uri_template": entry.uri_template, + "name": entry.name, + "title": entry.title, + "description": entry.description, + "mime_type": entry.mime_type, + "annotations": entry.annotations, + "meta": entry.meta, + } + + @staticmethod + def _error_dict(error: ResourceError) -> dict[str, object]: + return { + "code": error.code, + "message": error.message, + "retryable": error.retryable, + "suggestion": error.suggestion, + } + + @staticmethod + def _validate_limit(limit: int) -> None: + if not 1 <= limit <= _MAX_LIST_LIMIT: + raise ValueError("limit must be between 1 and 100") + + async def _mcp_providers(self, ctx: RunContext[AgentDepsT]) -> list[McpResourceProvider]: + agent_ctx = self._resolve_agent_context(ctx) + registry = agent_ctx.extension_registry + if registry is None: + return [] + from wolfharness.capabilities.resource_protocols import McpResourceProvider + + return sorted( + ( + cap + for cap in registry.get_visible_capabilities(self._make_scope(agent_ctx)) + if isinstance(cap, McpResourceProvider) + ), + key=lambda provider: provider.server_name, + ) + + @logfire.instrument("capability.resource_capability.list_mcp_resources") + async def list_mcp_resources( # noqa: PLR0915 + self, + ctx: RunContext[AgentDepsT], + server: Annotated[ + str | None, Field(description="Optional configured MCP server name") + ] = None, + cursor: Annotated[ + str | None, Field(description="Opaque cursor from the previous page") + ] = None, + limit: Annotated[ + int, + Field( + ge=1, + le=_MAX_LIST_LIMIT, + description="Number of resources to return, from 1 to 100", + ), + ] = _DEFAULT_LIST_LIMIT, + ) -> McpResourceListResult: + """List one host-aggregated page of MCP resources.""" + self._validate_limit(limit) + providers = await self._mcp_providers(ctx) + if server is not None: + providers = [provider for provider in providers if provider.server_name == server] + if not providers: + return self._resource_list_result( + "No MCP server is registered with that name.", + errors=[ + ResourceError( + "unknown_server", + f"Unknown MCP server: {server}", + False, + "Use list_mcp_resources without server to discover names.", + ) + ], + ) + provider_index = 0 + upstream_cursor: str | None = None + offset = 0 + if cursor: + try: + decoded = self._decode_cursor(cursor) + except (ValueError, TypeError) as exc: + return self._resource_list_result( + "The supplied resource cursor is invalid.", + errors=[ + ResourceError( + "invalid_cursor", + str(exc), + False, + "Restart pagination without a cursor.", + ) + ], + ) + if decoded.get("server") != server: + return self._resource_list_result( + "The cursor does not belong to the requested server.", + errors=[ + ResourceError( + "invalid_cursor", + "cursor/server mismatch", + False, + "Restart pagination with the original server argument.", + ) + ], + ) + provider_index, upstream_cursor, offset = self._cursor_state(decoded) + if provider_index >= len(providers): + return self._resource_list_result( + "The supplied resource cursor is no longer valid.", + errors=[ + ResourceError( + "invalid_cursor", + "cursor provider position is outside the current provider set", + False, + "Restart pagination without a cursor.", + ) + ], + ) + current_server = decoded.get("current_server") + if current_server != providers[provider_index].server_name: + return self._resource_list_result( + "The supplied resource cursor is no longer valid.", + errors=[ + ResourceError( + "invalid_cursor", + "cursor current server does not match provider position", + False, + "Restart pagination without a cursor.", + ) + ], + ) + + entries: list[ResourceEntry] = [] + errors: list[ResourceError] = [] + next_cursor: str | None = None + while provider_index < len(providers) and len(entries) < limit: + provider = providers[provider_index] + try: + if not await provider.supports_resources(): + errors.append( + ResourceError( + "resources_not_supported", + f"Server {provider.server_name} did not declare resources capability", + False, + "Use its tools or configure a server with resources capability.", + ) + ) + provider_index += 1 + upstream_cursor = None + offset = 0 + continue + page = await provider.list_resources_page( + upstream_cursor if isinstance(upstream_cursor, str) else None + ) + except (OSError, RuntimeError, TimeoutError, ValueError) as exc: + errors.append( + ResourceError( + "provider_unavailable", + f"Failed to list resources from {provider.server_name}: {exc}", + True, + "Retry the same request later.", + ) + ) + provider_index += 1 + upstream_cursor = None + offset = 0 + continue + available = page.entries[offset:] + take = min(limit - len(entries), len(available)) + entries.extend(available[:take]) + offset += take + if offset < len(page.entries): + next_cursor = self._encode_cursor( + server=server, + current_server=provider.server_name, + provider_index=provider_index, + upstream_cursor=upstream_cursor if isinstance(upstream_cursor, str) else None, + offset=offset, + ) + break + if page.next_cursor: + next_cursor = self._encode_cursor( + server=server, + current_server=provider.server_name, + provider_index=provider_index, + upstream_cursor=page.next_cursor, + offset=0, + ) + if len(entries) >= limit: + break + upstream_cursor = page.next_cursor + offset = 0 + continue + provider_index += 1 + upstream_cursor = None + offset = 0 + if len(entries) >= limit: + break + if next_cursor is None and provider_index < len(providers): + next_cursor = self._encode_cursor( + server=server, + current_server=providers[provider_index].server_name, + provider_index=provider_index, + upstream_cursor=upstream_cursor if isinstance(upstream_cursor, str) else None, + offset=offset, + ) + summary = f"Returned {len(entries)} MCP resource(s)" + return self._resource_list_result( + summary, resources=entries, next_cursor=next_cursor, errors=errors + ) + + @logfire.instrument("capability.resource_capability.list_mcp_resource_templates") + async def list_mcp_resource_templates( # noqa: PLR0915 + self, + ctx: RunContext[AgentDepsT], + server: Annotated[ + str | None, Field(description="Optional configured MCP server name") + ] = None, + cursor: Annotated[ + str | None, Field(description="Opaque cursor from the previous page") + ] = None, + limit: Annotated[ + int, + Field( + ge=1, + le=_MAX_LIST_LIMIT, + description="Number of templates to return, from 1 to 100", + ), + ] = _DEFAULT_LIST_LIMIT, + ) -> McpResourceTemplateListResult: + """List one host-aggregated page of MCP resource templates.""" + self._validate_limit(limit) + providers = await self._mcp_providers(ctx) + if server is not None: + providers = [provider for provider in providers if provider.server_name == server] + if not providers: + return self._template_list_result( + "No MCP server is registered with that name.", + templates=[], + errors=[ + ResourceError( + "unknown_server", + f"Unknown MCP server: {server}", + False, + "Use list_mcp_resource_templates without server to discover names.", + ) + ], + ) + provider_index = 0 + upstream_cursor = None + offset = 0 + if cursor: + try: + decoded = self._decode_cursor(cursor) + except (ValueError, TypeError) as exc: + return self._template_list_result( + "The supplied resource cursor is invalid.", + templates=[], + errors=[ + ResourceError( + "invalid_cursor", + str(exc), + False, + "Restart pagination without a cursor.", + ) + ], + ) + if decoded.get("server") != server: + return self._template_list_result( + "The cursor does not belong to the requested server.", + templates=[], + errors=[ + ResourceError( + "invalid_cursor", + "cursor/server mismatch", + False, + "Restart pagination with the original server argument.", + ) + ], + ) + provider_index, upstream_cursor, offset = self._cursor_state(decoded) + if provider_index >= len(providers): + return self._template_list_result( + "The supplied resource cursor is no longer valid.", + templates=[], + errors=[ + ResourceError( + "invalid_cursor", + "cursor provider position is outside the current provider set", + False, + "Restart pagination without a cursor.", + ) + ], + ) + current_server = decoded.get("current_server") + if current_server != providers[provider_index].server_name: + return self._template_list_result( + "The supplied resource cursor is no longer valid.", + templates=[], + errors=[ + ResourceError( + "invalid_cursor", + "cursor current server does not match provider position", + False, + "Restart pagination without a cursor.", + ) + ], + ) + entries: list[ResourceTemplateEntry] = [] + errors: list[ResourceError] = [] + next_cursor: str | None = None + while provider_index < len(providers) and len(entries) < limit: + provider = providers[provider_index] + try: + if not await provider.supports_resources(): + errors.append( + ResourceError( + "resources_not_supported", + f"Server {provider.server_name} did not declare resources capability", + False, + "Use its tools or configure a server with resources capability.", + ) + ) + provider_index += 1 + upstream_cursor = None + offset = 0 + continue + page = await provider.list_resource_templates_page( + upstream_cursor if isinstance(upstream_cursor, str) else None + ) + except (OSError, RuntimeError, TimeoutError, ValueError) as exc: + errors.append( + ResourceError( + "provider_unavailable", + f"Failed to list resource templates from {provider.server_name}: {exc}", + True, + "Retry the same request later.", + ) + ) + provider_index += 1 + upstream_cursor = None + offset = 0 + continue + available = page.entries[offset:] + take = min(limit - len(entries), len(available)) + entries.extend(available[:take]) + offset += take + if offset < len(page.entries): + next_cursor = self._encode_cursor( + server=server, + current_server=provider.server_name, + provider_index=provider_index, + upstream_cursor=upstream_cursor if isinstance(upstream_cursor, str) else None, + offset=offset, + ) + break + if page.next_cursor: + next_cursor = self._encode_cursor( + server=server, + current_server=provider.server_name, + provider_index=provider_index, + upstream_cursor=page.next_cursor, + offset=0, + ) + if len(entries) >= limit: + break + upstream_cursor = page.next_cursor + offset = 0 + continue + provider_index += 1 + upstream_cursor = None + offset = 0 + if next_cursor is None and provider_index < len(providers): + next_cursor = self._encode_cursor( + server=server, + current_server=providers[provider_index].server_name, + provider_index=provider_index, + upstream_cursor=upstream_cursor if isinstance(upstream_cursor, str) else None, + offset=offset, + ) + return self._template_list_result( + f"Returned {len(entries)} MCP resource template(s)", + templates=entries, + next_cursor=next_cursor, + errors=errors, + ) + + @staticmethod + def _resource_list_result( + summary: str, + *, + resources: list[ResourceEntry] | None = None, + next_cursor: str | None = None, + errors: list[ResourceError] | None = None, + ) -> McpResourceListResult: + return McpResourceListResult( + summary=summary, + resources=resources or [], + next_cursor=next_cursor, + errors=errors or [], + ) + + @staticmethod + def _template_list_result( + summary: str, + *, + templates: list[ResourceTemplateEntry] | None = None, + next_cursor: str | None = None, + errors: list[ResourceError] | None = None, + ) -> McpResourceTemplateListResult: + return McpResourceTemplateListResult( + summary=summary, + templates=templates or [], + next_cursor=next_cursor, + errors=errors or [], + ) + + @logfire.instrument("capability.resource_capability.read_mcp_resource") + async def read_mcp_resource( # noqa: PLR0915 + self, + ctx: RunContext[AgentDepsT], + server: Annotated[str, Field(description="Configured MCP server name")], + uri: Annotated[ + str, Field(description="Opaque resource URI copied from MCP list/search output") + ], + ) -> ToolReturn[McpResourceReadResult]: + """Read one resource from the named MCP server without URI rewriting.""" + providers = await self._mcp_providers(ctx) + provider = next((item for item in providers if item.server_name == server), None) + if provider is None: + error = ResourceError( + "unknown_server", + f"Unknown MCP server: {server}", + False, + "Use list_mcp_resources to discover server names.", + ) + return self._read_error(uri, error) + try: + if not await provider.supports_resources(): + error = ResourceError( + "resources_not_supported", + f"Server {server} did not declare resources capability", + False, + "Use the server's tools instead.", + ) + return self._read_error(uri, error) + contents = await provider.read_mcp_resource(uri) + except PermissionError as exc: + error = ResourceError( + "permission_denied", + str(exc), + False, + "Request a URI permitted by the upstream server.", + ) + return self._read_error(uri, error) + except TimeoutError as exc: + error = ResourceError("timeout", str(exc), True, "Retry the read later.") + return self._read_error(uri, error) + except (OSError, RuntimeError, ValueError) as exc: + error = ResourceError( + "provider_unavailable", + str(exc), + True, + "Retry the read or inspect the provider status.", + ) + return self._read_error(uri, error) + if not contents: + error = ResourceError( + "resource_not_found", + f"Resource not found: {uri}", + False, + "Copy the URI exactly from the server's list or search result.", + ) + return self._read_error(uri, error) + + model_contents: list[JsonObject] = [] + visible_content: list[str | BinaryContent] = [] + errors: list[ResourceError] = [] + truncated = False + original_char_count = 0 + for content in contents: + if isinstance(content, TextResourceContent): + original_char_count += len(content.text) + text = content.text + if len(text) > _DEFAULT_READ_TEXT_LIMIT: + text = text[:_DEFAULT_READ_TEXT_LIMIT] + truncated = True + content_entry = self._content_entry_with_meta( + { + "type": "text", + "uri": content.uri, + "mime_type": content.mime_type, + "text": text, + }, + content.meta, + ) + model_contents.append(content_entry) + visible_content.append(text) + elif isinstance(content, BlobResourceContent): + try: + blob = base64.b64decode(content.blob, validate=True) + except (binascii.Error, ValueError): + error = ResourceError( + "unsupported_mime_type", + "Provider returned invalid base64 content", + False, + "Ask the provider for a supported binary resource.", + ) + return self._read_error(uri, error) + mime_type = content.mime_type or "" + if mime_type not in _SUPPORTED_BLOB_MIME_TYPES: + content_entry = self._content_entry_with_meta( + { + "type": "blob", + "uri": content.uri, + "mime_type": mime_type, + "size": len(blob), + "attached": False, + "omission_reason": "unsupported_mime_type", + }, + content.meta, + ) + model_contents.append(content_entry) + errors.append( + ResourceError( + "unsupported_mime_type", + f"Binary MIME type is not supported: {mime_type or 'unknown'}", + False, + "Request a PDF, GIF, JPEG, PNG, or WebP resource.", + ) + ) + continue + if len(blob) > _MAX_BLOB_BYTES: + content_entry = self._content_entry_with_meta( + { + "type": "blob", + "uri": content.uri, + "mime_type": mime_type, + "size": len(blob), + "attached": False, + "omission_reason": "content_too_large", + }, + content.meta, + ) + model_contents.append(content_entry) + errors.append( + ResourceError( + "content_too_large", + f"Binary resource exceeds {_MAX_BLOB_BYTES} bytes", + False, + "Read a smaller resource or request metadata only.", + ) + ) + continue + content_entry = self._content_entry_with_meta( + { + "type": "blob", + "uri": content.uri, + "mime_type": mime_type, + "size": len(blob), + "attached": True, + }, + content.meta, + ) + model_contents.append(content_entry) + visible_content.append(BinaryContent(data=blob, media_type=mime_type)) + return ToolReturn( + return_value=McpResourceReadResult( + summary=f"Read MCP resource {uri} from {server}", + uri=uri, + contents=model_contents, + truncated=truncated, + original_char_count=original_char_count or None, + errors=errors, + ), + content=visible_content, + ) + + @staticmethod + def _read_error(uri: str, error: ResourceError) -> ToolReturn[McpResourceReadResult]: + result = McpResourceReadResult( + summary=error.message, + uri=uri, + errors=[error], + ) + return ToolReturn(return_value=result, content=error.message) + + @staticmethod + def _content_entry_with_meta( + entry: JsonObject, + meta: JsonObject | None, + ) -> JsonObject: + """Attach upstream content metadata without changing other fields.""" + if meta is not None: + entry["meta"] = meta + return entry + # ------------------------------------------------------------------ # Tool implementations # ------------------------------------------------------------------ @@ -231,6 +916,8 @@ async def list_resources( return_exceptions=True, ) + # Aggregate, deduplicate by URI, and identify sources. + seen_uris: set[str] = set() source_entries: list[tuple[str, ResourceEntry]] = [] for cap, result in zip(provider_caps, gathered, strict=True): if isinstance(result, BaseException): @@ -239,8 +926,23 @@ async def list_resources( source=type(cap).__name__, ) continue - source = type(cap).__name__ - source_entries.extend((source, entry) for entry in result) + # Use server_name if available, otherwise fall back to the + # first owned scheme or the class name. + source = ( + getattr(cap, "server_name", None) + or (next(iter(cap.owned_schemes)) if cap.owned_schemes else "") + or type(cap).__name__ + ) + for entry in result: + if entry.uri in seen_uris: + logfire.warning( + "Duplicate resource URI '{uri}' from {source} (skipped)", + uri=entry.uri, + source=source, + ) + continue + seen_uris.add(entry.uri) + source_entries.append((source, entry)) total = len(source_entries) page = source_entries[offset : offset + limit] @@ -301,7 +1003,9 @@ async def read_resource( resource_caps = registry.get_resource_access(scope) skill_caps = registry.get_skill_resources(scope) - content = await resolve_resource_content(uri, resource_caps, skill_caps) + content = await resolve_resource_content( + uri, resource_caps, skill_caps, scheme_registry=registry.scheme_registry + ) if content is None: return ToolReturn(return_value=f"Resource not found: {uri}") diff --git a/src/wolfharness/capabilities/resource_protocols.py b/src/wolfharness/capabilities/resource_protocols.py index 869ad8c82..0b9545154 100644 --- a/src/wolfharness/capabilities/resource_protocols.py +++ b/src/wolfharness/capabilities/resource_protocols.py @@ -19,6 +19,8 @@ from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable import warnings +from wolfharness.common_types import JsonObject, JsonValue # noqa: TC001 + if TYPE_CHECKING: from collections.abc import AsyncIterator, Sequence @@ -29,6 +31,41 @@ from wolfharness.capabilities.change_event import ChangeEvent +def resource_catalog_key(server: str, uri: str) -> str: + """Build the OpenCode-compatible escaped ``server:uri`` catalog key.""" + escaped_server = server.replace("%", "%25").replace(":", "%3A") + return f"{escaped_server}:{uri}" + + +def normalize_mcp_json_object(value: object) -> JsonObject | None: + """Convert MCP metadata models into the repository JSON value types.""" + if value is None: + return None + if hasattr(value, "model_dump"): + value = value.model_dump(mode="json", by_alias=True, exclude_none=True) + if not isinstance(value, dict): + return None + return { + key: _normalize_mcp_json_value(item) for key, item in value.items() if isinstance(key, str) + } + + +def _normalize_mcp_json_value(value: object) -> JsonValue: + if value is None or isinstance(value, (bool, int, float, str)): + return value + if hasattr(value, "model_dump"): + value = value.model_dump(mode="json", by_alias=True, exclude_none=True) + if isinstance(value, dict): + return { + key: _normalize_mcp_json_value(item) + for key, item in value.items() + if isinstance(key, str) + } + if isinstance(value, list): + return [_normalize_mcp_json_value(item) for item in value] + return str(value) + + # ---- Dataclasses ---- @@ -87,14 +124,24 @@ class ResourceEntry: Attributes: uri: Resource URI (e.g., ``"file:///path/to/resource"``). name: Human-readable resource name. + title: Optional display title. description: Optional description. mime_type: MIME type of the resource content. + size: Optional byte size reported by the provider. + annotations: Optional MCP annotations. + meta: Optional provider metadata. + server: Configured MCP server/client name. """ uri: str name: str = "" description: str = "" mime_type: str = "" + title: str = "" + size: int | None = None + annotations: JsonObject | None = None + meta: JsonObject | None = None + server: str = "" @dataclass(frozen=True, slots=True) @@ -131,7 +178,7 @@ class ResourceContent: uri: str mime_type: str | None = None - meta: dict[str, Any] | None = None + meta: JsonObject | None = None @dataclass(frozen=True, slots=True) @@ -177,6 +224,8 @@ class ResourceTemplateEntry: description: Optional description. mime_type: MIME type of expanded resources. annotations: Optional MCP annotations dict. + meta: Optional provider metadata. + server: Configured MCP server/client name. """ uri_template: str @@ -184,7 +233,67 @@ class ResourceTemplateEntry: title: str = "" description: str = "" mime_type: str = "" - annotations: dict[str, Any] | None = None + annotations: JsonObject | None = None + meta: JsonObject | None = None + server: str = "" + + +@dataclass(frozen=True, slots=True) +class ResourceError: + """Structured error returned by a resource provider.""" + + code: str + message: str + retryable: bool = False + suggestion: str = "" + + +@dataclass(frozen=True, slots=True) +class McpResourceListResult: + """One host-aggregated page of MCP resources.""" + + summary: str + resources: list[ResourceEntry] = field(default_factory=list) + next_cursor: str | None = None + errors: list[ResourceError] = field(default_factory=list) + + +@dataclass(frozen=True, slots=True) +class McpResourceTemplateListResult: + """One host-aggregated page of MCP resource templates.""" + + summary: str + templates: list[ResourceTemplateEntry] = field(default_factory=list) + next_cursor: str | None = None + errors: list[ResourceError] = field(default_factory=list) + + +@dataclass(frozen=True, slots=True) +class McpResourceReadResult: + """Structured result for a single MCP resource read.""" + + summary: str + uri: str + contents: list[JsonObject] = field(default_factory=list) + truncated: bool = False + original_char_count: int | None = None + errors: list[ResourceError] = field(default_factory=list) + + +@dataclass(frozen=True, slots=True) +class McpResourceListPage: + """Single upstream MCP resources/list page.""" + + entries: list[ResourceEntry] = field(default_factory=list) + next_cursor: str | None = None + + +@dataclass(frozen=True, slots=True) +class McpResourceTemplateListPage: + """Single upstream MCP resources/templates/list page.""" + + entries: list[ResourceTemplateEntry] = field(default_factory=list) + next_cursor: str | None = None @dataclass(frozen=True, slots=True) @@ -286,6 +395,43 @@ async def call_tool(self, name: str, args: dict[str, Any]) -> ToolResult: ... +class UriSchemeMismatchError(ValueError): + """Raised when a provider receives a URI whose scheme it does not own. + + Attributes: + scheme: The URI scheme that was not recognized. + provider_name: The name of the provider that rejected the URI. + uri: The full URI that was rejected. + """ + + def __init__(self, scheme: str, provider_name: str, uri: str) -> None: + self.scheme = scheme + self.provider_name = provider_name + self.uri = uri + super().__init__( + f"Provider '{provider_name}' does not own URI scheme '{scheme}' for URI: {uri}" + ) + + +class UriSchemeConflictError(ValueError): + """Raised when two providers claim the same URI scheme. + + Attributes: + scheme: The URI scheme with conflicting claims. + existing_provider: The provider already registered for this scheme. + conflicting_provider: The provider attempting to register. + """ + + def __init__(self, scheme: str, existing_provider: str, conflicting_provider: str) -> None: + self.scheme = scheme + self.existing_provider = existing_provider + self.conflicting_provider = conflicting_provider + super().__init__( + f"URI scheme '{scheme}' is already claimed by " + f"'{existing_provider}'; cannot register '{conflicting_provider}'" + ) + + @runtime_checkable class ResourceAccess(Protocol): """Protocol for accessing MCP resources. @@ -295,6 +441,17 @@ class ResourceAccess(Protocol): and existence checking. """ + @property + def owned_schemes(self) -> frozenset[str]: + """URI schemes this provider authoritatively handles. + + Returns: + ``frozenset`` of URI scheme strings (e.g., ``{"viking"}``). + An empty set (default) means the provider handles opaque + URIs and is consulted for unregistered schemes. + """ + return frozenset() + async def list_resources(self) -> Sequence[ResourceEntry]: """List available MCP resources. @@ -352,19 +509,41 @@ async def complete_resource_template( argument: CompletionArgument, context: dict[str, str] | None = None, ) -> CompletionResult: - """Complete a resource template parameter. + """Complete a resource template parameter.""" + ... - Args: - uri_template: The URI template to complete. - argument: The argument being completed. - context: Optional context arguments. - Returns: - ``CompletionResult`` with suggestion values. +@runtime_checkable +class McpResourceProvider(Protocol): + """Protocol for host-side MCP Resource catalog and reads.""" - Raises: - NotImplementedError: If completion is not supported. - """ + @property + def server_name(self) -> str: + """Return the stable configured MCP server identifier.""" + ... + + async def supports_resources(self) -> bool: + """Return whether the upstream server declared Resource capability.""" + ... + + async def list_resources(self) -> Sequence[ResourceEntry]: + """Read all resources for compatibility Host catalog consumers.""" + ... + + async def list_resources_page(self, cursor: str | None = None) -> McpResourceListPage: + """Read one upstream resources/list page.""" + ... + + async def list_resource_templates_page( + self, cursor: str | None = None + ) -> McpResourceTemplateListPage: + """Read one upstream resources/templates/list page.""" + ... + + async def read_mcp_resource( + self, uri: str + ) -> list[TextResourceContent | BlobResourceContent] | None: + """Read an MCP resource by opaque URI.""" ... diff --git a/src/wolfharness/capabilities/resource_resolver.py b/src/wolfharness/capabilities/resource_resolver.py index 138917fdd..e4eafb56f 100644 --- a/src/wolfharness/capabilities/resource_resolver.py +++ b/src/wolfharness/capabilities/resource_resolver.py @@ -15,6 +15,7 @@ from wolfharness.capabilities.resource_protocols import ( BlobResourceContent, TextResourceContent, + UriSchemeMismatchError, ) from wolfharness.skills.uri_resolver import ResolvedSkillURI, _name_alternatives @@ -23,6 +24,7 @@ from pydantic_ai.messages import UserContent from wolfharness.capabilities.resource_protocols import ResourceAccess, SkillResource + from wolfharness.capabilities.uri_scheme_registry import UriSchemeRegistry def _truncate_text(text: str, max_chars: int) -> str: @@ -123,6 +125,90 @@ async def _resolve_skill_reference( return None +def _filter_by_client_name( + resource_caps: list[ResourceAccess], + client_name: str, +) -> list[ResourceAccess] | None: + """Filter resource caps to only those matching ``client_name``. + + Returns a filtered list, or ``None`` if no caps matched. + """ + identified_caps = [ + resource_cap + for resource_cap in resource_caps + if getattr(resource_cap, "server_name", None) is not None + ] + if not identified_caps: + return resource_caps + selected_caps = [ + resource_cap + for resource_cap in identified_caps + if getattr(resource_cap, "server_name", None) == client_name + ] + return selected_caps or None + + +async def _resolve_skill_uri( + uri: str, + skill_caps: list[SkillResource], + max_text_chars: int, +) -> list[UserContent] | None: + """Resolve a ``skill://`` URI and return its content. + + Handles both reference files (``skill://name/path/to/file.md``) and + main skill content (``skill://name``, which reads SKILL.md). + + Args: + uri: The ``skill://`` URI to resolve. + skill_caps: List of ``SkillResource`` providers. + max_text_chars: Maximum text characters before truncation. + + Returns: + Content items if the skill was found, or ``None`` if not. + """ + if not uri.startswith("skill://"): + return None + resolved = ResolvedSkillURI.parse(uri) + skill_name = resolved.skill_name + + # If the URI contains a reference path, read the reference file. + # Exception: "SKILL.md" (case-insensitive) is the skill's main file — + # use read_skill() for backward compatibility and virtual skill support. + if resolved.reference_path is not None and resolved.reference_path.upper() != "SKILL.MD": + try: + ref_content = await _resolve_skill_reference( + skill_caps, skill_name, resolved.reference_path + ) + except Exception: # noqa: BLE001 + logfire.exception( + "Failed to read skill reference '{skill_name}/{ref}'", + skill_name=skill_name, + ref=resolved.reference_path, + ) + return None + if ref_content is not None: + truncated = _truncate_text(ref_content, max_text_chars) + return [f'\n{truncated}\n'] + return None + + # No reference path — read SKILL.md content + for skill_cap in skill_caps: + try: + content = await skill_cap.read_skill(skill_name) + except Exception: # noqa: BLE001 + logfire.exception( + "Failed to read skill '{skill_name}' from {cap}", + skill_name=skill_name, + cap=type(skill_cap).__name__, + ) + continue + if content is None: + continue + truncated = _truncate_text(content, max_text_chars) + return [f'\n{truncated}\n'] + return None + + @logfire.instrument("capability.resource_resolver.resolve") async def resolve_resource_content( uri: str, @@ -130,6 +216,8 @@ async def resolve_resource_content( skill_caps: list[SkillResource], *, max_text_chars: int = 10_000, + client_name: str | None = None, + scheme_registry: UriSchemeRegistry | None = None, ) -> list[UserContent] | None: """Resolve a resource URI and return its content as ``UserContent`` items. @@ -139,63 +227,87 @@ async def resolve_resource_content( from the skill's filesystem directory - Other URIs → ``ResourceAccess`` providers (``read_resource()``) + When ``scheme_registry`` is provided, URIs with a registered scheme are + routed directly to the authoritative provider (deterministic, O(1)). + Unregistered schemes fall back to opaque providers (empty ``owned_schemes``). + When ``scheme_registry`` is ``None``, falls back to the legacy iteration + over all providers. + Args: uri: The resource URI to resolve. resource_caps: List of ``ResourceAccess`` providers to query for non-skill URIs. skill_caps: List of ``SkillResource`` providers to query for ``skill://`` URIs. max_text_chars: Maximum text characters before truncation. + client_name: Optional exact MCP server identifier for host-injected resources. + scheme_registry: Optional ``UriSchemeRegistry`` for scheme-based routing. Returns: A list of ``UserContent`` items (strings and/or ``BinaryContent``) if the resource was found, or ``None`` if no provider could resolve the URI. """ # ---- skill:// routing ---- + skill_result = await _resolve_skill_uri(uri, skill_caps, max_text_chars) + if skill_result is not None: + return skill_result if uri.startswith("skill://"): - resolved = ResolvedSkillURI.parse(uri) - skill_name = resolved.skill_name + return None + + # ---- Base provider filtering ---- + # A connected MCP capability can still provide tools when its initialize + # handshake explicitly omitted ``resources``. Keep that server out of + # Host ResourceSource routing while leaving generic legacy providers + # (which have no negotiated state) untouched. + selected_caps = [ + resource_cap + for resource_cap in resource_caps + if getattr(resource_cap, "resources_supported", None) is not False + ] + + # ---- client_name filtering ---- + if client_name is not None: + filtered = _filter_by_client_name(selected_caps, client_name) + if filtered is None: + return None + selected_caps = filtered - # If the URI contains a reference path, read the reference file. - # Exception: "SKILL.md" (case-insensitive) is the skill's main file — - # use read_skill() for backward compatibility and virtual skill support. - if resolved.reference_path is not None and resolved.reference_path.upper() != "SKILL.MD": + # ---- Scheme-based routing (via UriSchemeRegistry) ---- + from urllib.parse import urlparse + + parsed = urlparse(uri) + scheme = parsed.scheme + + if scheme and scheme_registry is not None: + provider = scheme_registry.lookup(scheme) + if provider is not None and (client_name is None or provider in selected_caps): try: - ref_content = await _resolve_skill_reference( - skill_caps, skill_name, resolved.reference_path + contents = await provider.read_resource(uri) + except UriSchemeMismatchError: + logfire.warning( + "Provider '{name}' rejected URI '{uri}' (scheme mismatch)", + name=getattr(provider, "server_name", type(provider).__name__), + uri=uri, ) + return None except Exception: # noqa: BLE001 logfire.exception( - "Failed to read skill reference '{skill_name}/{ref}'", - skill_name=skill_name, - ref=resolved.reference_path, + "Failed to read resource '{uri}' from {cap}", + uri=uri, + cap=type(provider).__name__, ) return None - if ref_content is not None: - truncated = _truncate_text(ref_content, max_text_chars) - return [f'\n{truncated}\n'] + if contents: + return _convert_resource_parts(uri, contents, max_text_chars) return None - # No reference path — read SKILL.md content - for skill_cap in skill_caps: - try: - content = await skill_cap.read_skill(skill_name) - except Exception: # noqa: BLE001 - logfire.exception( - "Failed to read skill '{skill_name}' from {cap}", - skill_name=skill_name, - cap=type(skill_cap).__name__, - ) - continue - if content is None: - continue - truncated = _truncate_text(content, max_text_chars) - return [f'\n{truncated}\n'] - return None + # ---- Fallback for unregistered schemes ---- + # If the URI has a scheme but no registered owner, try only opaque + # providers (those with empty owned_schemes). If the URI has no scheme + # at all, try all providers (backward-compatible behavior). + if scheme: + selected_caps = [cap for cap in selected_caps if not cap.owned_schemes] - # ---- URI schemes → ResourceAccess providers ---- - # No scheme is rejected up front: MCP servers may register resources - # under any scheme, including http(s). Each provider decides whether - # it owns the URI. - for resource_cap in resource_caps: + # ---- Legacy iteration over remaining providers ---- + for resource_cap in selected_caps: try: contents = await resource_cap.read_resource(uri) except Exception: # noqa: BLE001 @@ -209,19 +321,37 @@ async def resolve_resource_content( continue if not contents: continue - - parts: list[UserContent] = [] - for c in contents: - if isinstance(c, TextResourceContent): - truncated = _truncate_text(c.text, max_text_chars) - parts.append(f'\n{truncated}\n') - elif isinstance(c, BlobResourceContent): - decoded = base64.b64decode(c.blob) - media_type = c.mime_type or "application/octet-stream" - parts.append(f'\n') - parts.append(BinaryContent(data=decoded, media_type=media_type)) - parts.append("\n") + parts = _convert_resource_parts(uri, contents, max_text_chars) if parts: return parts return None + + +def _convert_resource_parts( + uri: str, + contents: list[TextResourceContent | BlobResourceContent], + max_text_chars: int, +) -> list[UserContent] | None: + """Convert resource content blocks to ``UserContent`` items. + + Args: + uri: The resource URI (for attribution in text wrappers). + contents: List of resource content blocks. + max_text_chars: Maximum text characters before truncation. + + Returns: + A list of ``UserContent`` items, or ``None`` if empty. + """ + parts: list[UserContent] = [] + for c in contents: + if isinstance(c, TextResourceContent): + truncated = _truncate_text(c.text, max_text_chars) + parts.append(f'\n{truncated}\n') + elif isinstance(c, BlobResourceContent): + decoded = base64.b64decode(c.blob) + media_type = c.mime_type or "application/octet-stream" + parts.append(f'\n') + parts.append(BinaryContent(data=decoded, media_type=media_type)) + parts.append("\n") + return parts or None diff --git a/src/wolfharness/capabilities/skill_manager_cap.py b/src/wolfharness/capabilities/skill_manager_cap.py index e817232f2..7b78babca 100644 --- a/src/wolfharness/capabilities/skill_manager_cap.py +++ b/src/wolfharness/capabilities/skill_manager_cap.py @@ -228,6 +228,17 @@ class SkillManagerCap( _inject_mode: Instruction injection mode (description/matcher/all). """ + @property + def owned_schemes(self) -> frozenset[str]: + """URI schemes this provider authoritatively handles. + + SkillManagerCap owns the ``skill`` URI scheme exclusively. + + Returns: + ``frozenset({"skill"})``. + """ + return frozenset({"skill"}) + def __init__( self, local_skills: dict[str, Skill] | None = None, diff --git a/src/wolfharness/capabilities/uri_scheme_registry.py b/src/wolfharness/capabilities/uri_scheme_registry.py new file mode 100644 index 000000000..1b7ca1b65 --- /dev/null +++ b/src/wolfharness/capabilities/uri_scheme_registry.py @@ -0,0 +1,120 @@ +"""UriSchemeRegistry — central authority for URI scheme ownership. + +Maps each URI scheme to exactly one resource provider, enabling +deterministic scheme-based routing and conflict detection at +registration time. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from collections.abc import Sequence + +from wolfharness.capabilities.resource_protocols import ( + UriSchemeConflictError, +) + + +if TYPE_CHECKING: + from wolfharness.capabilities.resource_protocols import ResourceAccess + + +class UriSchemeRegistry: + """Maps URI schemes to authorized resource providers. + + Every scheme is owned by at most one provider. Registration + detects conflicting claims and raises ``UriSchemeConflictError``. + Lookup returns the registered provider for a scheme, or ``None`` + for unregistered schemes. + + Providers with no owned schemes (opaque passthrough) are not + registered here — they are collected separately and consulted + when no scheme owner is found. + """ + + def __init__(self) -> None: + self._scheme_to_provider: dict[str, ResourceAccess] = {} + self._scheme_to_name: dict[str, str] = {} + + def register( + self, + provider_name: str, + schemes: frozenset[str], + provider: ResourceAccess, + ) -> None: + """Register a provider as the authoritative owner of URI schemes. + + Args: + provider_name: Human-readable name of the provider. + schemes: Set of URI scheme strings this provider owns. + provider: The ``ResourceAccess`` instance to route to. + + Raises: + UriSchemeConflictError: If any scheme is already claimed. + """ + for scheme in schemes: + existing = self._scheme_to_name.get(scheme) + if existing is not None: + raise UriSchemeConflictError( + scheme=scheme, + existing_provider=existing, + conflicting_provider=provider_name, + ) + for scheme in schemes: + self._scheme_to_provider[scheme] = provider + self._scheme_to_name[scheme] = provider_name + + def lookup(self, scheme: str) -> ResourceAccess | None: + """Return the provider authorized for a URI scheme. + + Args: + scheme: The URI scheme to look up (e.g. ``"viking"``). + + Returns: + The ``ResourceAccess`` provider for that scheme, or + ``None`` if no provider is registered for the scheme. + """ + return self._scheme_to_provider.get(scheme) + + def registered_schemes(self) -> frozenset[str]: + """Return all registered URI schemes. + + Returns: + ``frozenset`` of scheme strings. + """ + return frozenset(self._scheme_to_provider) + + def owner_of(self, scheme: str) -> str | None: + """Return the provider name registered for a URI scheme. + + Args: + scheme: The URI scheme to query. + + Returns: + The provider name, or ``None`` if unregistered. + """ + return self._scheme_to_name.get(scheme) + + def unregister(self, provider: ResourceAccess) -> None: + """Remove all schemes owned by a provider. + + Args: + provider: The provider to unregister. + """ + schemes_to_remove = [ + scheme for scheme, p in self._scheme_to_provider.items() if p is provider + ] + for scheme in schemes_to_remove: + del self._scheme_to_provider[scheme] + del self._scheme_to_name[scheme] + + def registered_providers(self) -> Sequence[ResourceAccess]: + """Return all registered providers (deduplicated). + + Returns: + Sequence of unique ``ResourceAccess`` instances. + """ + return list(dict.fromkeys(self._scheme_to_provider.values())) diff --git a/src/wolfharness/capabilities/viking/__init__.py b/src/wolfharness/capabilities/viking/__init__.py index fdb7f1441..16c306518 100644 --- a/src/wolfharness/capabilities/viking/__init__.py +++ b/src/wolfharness/capabilities/viking/__init__.py @@ -91,6 +91,17 @@ class VikingCapability(AbstractCapability[Any]): public_download_base_url: Base URL for public download links. """ + @property + def owned_schemes(self) -> frozenset[str]: + """URI schemes this provider authoritatively handles. + + VikingCapability owns the ``viking`` URI scheme exclusively. + + Returns: + ``frozenset({"viking"})``. + """ + return frozenset({"viking"}) + mode: Literal["retrieve", "write", "graph", "all"] = "all" url: str | None = None api_key: str | None = None @@ -431,6 +442,28 @@ def _resolve_skills_uri(self) -> str: user_id = self._identity.user_id if self._identity is not None else (self.user or "default") return f"viking://user/{user_id}/skills/" + def _check_uri_scheme(self, uri: str) -> None: + """Raise ``UriSchemeMismatchError`` if ``uri`` is not a ``viking://`` URI. + + This is defense-in-depth: even if a caller bypasses the + ``UriSchemeRegistry``, this provider rejects URIs it does not own. + + Args: + uri: The URI to validate. + + Raises: + UriSchemeMismatchError: If the URI scheme is not ``viking``. + """ + from wolfharness.capabilities.resource_protocols import UriSchemeMismatchError + + if not uri.startswith("viking://"): + scheme = uri.split(":", maxsplit=1)[0] if ":" in uri else "" + raise UriSchemeMismatchError( + scheme=scheme, + provider_name="VikingCapability", + uri=uri, + ) + def _check_uri_allowed(self, uri: str, *, tool_name: str = "") -> str | None: """Return an error message if ``uri`` is outside the allowed prefixes. diff --git a/src/wolfharness/common_types.py b/src/wolfharness/common_types.py index d4c5d417f..6bd870eca 100644 --- a/src/wolfharness/common_types.py +++ b/src/wolfharness/common_types.py @@ -36,18 +36,22 @@ # Import path string for dynamic tool loading (e.g., "mymodule:my_tool") type ImportPathString = str type ToolType = ImportPathString | AnyCallable | Tool - # Define what we consider JSON-serializable - type JsonPrimitive = bool | int | float | str | None type SessionIdType = str | UUID | None type ProcessorCallback[TResult] = Callable[..., TResult | Awaitable[TResult]] +# Define JSON aliases at runtime as well as for static type checking. The +# MCP Resource result models are exposed through pydantic-ai tool schemas, so +# leaving these aliases inside ``TYPE_CHECKING`` would make their annotations +# unresolved at runtime. +type JsonPrimitive = bool | int | float | str | None +type JsonValue = JsonPrimitive | JsonArray | JsonObject +type JsonObject = dict[str, JsonValue] +type JsonArray = list[JsonValue] + # In reflex for example, the complex ones create issues.. SimpleJsonType = dict[ str, bool | int | float | str | list[str] | dict[str, bool | int | float | str] ] -type JsonValue = JsonPrimitive | JsonArray | JsonObject -type JsonObject = dict[str, JsonValue] -type JsonArray = list[JsonValue] MCPConnectionStatus = Literal[ diff --git a/src/wolfharness/delegation/pool.py b/src/wolfharness/delegation/pool.py index 2de9e73ed..f42e6ea58 100644 --- a/src/wolfharness/delegation/pool.py +++ b/src/wolfharness/delegation/pool.py @@ -662,8 +662,10 @@ async def _rebuild_skill_capabilities(self) -> None: # Unregister and close old SkillManagerCap from ExtensionRegistry if present. pool_scope = Scope(level=ScopeLevel.POOL) + scheme_registry = self._extension_registry.scheme_registry for existing_cap in list(self._skill_capabilities): if isinstance(existing_cap, SkillManagerCap): + scheme_registry.unregister(existing_cap) self._extension_registry.unregister(existing_cap, pool_scope) # Close child McpServerCap instances to release MCP connections. try: @@ -685,12 +687,27 @@ async def _rebuild_skill_capabilities(self) -> None: # Register the new SkillManagerCap with ExtensionRegistry at POOL scope. self._extension_registry.register(cap, pool_scope) + # Register SkillManagerCap's owned schemes in the URI scheme registry. + if cap.owned_schemes: + scheme_registry.register( + provider_name="SkillManagerCap", + schemes=cap.owned_schemes, + provider=cap, + ) + # Register each top-level McpServerCap independently at POOL scope so # they are directly discoverable via get_resource_access() for ``@`` # mention and ResourceCapability (RFC-0058). They no longer live only # inside SkillManagerCap.children. for provider in self.mcp.providers: self._extension_registry.register(provider, pool_scope) + # Register provider's owned schemes in the URI scheme registry. + if provider.owned_schemes: + scheme_registry.register( + provider_name=getattr(provider, "server_name", type(provider).__name__), + schemes=provider.owned_schemes, + provider=provider, + ) logger.debug( "Rebuilt skill capabilities", @@ -720,10 +737,12 @@ def resource_capability(self) -> Any: async def _setup_resource_capability(self) -> None: """Create the ``ResourceCapability`` instance. - The capability provides 5 agent-facing tools (``list_resources``, - ``read_resource``, ``resource_exists``, ``list_resource_templates``, - ``complete_resource_template``) that aggregate resource access - across all visible providers in the ``ExtensionRegistry``. + The capability provides three model-facing MCP Resource tools + (``list_mcp_resources``, ``list_mcp_resource_templates``, and + ``read_mcp_resource``) that aggregate resource access across all + visible MCP providers in the ``ExtensionRegistry``. Legacy five-method + Python helpers remain available for internal compatibility but are not + registered in the model toolset. The capability is stateless — it reads ``AgentContext`` at runtime to resolve providers. Per-agent opt-out is handled in diff --git a/src/wolfharness/host/factory.py b/src/wolfharness/host/factory.py index 624d90325..f6056dcab 100644 --- a/src/wolfharness/host/factory.py +++ b/src/wolfharness/host/factory.py @@ -190,6 +190,15 @@ def register_config_capabilities( from wolfharness.models.agents import NativeAgentConfig from wolfharness_config.capabilities import build_config_capabilities + # MCP connections created with the pool have POOL lifetime. Register + # their Resource providers independently of the model-tool feature gate + # so Host catalogs and ResourceSource injection remain available when + # resources.enabled is false. + pool_scope = Scope(level=ScopeLevel.POOL) + for provider in host_context.mcp.get_mcp_providers(): + if provider.resources_supported is not False: + self._pool.extension_registry.register(provider, pool_scope) + for agent_name, cfg in manifest.agents.items(): if not isinstance(cfg, NativeAgentConfig) or not cfg.capabilities: continue @@ -197,6 +206,15 @@ def register_config_capabilities( agent_scope = Scope(level=ScopeLevel.AGENT, agent_name=agent_name) for cap in config_caps: self._pool.extension_registry.register(cap, agent_scope) + # Register config-defined capability's owned schemes in the + # URI scheme registry (e.g. VikingCapability owns "viking"). + owned: frozenset[str] = getattr(cap, "owned_schemes", frozenset()) + if owned: + self._pool.extension_registry.scheme_registry.register( + provider_name=type(cap).__name__, + schemes=owned, + provider=cap, + ) @staticmethod def _build_agent_descriptions( diff --git a/src/wolfharness/mcp_server/client.py b/src/wolfharness/mcp_server/client.py index 5820f7d48..2d6e114a1 100644 --- a/src/wolfharness/mcp_server/client.py +++ b/src/wolfharness/mcp_server/client.py @@ -20,6 +20,13 @@ from schemez import FunctionSchema from wolfharness.agents.context import AgentContext +from wolfharness.capabilities.resource_protocols import ( + McpResourceListPage, + McpResourceTemplateListPage, + ResourceEntry, + ResourceTemplateEntry, + normalize_mcp_json_object, +) from wolfharness.log import get_logger from wolfharness.mcp_server.constants import MCP_TO_LOGGING from wolfharness.mcp_server.helpers import extract_text_content, mcp_tool_to_fn_schema @@ -155,6 +162,14 @@ def _get_message_handler(self) -> MessageHandlerT | MessageHandler: ) return self._wolfharness_message_handler + @property + def client_name(self) -> str: + """Return the configured display name used for Resource identity.""" + 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_info(self) -> dict[str, str] | None: """Get server info (name and version) from the connected client. @@ -421,6 +436,60 @@ async def list_tools(self) -> list[MCPTool]: else: return filtered + async def supports_resources(self) -> bool: + """Return whether the connected server declared MCP Resource support.""" + self._ensure_connected() + return self._has_server_capability("resources") + + async def list_resources_mcp(self, cursor: str | None = None) -> McpResourceListPage: + """Read one raw MCP resources/list page without auto-pagination.""" + self._ensure_connected() + if not self._has_server_capability("resources"): + return McpResourceListPage() + result = await self._client.list_resources_mcp(cursor=cursor) + entries = [ + ResourceEntry( + uri=str(resource.uri), + server=self.client_name, + name=resource.name or "", + title=getattr(resource, "title", "") or "", + description=resource.description or "", + mime_type=getattr(resource, "mimeType", "") or "", + size=getattr(resource, "size", None), + annotations=normalize_mcp_json_object(getattr(resource, "annotations", None)), + meta=normalize_mcp_json_object( + getattr(resource, "meta", getattr(resource, "_meta", None)) + ), + ) + for resource in result.resources + ] + return McpResourceListPage(entries=entries, next_cursor=result.nextCursor) + + async def list_resource_templates_mcp( + self, cursor: str | None = None + ) -> McpResourceTemplateListPage: + """Read one raw MCP resources/templates/list page.""" + self._ensure_connected() + if not self._has_server_capability("resources"): + return McpResourceTemplateListPage() + result = await self._client.list_resource_templates_mcp(cursor=cursor) + entries = [ + ResourceTemplateEntry( + uri_template=str(template.uriTemplate), + server=self.client_name, + name=template.name or "", + title=getattr(template, "title", "") or "", + description=template.description or "", + mime_type=getattr(template, "mimeType", "") or "", + annotations=normalize_mcp_json_object(getattr(template, "annotations", None)), + meta=normalize_mcp_json_object( + getattr(template, "meta", getattr(template, "_meta", None)) + ), + ) + for template in result.resourceTemplates + ] + return McpResourceTemplateListPage(entries=entries, next_cursor=result.nextCursor) + async def list_prompts(self) -> list[MCPPrompt]: """Get available prompts from the server.""" self._ensure_connected() @@ -475,6 +544,8 @@ async def read_resource(self, uri: str) -> list[TextResourceContents | BlobResou self._ensure_connected() try: return await self._client.read_resource(uri) + except (OSError, PermissionError, TimeoutError): + raise except Exception as e: raise RuntimeError(f"Failed to read resource {uri!r}: {e}") from e diff --git a/src/wolfharness/mcp_server/manager.py b/src/wolfharness/mcp_server/manager.py index a325c2994..e44b3cec7 100644 --- a/src/wolfharness/mcp_server/manager.py +++ b/src/wolfharness/mcp_server/manager.py @@ -438,6 +438,17 @@ async def setup_server( tool_prefix=candidate, ) provider = await self.exit_stack.enter_async_context(provider) + # Negotiate the Resource capability while the manager-owned + # client is connected. A tools-only server remains available to + # the MCP tool path, but is excluded from the Resource registry. + try: + await provider.supports_resources() + except (OSError, RuntimeError, TimeoutError, ValueError): + logger.debug( + "MCP Resource capability negotiation failed", + client_id=config.client_id, + exc_info=True, + ) self.providers.append(provider) except Exception as e: # Record the failure so get_server_status() can report it diff --git a/src/wolfharness_config/nodes.py b/src/wolfharness_config/nodes.py index 29458b9e1..eea4edfae 100644 --- a/src/wolfharness_config/nodes.py +++ b/src/wolfharness_config/nodes.py @@ -199,13 +199,13 @@ def get_mcp_servers(self) -> list[MCPServerConfig]: class ResourceConfig(Schema): """Configuration for resource access tools. - Controls whether the ``ResourceCapability`` (unified resource access - via ``list_resources``, ``read_resource``, ``resource_exists``, - ``list_resource_templates``, ``complete_resource_template``) is - automatically attached to the agent. + Controls whether the ``ResourceCapability`` (the three model-facing MCP + Resource tools ``list_mcp_resources``, ``list_mcp_resource_templates``, + and ``read_mcp_resource``) is automatically attached to the agent. Attributes: - enabled: When ``True`` (default), the resource tools are available. + enabled: When ``True`` (default), the three model-facing MCP Resource + tools are available. Set to ``False`` to opt out. Example: @@ -358,9 +358,8 @@ class BaseAgentConfig(NodeConfig): """Configuration for unified resource access tools. When ``enabled`` is ``True`` (default), the ``ResourceCapability`` - providing ``list_resources``, ``read_resource``, ``resource_exists``, - ``list_resource_templates``, and ``complete_resource_template`` tools - is automatically attached to the agent. + providing ``list_mcp_resources``, ``list_mcp_resource_templates``, and + ``read_mcp_resource`` is automatically attached to the agent. Set ``enabled: false`` to opt out: diff --git a/src/wolfharness_server/opencode_server/ENDPOINTS.md b/src/wolfharness_server/opencode_server/ENDPOINTS.md index cd2b8bfd7..81fb63aec 100644 --- a/src/wolfharness_server/opencode_server/ENDPOINTS.md +++ b/src/wolfharness_server/opencode_server/ENDPOINTS.md @@ -171,6 +171,29 @@ Last audited against OpenCode source: **2026-02-24** | [x] | GET | `/experimental/resource` | List MCP resources from connected servers | | [x] | GET | `/experimental/session` | List sessions globally (cross-project, paginated) | +### MCP Resource compatibility + +`GET /experimental/resource` returns a record of resources only. Each key is +the escaped `{client_name}:{uri}` pair (`%` is escaped before `:`), and each +value keeps the upstream `name`, `uri`, `description`, `mimeType`, and +`client`. Providers are selected by the current `agent_name` + `session_id` +scope; a provider failure does not discard resources returned by other +servers. + +There is intentionally no HTTP resource-template endpoint. Resource templates +are enumerated through the model-facing +`list_mcp_resource_templates(server, cursor, limit)` tool. The model-facing +resource surface contains exactly three tools: + +- `list_mcp_resources(server=None, cursor=None, limit=50)` +- `list_mcp_resource_templates(server=None, cursor=None, limit=50)` +- `read_mcp_resource(server, uri)` + +The legacy Python resource helpers remain available to Host internals but are +not registered in the model toolset. `resources.enabled=false` hides only the +three model tools; Host catalog lookup and `ResourceSource` injection continue +to use the provider registry. + --- ## Worktrees (Experimental) diff --git a/src/wolfharness_server/opencode_server/converters.py b/src/wolfharness_server/opencode_server/converters.py index 86a2a8e23..7de0c04d5 100644 --- a/src/wolfharness_server/opencode_server/converters.py +++ b/src/wolfharness_server/opencode_server/converters.py @@ -160,10 +160,20 @@ async def _resolve_resource( resource_caps = registry.get_resource_access(scope) skill_caps = registry.get_skill_resources(scope) - content = await resolve_resource_content(source.uri, resource_caps, skill_caps) + content = await resolve_resource_content( + source.uri, + resource_caps, + skill_caps, + client_name=source.client_name, + ) if content is None: logger.warning("Resource not found", client_name=source.client_name, uri=source.uri) - return content + return None + + # Prepend the server source so the agent knows which MCP server to use + # for follow-up operations on URIs in this resource content. + header = f"[Resource from {source.client_name} server] {source.uri}" + return [header, *content] async def extract_user_prompt_from_parts( diff --git a/src/wolfharness_server/opencode_server/routes/agent_routes.py b/src/wolfharness_server/opencode_server/routes/agent_routes.py index bf7cd316e..81c775cfa 100644 --- a/src/wolfharness_server/opencode_server/routes/agent_routes.py +++ b/src/wolfharness_server/opencode_server/routes/agent_routes.py @@ -51,6 +51,17 @@ logger = get_logger(__name__) +def _resource_tools_enabled(state: Any) -> bool: + """Return the configured model-resource-tool gate for the active agent.""" + host_context = state.agent.host_context + if host_context is None: + return False + from wolfharness.models.agents import NativeAgentConfig + + config = host_context.manifest.agents.get(state.agent.name) + return isinstance(config, NativeAgentConfig) and config.resources.enabled + + def _extract_hints(template: str | None) -> list[str]: """Extract input hints from a command template. @@ -492,8 +503,8 @@ async def get_console_state() -> dict[str, Any]: async def list_mcp_resources(state: StateDep) -> dict[str, McpResource]: """Get all available MCP resources from connected servers. - Returns a dictionary mapping resource keys to McpResource objects. - Keys are formatted as "{client}:{resource_name}" for uniqueness. + Returns a dictionary mapping escaped ``server:uri`` keys to McpResource + objects. The configured server/client name is preserved in each entry. Uses the ``ExtensionRegistry`` to discover ``ResourceAccess`` providers at SESSION scope (POOL + AGENT + SESSION). @@ -503,7 +514,10 @@ async def list_mcp_resources(state: StateDep) -> dict[str, McpResource]: import asyncio from wolfharness.capabilities.extension_registry import Scope, ScopeLevel - from wolfharness.capabilities.resource_protocols import ResourceAccess + from wolfharness.capabilities.resource_protocols import ( + McpResourceProvider, + resource_catalog_key, + ) agent = state.agent host_ctx = agent.host_context @@ -518,11 +532,29 @@ async def list_mcp_resources(state: StateDep) -> dict[str, McpResource]: ) else: scope = Scope(level=ScopeLevel.AGENT, agent_name=agent.name) - resource_caps = registry.get_resource_access(scope) + resource_caps = registry.get_mcp_resource_providers(scope) else: caps = agent._all_capabilities - resource_caps = [cap for cap in caps if isinstance(cap, ResourceAccess)] + resource_caps = [cap for cap in caps if isinstance(cap, McpResourceProvider)] + if resource_caps: + candidates = resource_caps + support_results = await asyncio.gather( + *(cap.supports_resources() for cap in candidates), + return_exceptions=True, + ) + for cap, supported in zip(candidates, support_results, strict=False): + if isinstance(supported, BaseException): + logger.warning( + "Failed to negotiate MCP resource capability", + server=cap.server_name, + error=str(supported), + ) + resource_caps = [ + cap + for cap, supported in zip(candidates, support_results, strict=False) + if supported is True + ] if resource_caps: results = await asyncio.gather( *(cap.list_resources() for cap in resource_caps), @@ -530,21 +562,22 @@ async def list_mcp_resources(state: StateDep) -> dict[str, McpResource]: ) for cap, res in zip(resource_caps, results, strict=False): if isinstance(res, BaseException): + logger.warning( + "Failed to list MCP resources", + server=cap.server_name, + error=str(res), + ) continue + client = cap.server_name for resource in res: - # ResourceEntry doesn't have a .client field; - # use the capability class name as the client identifier. - client = type(cap).__name__ - client_name = client.replace("/", "_") - resource_name = resource.name.replace("/", "_") - result[f"{client_name}:{resource_name}"] = McpResource( - name=resource.uri, + result[resource_catalog_key(client, resource.uri)] = McpResource( + name=resource.name, uri=resource.uri, description=resource.description, mime_type=resource.mime_type, client=client, ) - except Exception: # noqa: BLE001 + except (OSError, RuntimeError, TimeoutError, ValueError): return {} else: return result @@ -757,9 +790,26 @@ async def list_tool_ids(state: StateDep) -> list[str]: """ try: tools = await state.agent._get_all_tools() - return [tool.name for tool in tools] + tool_ids = [tool.name for tool in tools] + # ResourceCapability is attached to the per-run toolset, rather than + # the shared provider list used by ``_get_all_tools``. Include its + # formal names here so OpenCode clients can discover them before the + # first model run, while still honoring the per-agent gate. + pool = state.pool_or_none + if ( + pool is not None + and _resource_tools_enabled(state) + and pool.resource_capability is not None + ): + resource_tools = pool.resource_capability.get_toolset() + if resource_tools is not None: + for name in resource_tools.tools: + if name not in tool_ids: + tool_ids.append(name) except Exception: # noqa: BLE001 return [] + else: + return tool_ids class ToolListItem(BaseModel): @@ -796,6 +846,24 @@ async def list_tools_with_schemas( # noqa: D417 params = tool.schema["function"]["parameters"] item = ToolListItem(id=tool.name, description=tool.description or "", parameters=params) result.append(item) + pool = state.pool_or_none + if ( + pool is not None + and _resource_tools_enabled(state) + and pool.resource_capability is not None + ): + resource_tools = pool.resource_capability.get_toolset() + if resource_tools is not None: + known_ids = {item.id for item in result} + for name, tool in resource_tools.tools.items(): + if name not in known_ids: + result.append( + ToolListItem( + id=name, + description=tool.function_schema.description or "", + parameters=tool.function_schema.json_schema, + ) + ) except Exception: # noqa: BLE001 return [] else: diff --git a/tests/capabilities/test_extension_registry.py b/tests/capabilities/test_extension_registry.py index d044d518e..743f894a9 100644 --- a/tests/capabilities/test_extension_registry.py +++ b/tests/capabilities/test_extension_registry.py @@ -71,6 +71,10 @@ class FakeMcpResource(FakeCapability): of ``str | None``. """ + @property + def owned_schemes(self) -> frozenset[str]: + return frozenset() + def __init__(self, name: str = "mcp_cap", resources: dict[str, str] | None = None) -> None: super().__init__(name) self._resources = resources or {"mcp://server/path": "resource content"} diff --git a/tests/capabilities/test_mcp_server_cap.py b/tests/capabilities/test_mcp_server_cap.py index bc5280f23..3af125d41 100644 --- a/tests/capabilities/test_mcp_server_cap.py +++ b/tests/capabilities/test_mcp_server_cap.py @@ -30,6 +30,8 @@ CompletionArgument, CompletionResult, McpResource, + McpResourceListPage, + McpResourceTemplateListPage, ResourceAccess, ResourceTemplateAccess, ResourceTemplateEntry, @@ -68,6 +70,7 @@ class FakeMCPClient: _prompt_change_callback: Any = None _subscribed_uris: list[str] = field(default_factory=list) _unsubscribed_uris: list[str] = field(default_factory=list) + _supports_resources: bool = True config: Any = None async def __aenter__(self) -> Self: @@ -87,11 +90,49 @@ async def list_resources(self) -> list[Any]: raise RuntimeError("Not connected") return list(self._resources) + async def supports_resources(self) -> bool: + return self._supports_resources + + async def list_resources_mcp(self, cursor: str | None = None) -> McpResourceListPage: + del cursor + from wolfharness.capabilities.resource_protocols import ResourceEntry + + return McpResourceListPage( + entries=[ + ResourceEntry( + uri=str(resource.uri), + server=self.config.client_id if self.config is not None else "", + name=resource.name, + description=resource.description, + mime_type=resource.mimeType, + ) + for resource in self._resources + ] + ) + async def list_resource_templates(self) -> list[Any]: if not self._connected: raise RuntimeError("Not connected") return list(self._resource_templates) + async def list_resource_templates_mcp( + self, cursor: str | None = None + ) -> McpResourceTemplateListPage: + del cursor + from wolfharness.capabilities.resource_protocols import ResourceTemplateEntry + + return McpResourceTemplateListPage( + entries=[ + ResourceTemplateEntry( + uri_template=str(template.uriTemplate), + server=self.config.client_id if self.config is not None else "", + name=template.name, + description=template.description, + ) + for template in self._resource_templates + ] + ) + async def read_resource(self, uri: str) -> list[Any]: if not self._connected: raise RuntimeError("Not connected") @@ -163,6 +204,7 @@ def __init__(self, client: FakeMCPClient) -> None: async def get_client(self, config: Any, skill_name: str | None = None) -> FakeMCPClient: self.get_client_call_count += 1 self._client._connected = True + self._client.config = config return self._client @@ -185,6 +227,8 @@ def _make_resource( res.title = None res.description = description res.mimeType = mime_type + res.meta = None + res.annotations = None return res @@ -223,6 +267,7 @@ def _make_resource_template( tmpl.description = description tmpl.mimeType = mime_type tmpl.annotations = None + tmpl.meta = None return tmpl @@ -472,15 +517,49 @@ async def test_list_resources_delegation() -> None: _make_resource("file:///path1", "res1", "Resource 1", "text/plain"), _make_resource("file:///path2", "res2"), ] + resources[0].title = "Resource One" + resources[0].size = 42 + resources[0].annotations = {"audience": ["user"]} + resources[0].meta = {"origin": "upstream"} client = FakeMCPClient(_resources=resources) cap = McpServerCap(config=_make_config(), session_pool=FakeSessionPool(client)) result = await cap.list_resources() assert len(result) == 2 assert result[0].uri == "file:///path1" - assert result[0].name == "res1" + # Title is preferred as the display name (RFC-0058 / PR #372 behavior); + # the raw title is additionally preserved in its own field. + assert result[0].name == "Resource One" assert result[0].description == "Resource 1" assert result[0].mime_type == "text/plain" + assert result[0].title == "Resource One" + assert result[0].size == 42 + assert result[0].annotations == {"audience": ["user"]} + assert result[0].meta == {"origin": "upstream"} + + +@pytest.mark.anyio +async def test_paged_resource_contract_preserves_server_and_cursor() -> None: + resources = [_make_resource("file:///path1", "res1", "Resource 1", "text/plain")] + client = FakeMCPClient(_resources=resources) + cap = McpServerCap(config=_make_config(), session_pool=FakeSessionPool(client)) + + assert await cap.supports_resources() is True + page = await cap.list_resources_page() + assert page.entries[0].server == cap.server_name + assert page.entries[0].uri == "file:///path1" + templates = await cap.list_resource_templates_page() + assert templates.entries == [] + + +@pytest.mark.anyio +async def test_resource_capability_negotiation_is_cached() -> None: + client = FakeMCPClient(_supports_resources=False) + cap = McpServerCap(config=_make_config(), session_pool=FakeSessionPool(client)) + + assert cap.client_name == cap.server_name == _make_config().client_id + assert await cap.supports_resources() is False + assert cap.resources_supported is False class FlakySessionPool(FakeSessionPool): diff --git a/tests/capabilities/test_resource_capability.py b/tests/capabilities/test_resource_capability.py index 0ce45b9a1..cb19171fb 100644 --- a/tests/capabilities/test_resource_capability.py +++ b/tests/capabilities/test_resource_capability.py @@ -1,4 +1,4 @@ -"""Tests for ResourceCapability — unified resource access via 5 agent-facing tools.""" +"""Tests for ResourceCapability — unified MCP Resource access.""" from __future__ import annotations @@ -18,6 +18,8 @@ BlobResourceContent, CompletionArgument, CompletionResult, + McpResourceListPage, + McpResourceTemplateListPage, ResourceEntry, ResourceTemplateEntry, SkillEntry, @@ -41,6 +43,10 @@ class FakeResourceAccess: """Minimal ResourceAccess implementation for testing.""" + @property + def owned_schemes(self) -> frozenset[str]: + return frozenset() + def __init__( self, *, @@ -126,6 +132,76 @@ async def complete_resource_template( return CompletionResult(values=[]) +class FakeMcpResourceProvider: + """Paged MCP Resource provider used by the formal three-tool tests.""" + + def __init__( + self, + server_name: str, + *, + supported: bool = True, + read_contents: list[TextResourceContent | BlobResourceContent] | None = None, + ) -> None: + self._server_name = server_name + self.supported = supported + self.read_contents = read_contents + self.resource_pages: dict[str | None, McpResourceListPage] = { + None: McpResourceListPage( + entries=[ + ResourceEntry( + uri="kb:///resources/catalog", + server=server_name, + name="catalog", + title="Catalog", + size=12, + meta={"source": server_name}, + ) + ], + next_cursor=None, + ) + } + + @property + def server_name(self) -> str: + return self._server_name + + @property + def resources_supported(self) -> bool: + return self.supported + + async def supports_resources(self) -> bool: + return self.supported + + async def list_resources_page(self, cursor: str | None = None) -> McpResourceListPage: + return self.resource_pages.get(cursor, McpResourceListPage()) + + async def list_resources(self) -> list[ResourceEntry]: + return self.resource_pages[None].entries + + async def list_resource_templates_page( + self, cursor: str | None = None + ) -> McpResourceTemplateListPage: + return McpResourceTemplateListPage( + entries=[ + ResourceTemplateEntry(uri_template="kb:///resources/{id}", server=self.server_name) + ] + ) + + async def read_mcp_resource( + self, uri: str + ) -> list[TextResourceContent | BlobResourceContent] | None: + if self.read_contents is not None: + return list(self.read_contents) + return [ + TextResourceContent( + uri=uri, + mime_type="text/plain", + meta={"source": self._server_name}, + text="catalog content", + ) + ] + + # ============================================================================= # Helpers # ============================================================================= @@ -180,16 +256,16 @@ def test_is_abstract_capability() -> None: def test_get_toolset_returns_function_toolset() -> None: - """get_toolset() returns a FunctionToolset with 5 tools.""" + """get_toolset() returns exactly the three formal MCP Resource tools.""" cap = ResourceCapability() toolset = cap.get_toolset() assert toolset is not None assert isinstance(toolset, FunctionToolset) - assert "list_resources" in toolset.tools - assert "read_resource" in toolset.tools - assert "resource_exists" in toolset.tools - assert "list_resource_templates" in toolset.tools - assert "complete_resource_template" in toolset.tools + assert set(toolset.tools) == { + "list_mcp_resources", + "list_mcp_resource_templates", + "read_mcp_resource", + } def test_name_property() -> None: @@ -199,18 +275,147 @@ def test_name_property() -> None: def test_get_instructions_returns_description() -> None: - """get_instructions() returns a non-None description mentioning all tools.""" + """get_instructions() describes progressive MCP Resource reads.""" cap = ResourceCapability() instructions = cap.get_instructions() assert instructions is not None - assert "list_resources" in instructions - assert "read_resource" in instructions - assert "resource_exists" in instructions - assert "list_resource_templates" in instructions - assert "complete_resource_template" in instructions + assert "list_mcp_resources" in instructions + assert "list_mcp_resource_templates" in instructions + assert "read_mcp_resource" in instructions # skill:// is NOT advertised (D2 — unadvertised fallback) assert "skill://" not in instructions - assert "mcp://" in instructions + assert "opaque URI" in instructions + + +async def test_formal_mcp_resource_tools_list_and_read_exact_server() -> None: + """Formal tools preserve server identity and return typed results.""" + provider = FakeMcpResourceProvider("unikb") + registry = _make_registry_with_caps(provider) + cap = ResourceCapability() + ctx = _make_ctx(_make_agent_context(registry)) + + listed = await cap.list_mcp_resources(ctx, limit=1) + assert listed.resources[0].server == "unikb" + assert listed.resources[0].meta == {"source": "unikb"} + templates = await cap.list_mcp_resource_templates(ctx) + assert templates.templates[0].server == "unikb" + read = await cap.read_mcp_resource(ctx, server="unikb", uri="kb:///resources/catalog") + assert read.return_value.uri == "kb:///resources/catalog" + assert read.return_value.contents[0]["text"] == "catalog content" + assert read.return_value.contents[0]["meta"] == {"source": "unikb"} + + +async def test_formal_mcp_resource_tools_reject_unknown_server() -> None: + provider = FakeMcpResourceProvider("unikb") + registry = _make_registry_with_caps(provider) + cap = ResourceCapability() + ctx = _make_ctx(_make_agent_context(registry)) + + result = await cap.read_mcp_resource(ctx, server="other", uri="kb:///resources/catalog") + assert result.return_value.errors[0].code == "unknown_server" + + +def test_registry_excludes_provider_after_capability_negotiation() -> None: + provider = FakeMcpResourceProvider("tools-only", supported=False) + registry = _make_registry_with_caps(provider) + + assert registry.get_mcp_resource_providers(Scope(level=ScopeLevel.POOL)) == [] + + +async def test_formal_tools_report_resources_not_supported_for_known_server() -> None: + provider = FakeMcpResourceProvider("tools-only", supported=False) + cap = ResourceCapability() + ctx = _make_ctx(_make_agent_context(_make_registry_with_caps(provider))) + + listed = await cap.list_mcp_resources(ctx, server="tools-only") + assert listed.errors[0].code == "resources_not_supported" + read = await cap.read_mcp_resource(ctx, server="tools-only", uri="kb:///same") + assert read.return_value.errors[0].code == "resources_not_supported" + + +async def test_formal_resource_cursor_tracks_current_server_across_pages() -> None: + alpha = FakeMcpResourceProvider("alpha") + alpha.resource_pages[None] = McpResourceListPage( + entries=[ + ResourceEntry(uri="docs://alpha/1", server="alpha", name="a1"), + ResourceEntry(uri="docs://alpha/2", server="alpha", name="a2"), + ] + ) + beta = FakeMcpResourceProvider("beta") + beta.resource_pages[None] = McpResourceListPage( + entries=[ResourceEntry(uri="docs://beta/1", server="beta", name="b1")] + ) + registry = _make_registry_with_caps(beta, alpha) + cap = ResourceCapability() + ctx = _make_ctx(_make_agent_context(registry)) + + first = await cap.list_mcp_resources(ctx, limit=1) + assert [entry.name for entry in first.resources] == ["a1"] + assert first.next_cursor is not None + decoded = cap._decode_cursor(first.next_cursor) + assert decoded["version"] == 2 + assert decoded["current_server"] == "alpha" + + second = await cap.list_mcp_resources(ctx, cursor=first.next_cursor, limit=1) + assert [entry.name for entry in second.resources] == ["a2"] + assert second.next_cursor is not None + third = await cap.list_mcp_resources(ctx, cursor=second.next_cursor, limit=1) + assert [entry.name for entry in third.resources] == ["b1"] + + conflict = await cap.list_mcp_resources( + ctx, + server="beta", + cursor=first.next_cursor, + limit=1, + ) + assert conflict.errors[0].code == "invalid_cursor" + + +async def test_formal_resource_list_keeps_results_when_provider_fails() -> None: + class FailingProvider(FakeMcpResourceProvider): + async def list_resources_page(self, cursor: str | None = None) -> McpResourceListPage: + raise RuntimeError("temporary provider outage") + + failing = FailingProvider("alpha") + healthy = FakeMcpResourceProvider("beta") + healthy.resource_pages[None] = McpResourceListPage( + entries=[ResourceEntry(uri="docs://beta/1", server="beta", name="healthy")] + ) + registry = _make_registry_with_caps(failing, healthy) + cap = ResourceCapability() + result = await cap.list_mcp_resources(_make_ctx(_make_agent_context(registry)), limit=10) + + assert [entry.name for entry in result.resources] == ["healthy"] + assert result.errors[0].code == "provider_unavailable" + assert result.errors[0].retryable is True + + +async def test_formal_read_truncates_text_and_omits_unsupported_or_large_blobs() -> None: + provider = FakeMcpResourceProvider( + "server", + read_contents=[ + TextResourceContent(uri="kb:///long", text="x" * 10_001), + BlobResourceContent( + uri="kb:///gif", mime_type="image/bmp", blob=base64.b64encode(b"bmp").decode() + ), + ], + ) + registry = _make_registry_with_caps(provider) + cap = ResourceCapability() + result = await cap.read_mcp_resource( + _make_ctx(_make_agent_context(registry)), server="server", uri="kb:///long" + ) + + assert result.return_value.truncated is True + assert result.return_value.original_char_count == 10_001 + assert len(result.return_value.contents[0]["text"]) == 10_000 + assert result.return_value.errors[0].code == "unsupported_mime_type" + + blob_result = await cap.read_mcp_resource( + _make_ctx(_make_agent_context(registry)), server="server", uri="kb:///gif" + ) + assert blob_result.return_value.contents[-1]["attached"] is False + assert blob_result.return_value.errors[0].code == "unsupported_mime_type" async def test_stateless_lifecycle() -> None: diff --git a/tests/capabilities/test_skill_manager_cap_resource_access.py b/tests/capabilities/test_skill_manager_cap_resource_access.py index bd950e252..de19f0e69 100644 --- a/tests/capabilities/test_skill_manager_cap_resource_access.py +++ b/tests/capabilities/test_skill_manager_cap_resource_access.py @@ -23,6 +23,10 @@ class FakeResourceChild: """Stand-in for a per-skill ``McpServerCap`` implementing ``ResourceAccess``.""" + @property + def owned_schemes(self) -> frozenset[str]: + return frozenset() + def __init__( self, name: str, diff --git a/tests/delegation/test_pool_skills.py b/tests/delegation/test_pool_skills.py index 68a919874..fec5fda53 100644 --- a/tests/delegation/test_pool_skills.py +++ b/tests/delegation/test_pool_skills.py @@ -434,6 +434,9 @@ async def __aenter__(self) -> Self: async def __aexit__(self, *args: object) -> None: return None + async def supports_resources(self) -> bool: + return False + monkeypatch.setattr( "wolfharness.mcp_server.client.MCPClient", _FakeClient, diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index bcc9000f4..5dff4db68 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -449,6 +449,69 @@ def session_e2e_config_with_mcp(tmp_path_factory: pytest.TempPathFactory) -> Pat return config_path +def _resource_mcp_config( + config_path: Path, + fixtures_dir: Path, + *, + resources_enabled: bool, +) -> Path: + """Write a two-provider MCP resource config used by L4 tests.""" + fake_server_path = fixtures_dir / "fake_mcp_server_resources.py" + config = { + "agents": { + "test_agent": { + "type": "native", + "model": "test", + "system_prompt": "You are a test assistant.", + "resources": {"enabled": resources_enabled}, + } + }, + "mcp_servers": [ + { + "name": "alpha:%", + "type": "stdio", + "command": sys.executable, + "args": [str(fake_server_path)], + "enabled": True, + }, + { + "name": "beta", + "type": "stdio", + "command": sys.executable, + "args": [str(fake_server_path)], + "enabled": True, + }, + ], + "storage": {"providers": [{"type": "memory"}]}, + } + config_path.write_text(yaml.dump(config, default_flow_style=False)) + return config_path + + +@pytest.fixture(scope="session") +def session_e2e_config_with_mcp_resources(tmp_path_factory: pytest.TempPathFactory) -> Path: + """Session config with two MCP providers exposing identical resources.""" + config_dir = tmp_path_factory.mktemp("e2e_mcp_resources") + return _resource_mcp_config( + config_dir / "e2e_config_mcp_resources.yml", + Path(__file__).resolve().parent.parent / "fixtures", + resources_enabled=True, + ) + + +@pytest.fixture(scope="session") +def session_e2e_config_with_mcp_resources_disabled( + tmp_path_factory: pytest.TempPathFactory, +) -> Path: + """Session config where resource model tools are disabled.""" + config_dir = tmp_path_factory.mktemp("e2e_mcp_resources_disabled") + return _resource_mcp_config( + config_dir / "e2e_config_mcp_resources_disabled.yml", + Path(__file__).resolve().parent.parent / "fixtures", + resources_enabled=False, + ) + + # --------------------------------------------------------------------------- # Subprocess server cache infrastructure # --------------------------------------------------------------------------- @@ -980,6 +1043,69 @@ async def subprocess_server_with_mcp( yield server +async def _subprocess_server_with_config( + request: pytest.FixtureRequest, + process_registry: ProcessRegistry, + config_path: Path, + allow_model_requests: Any, +) -> AsyncIterator[SubprocessServer]: + """Spawn an OpenCode subprocess for a supplied MCP-resource config.""" + _ = allow_model_requests + params = getattr(request, "param", {"serve_command": "serve-opencode"}) + serve_command: str = params.get("serve_command", "serve-opencode") + host: str = params.get("host", "127.0.0.1") + is_stdio: bool = params.get("is_stdio", False) + health_timeout: float = params.get("health_timeout", 30.0) + health_path: str = params.get("health_path", "/session") + extra_args: list[str] | None = params.get("extra_args") + + async for server in _spawn_server( + serve_command, + config_path, + process_registry=process_registry, + host=host, + is_stdio=is_stdio, + health_timeout=health_timeout, + health_path=health_path, + extra_args=extra_args, + ): + yield server + + +@pytest.fixture +async def subprocess_server_with_mcp_resources( + request: pytest.FixtureRequest, + process_registry: ProcessRegistry, + session_e2e_config_with_mcp_resources: Path, + allow_model_requests: Any, +) -> AsyncIterator[SubprocessServer]: + """Spawn OpenCode with two identical-resource MCP providers.""" + async for server in _subprocess_server_with_config( + request, + process_registry, + session_e2e_config_with_mcp_resources, + allow_model_requests, + ): + yield server + + +@pytest.fixture +async def subprocess_server_with_mcp_resources_disabled( + request: pytest.FixtureRequest, + process_registry: ProcessRegistry, + session_e2e_config_with_mcp_resources_disabled: Path, + allow_model_requests: Any, +) -> AsyncIterator[SubprocessServer]: + """Spawn OpenCode with resource providers but hidden model tools.""" + async for server in _subprocess_server_with_config( + request, + process_registry, + session_e2e_config_with_mcp_resources_disabled, + allow_model_requests, + ): + yield server + + # --------------------------------------------------------------------------- # ACP WebSocket (streamable-http) server fixture (B1.1) # --------------------------------------------------------------------------- diff --git a/tests/e2e/test_opencode_mcp_resources.py b/tests/e2e/test_opencode_mcp_resources.py new file mode 100644 index 000000000..0c7bb2fc7 --- /dev/null +++ b/tests/e2e/test_opencode_mcp_resources.py @@ -0,0 +1,205 @@ +"""L4 OpenCode tests for MCP resource discovery and exact routing.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import httpx +import pytest + +from tests.e2e.conftest import SKIP_NO_BINARY, SKIP_WINDOWS + + +if TYPE_CHECKING: + from pathlib import Path + + from tests.e2e.conftest import SubprocessServer + + +pytestmark = [ + pytest.mark.e2e, + pytest.mark.skipif(SKIP_NO_BINARY, reason="wolfharness binary not on PATH"), + pytest.mark.skipif(SKIP_WINDOWS, reason="Windows subprocess issues"), +] + + +@pytest.mark.parametrize( + "subprocess_server_with_mcp_resources", + [{"serve_command": "serve-opencode", "is_stdio": False, "health_path": "/session"}], + indirect=True, +) +async def test_resource_catalog_preserves_server_identity( + subprocess_server_with_mcp_resources: SubprocessServer, +) -> None: + """The Host catalog keeps both providers when they expose one URI.""" + async with httpx.AsyncClient(timeout=30.0) as client: + first_session = await client.post( + f"{subprocess_server_with_mcp_resources.base_url}/session", json={} + ) + second_session = await client.post( + f"{subprocess_server_with_mcp_resources.base_url}/session", json={} + ) + assert first_session.status_code in (200, 201), first_session.text + assert second_session.status_code in (200, 201), second_session.text + response = await client.get( + f"{subprocess_server_with_mcp_resources.base_url}/experimental/resource" + ) + second_response = await client.get( + f"{subprocess_server_with_mcp_resources.base_url}/experimental/resource" + ) + + assert response.status_code == 200, response.text + assert second_response.status_code == 200, second_response.text + catalog = response.json() + assert second_response.json() == catalog + assert "alpha%3A%25:file:///shared.txt" in catalog + assert "beta:file:///shared.txt" in catalog + assert catalog["alpha%3A%25:file:///shared.txt"]["client"] == "alpha:%" + assert catalog["beta:file:///shared.txt"]["client"] == "beta" + assert catalog["beta:file:///shared.txt"]["mimeType"] == "text/plain" + + +@pytest.mark.parametrize( + "subprocess_server_with_mcp_resources", + [{"serve_command": "serve-opencode", "is_stdio": False, "health_path": "/session"}], + indirect=True, +) +async def test_resource_tools_are_the_only_formal_resource_tools( + subprocess_server_with_mcp_resources: SubprocessServer, +) -> None: + """The real HTTP tool surface exposes exactly the three formal resource tools.""" + async with httpx.AsyncClient(timeout=30.0) as client: + session_response = await client.post( + f"{subprocess_server_with_mcp_resources.base_url}/session", json={} + ) + assert session_response.status_code in (200, 201), session_response.text + session_id = session_response.json().get("id") or session_response.json().get("sessionID") + assert session_id + message_response = await client.post( + f"{subprocess_server_with_mcp_resources.base_url}/session/{session_id}/message", + json={"parts": [{"type": "text", "text": "initialize tools"}]}, + ) + assert message_response.status_code in (200, 201, 202), message_response.text + response = await client.get( + f"{subprocess_server_with_mcp_resources.base_url}/experimental/tool/ids" + ) + schema_response = await client.get( + f"{subprocess_server_with_mcp_resources.base_url}/experimental/tool" + ) + + assert response.status_code == 200, response.text + assert schema_response.status_code == 200, schema_response.text + tool_ids = set(response.json()) + schema_tools = {item["id"]: item for item in schema_response.json()} + assert { + "list_mcp_resources", + "list_mcp_resource_templates", + "read_mcp_resource", + } <= tool_ids + assert { + "list_mcp_resources", + "list_mcp_resource_templates", + "read_mcp_resource", + } <= schema_tools.keys() + assert schema_tools["read_mcp_resource"]["parameters"]["required"] == ["server", "uri"] + assert ( + not { + "list_resources", + "list_resource_templates", + "read_resource", + "read_section", + "read_section_text", + } + & tool_ids + ) + + +@pytest.mark.parametrize( + "subprocess_server_with_mcp_resources_disabled", + [{"serve_command": "serve-opencode", "is_stdio": False, "health_path": "/session"}], + indirect=True, +) +async def test_resource_gate_hides_tools_but_not_host_catalog( + subprocess_server_with_mcp_resources_disabled: SubprocessServer, +) -> None: + """resources.enabled=false only hides model tools, not Host resources.""" + base_url = subprocess_server_with_mcp_resources_disabled.base_url + async with httpx.AsyncClient(timeout=30.0) as client: + session_response = await client.post(f"{base_url}/session", json={}) + assert session_response.status_code in (200, 201), session_response.text + session_id = session_response.json().get("id") or session_response.json().get("sessionID") + assert session_id + message_response = await client.post( + f"{base_url}/session/{session_id}/message", + json={"parts": [{"type": "text", "text": "initialize tools"}]}, + ) + assert message_response.status_code in (200, 201, 202), message_response.text + catalog_response = await client.get(f"{base_url}/experimental/resource") + tools_response = await client.get(f"{base_url}/experimental/tool/ids") + resource_message_response = await client.post( + f"{base_url}/session/{session_id}/message", + json={ + "noReply": True, + "parts": [ + { + "type": "file", + "mime": "text/plain", + "url": "", + "source": { + "type": "resource", + "clientName": "beta", + "uri": "file:///shared.txt", + "text": {"value": "shared resource", "start": 0, "end": 15}, + }, + } + ], + }, + ) + + assert catalog_response.status_code == 200, catalog_response.text + assert "beta:file:///shared.txt" in catalog_response.json() + assert tools_response.status_code == 200, tools_response.text + assert "read_mcp_resource" not in set(tools_response.json()) + assert resource_message_response.status_code in (200, 201, 202), resource_message_response.text + + +@pytest.mark.parametrize( + "subprocess_server_with_mcp_resources", + [{"serve_command": "serve-opencode", "is_stdio": False, "health_path": "/session"}], + indirect=True, +) +async def test_file_part_resource_source_is_accepted( + subprocess_server_with_mcp_resources: SubprocessServer, + e2e_config: Path, +) -> None: + """OpenCode accepts a real FilePart carrying a ResourceSource.""" + _ = e2e_config + base_url = subprocess_server_with_mcp_resources.base_url + async with httpx.AsyncClient(timeout=30.0) as client: + session_response = await client.post(f"{base_url}/session", json={}) + assert session_response.status_code in (200, 201), session_response.text + session_id = session_response.json().get("id") or session_response.json().get("sessionID") + assert session_id + + payload: dict[str, Any] = { + "noReply": True, + "parts": [ + { + "type": "file", + "mime": "text/plain", + "url": "", + "source": { + "type": "resource", + "clientName": "beta", + "uri": "file:///shared.txt", + "text": {"value": "shared resource", "start": 0, "end": 15}, + }, + } + ], + } + message_response = await client.post( + f"{base_url}/session/{session_id}/message", + json=payload, + ) + + assert message_response.status_code in (200, 201, 202), message_response.text diff --git a/tests/fixtures/fake_mcp_server_resources.py b/tests/fixtures/fake_mcp_server_resources.py new file mode 100644 index 000000000..0e3199b2b --- /dev/null +++ b/tests/fixtures/fake_mcp_server_resources.py @@ -0,0 +1,36 @@ +"""Fake MCP server exposing resources for AgentPool L4 tests.""" + +from __future__ import annotations + +from fastmcp import FastMCP + + +mcp = FastMCP("fake-resource-server", version="0.1.0") + + +@mcp.tool() +def search_kb(query: str) -> str: + """Return a deterministic search result for the requested query.""" + return f"Results for: {query}" + + +@mcp.resource("file:///shared.txt", mime_type="text/plain") +def shared_text() -> str: + """Return the shared text resource.""" + return "
shared resource text
" + + +@mcp.resource("file:///image.png", mime_type="image/png") +def image_resource() -> bytes: + """Return a small PNG-shaped binary resource.""" + return b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR" + + +@mcp.resource("file:///{path}", mime_type="text/plain") +def templated_file(path: str) -> str: + """Return a deterministic dynamic resource.""" + return f"template:{path}" + + +if __name__ == "__main__": + mcp.run() diff --git a/tests/fixtures/test_resource_cap.py b/tests/fixtures/test_resource_cap.py index 86a9afa06..d2ba398a9 100644 --- a/tests/fixtures/test_resource_cap.py +++ b/tests/fixtures/test_resource_cap.py @@ -52,6 +52,10 @@ class TestResourceAccessCap(AbstractCapability[Any]): read_uri: str = "test://doc.md" _owns_client: bool = field(default=False, repr=False) + @property + def owned_schemes(self) -> frozenset[str]: + return frozenset() + def __post_init__(self) -> None: pass @@ -121,6 +125,10 @@ class TestToolAndResourceCap(AbstractCapability[Any]): read_uri: str = "test://tool-doc.md" _owns_client: bool = field(default=False, repr=False) + @property + def owned_schemes(self) -> frozenset[str]: + return frozenset() + def __post_init__(self) -> None: pass diff --git a/tests/servers/opencode_server/test_resource_resolution.py b/tests/servers/opencode_server/test_resource_resolution.py index 50df609cb..199d3123e 100644 --- a/tests/servers/opencode_server/test_resource_resolution.py +++ b/tests/servers/opencode_server/test_resource_resolution.py @@ -47,6 +47,10 @@ class FakeResourceAccess: protocol: ``list_resources``, ``read_resource``, ``resource_exists``. """ + @property + def owned_schemes(self) -> frozenset[str]: + return frozenset() + def __init__( self, read_result: list[TextResourceContent | BlobResourceContent] | None = None, @@ -223,6 +227,27 @@ async def test_resolve_resource_multiple_providers() -> None: assert result == ['\nfound\n'] +async def test_resolve_resource_routes_same_uri_by_server_name() -> None: + """A ResourceSource server name must prevent same-URI provider mixing.""" + + class NamedResourceAccess(FakeResourceAccess): + def __init__(self, server_name: str, text: str) -> None: + super().__init__(read_result=[TextResourceContent(uri="kb:///same", text=text)]) + self.server_name = server_name + + first = NamedResourceAccess("alpha", "alpha content") + second = NamedResourceAccess("beta", "beta content") + + result = await resolve_resource_content( + "kb:///same", + resource_caps=[first, second], + skill_caps=[], + client_name="beta", + ) + + assert result == ['\nbeta content\n'] + + async def test_resolve_resource_skill_uri() -> None: """``skill://`` URI routed to SkillResource, not ResourceAccess.""" skill_cap = FakeSkillResource(read_result="content") @@ -485,12 +510,13 @@ async def test_extract_user_prompt_with_binary_resource() -> None: result = await extract_user_prompt_from_parts([part], "test-session", agent=agent) result_list = list(result) - assert len(result_list) == 3 - assert result_list[0] == '\n' - assert isinstance(result_list[1], BinaryContent) - assert result_list[1].data == b"img" - assert result_list[1].media_type == "image/png" - assert result_list[2] == "\n" + assert len(result_list) == 4 + assert result_list[0] == "[Resource from viking server] viking://img.png" + assert result_list[1] == '\n' + assert isinstance(result_list[2], BinaryContent) + assert result_list[2].data == b"img" + assert result_list[2].media_type == "image/png" + assert result_list[3] == "\n" async def test_extract_user_prompt_resource_no_agent() -> None: @@ -547,11 +573,12 @@ async def test_extract_user_prompt_mixed_parts() -> None: [text_part, resource_part, agent_part], "test-session", agent=agent ) result_list = list(result) - # 1 text + 1 resource (XML-wrapped) + 1 agent instruction - assert len(result_list) == 3 + # 1 text + 1 header + 1 resource (XML-wrapped) + 1 agent instruction + assert len(result_list) == 4 assert result_list[0] == "prefix text" - assert result_list[1] == '\nresource content\n' - assert "researcher" in result_list[2] + assert result_list[1] == "[Resource from viking server] viking://doc" + assert result_list[2] == '\nresource content\n' + assert "researcher" in result_list[3] # ============================================================================= @@ -653,9 +680,10 @@ async def test_resolve_resource_timing_bug() -> None: result = await extract_user_prompt_from_parts([part], "test-session", agent=agent) result_list = list(result) - # Should resolve the resource content — not drop it silently. - assert len(result_list) == 1 - assert result_list[0] == '\nhello world\n' + # Should prepend a server-source header and resolve the resource content. + assert len(result_list) == 2 + assert result_list[0] == "[Resource from test server] test://doc.md" + assert result_list[1] == '\nhello world\n' # ============================================================================= @@ -733,8 +761,9 @@ async def test_e2e_at_mention_resolves_resource() -> None: result = await extract_user_prompt_from_parts([part], "test-session", agent=agent) result_list = list(result) - assert len(result_list) == 1 - assert result_list[0] == '\nhello world\n' + assert len(result_list) == 2 + assert result_list[0] == "[Resource from test server] test://doc.md" + assert result_list[1] == '\nhello world\n' async def test_e2e_at_mention_wrong_uri_returns_empty() -> None: @@ -871,8 +900,9 @@ async def test_e2e_skill_resource_resolution() -> None: result = await extract_user_prompt_from_parts([part], "test-session", agent=agent) result_list = list(result) - assert len(result_list) == 1 - assert "skill body" in result_list[0] + assert len(result_list) == 2 + assert result_list[0] == "[Resource from test server] skill://test-skill" + assert "skill body" in result_list[1] # =============================================================================