diff --git a/src/fastmcp/contrib/component_manager/component_service.py b/src/fastmcp/contrib/component_manager/component_service.py index 4b991d86ef..6b37573582 100644 --- a/src/fastmcp/contrib/component_manager/component_service.py +++ b/src/fastmcp/contrib/component_manager/component_service.py @@ -7,7 +7,7 @@ from fastmcp.prompts.prompt import Prompt from fastmcp.resources.resource import Resource from fastmcp.resources.template import ResourceTemplate -from fastmcp.server.providers import MountedProvider +from fastmcp.server.providers import FastMCPProvider, Provider, TransformingProvider from fastmcp.server.server import FastMCP from fastmcp.tools.tool import Tool from fastmcp.utilities.logging import get_logger @@ -15,6 +15,41 @@ logger = get_logger(__name__) +def _get_mounted_server_and_key( + provider: Provider, + key: str, + component_type: str, +) -> tuple[FastMCP, str] | None: + """Get the mounted server and unprefixed key for a component. + + Args: + provider: The provider to check. + key: The transformed component key. + component_type: Either "tool" (for tools/prompts) or "resource". + + Returns: + Tuple of (server, original_key) if the key matches this provider, + or None if it doesn't. + """ + if isinstance(provider, TransformingProvider): + # TransformingProvider - reverse the transformation + if component_type == "resource": + original = provider._reverse_resource_uri(key) + else: + original = provider._reverse_tool_name(key) + + if original is not None: + # Recursively check the wrapped provider + return _get_mounted_server_and_key( + provider._wrapped, original, component_type + ) + elif isinstance(provider, FastMCPProvider): + # Direct FastMCPProvider - no transformation, key is used directly + return provider.server, key + + return None + + class ComponentService: """Service for managing components like tools, resources, and prompts.""" @@ -41,14 +76,14 @@ async def _enable_tool(self, key: str) -> Tool: tool.enable() return tool - # 2. Check mounted servers via MountedProvider + # 2. Check mounted servers via FastMCPProvider/TransformingProvider for provider in self._server._providers: - if isinstance(provider, MountedProvider): - unprefixed = provider._strip_tool_prefix(key) - if unprefixed is not None: - mounted_service = ComponentService(provider.server) - tool = await mounted_service._enable_tool(unprefixed) - return tool + result = _get_mounted_server_and_key(provider, key, "tool") + if result is not None: + server, unprefixed = result + mounted_service = ComponentService(server) + tool = await mounted_service._enable_tool(unprefixed) + return tool raise NotFoundError(f"Unknown tool: {key}") async def _disable_tool(self, key: str) -> Tool: @@ -68,14 +103,14 @@ async def _disable_tool(self, key: str) -> Tool: tool.disable() return tool - # 2. Check mounted servers via MountedProvider + # 2. Check mounted servers via FastMCPProvider/TransformingProvider for provider in self._server._providers: - if isinstance(provider, MountedProvider): - unprefixed = provider._strip_tool_prefix(key) - if unprefixed is not None: - mounted_service = ComponentService(provider.server) - tool = await mounted_service._disable_tool(unprefixed) - return tool + result = _get_mounted_server_and_key(provider, key, "tool") + if result is not None: + server, unprefixed = result + mounted_service = ComponentService(server) + tool = await mounted_service._disable_tool(unprefixed) + return tool raise NotFoundError(f"Unknown tool: {key}") async def _enable_resource(self, key: str) -> Resource | ResourceTemplate: @@ -99,16 +134,16 @@ async def _enable_resource(self, key: str) -> Resource | ResourceTemplate: template.enable() return template - # 2. Check mounted servers via MountedProvider + # 2. Check mounted servers via FastMCPProvider/TransformingProvider for provider in self._server._providers: - if isinstance(provider, MountedProvider): - unprefixed = provider._strip_resource_prefix(key) - if unprefixed is not None: - mounted_service = ComponentService(provider.server) - mounted_resource: ( - Resource | ResourceTemplate - ) = await mounted_service._enable_resource(unprefixed) - return mounted_resource + result = _get_mounted_server_and_key(provider, key, "resource") + if result is not None: + server, unprefixed = result + mounted_service = ComponentService(server) + mounted_resource: ( + Resource | ResourceTemplate + ) = await mounted_service._enable_resource(unprefixed) + return mounted_resource raise NotFoundError(f"Unknown resource: {key}") async def _disable_resource(self, key: str) -> Resource | ResourceTemplate: @@ -132,16 +167,16 @@ async def _disable_resource(self, key: str) -> Resource | ResourceTemplate: template.disable() return template - # 2. Check mounted servers via MountedProvider + # 2. Check mounted servers via FastMCPProvider/TransformingProvider for provider in self._server._providers: - if isinstance(provider, MountedProvider): - unprefixed = provider._strip_resource_prefix(key) - if unprefixed is not None: - mounted_service = ComponentService(provider.server) - mounted_resource: ( - Resource | ResourceTemplate - ) = await mounted_service._disable_resource(unprefixed) - return mounted_resource + result = _get_mounted_server_and_key(provider, key, "resource") + if result is not None: + server, unprefixed = result + mounted_service = ComponentService(server) + mounted_resource: ( + Resource | ResourceTemplate + ) = await mounted_service._disable_resource(unprefixed) + return mounted_resource raise NotFoundError(f"Unknown resource: {key}") async def _enable_prompt(self, key: str) -> Prompt: @@ -161,14 +196,14 @@ async def _enable_prompt(self, key: str) -> Prompt: prompt.enable() return prompt - # 2. Check mounted servers via MountedProvider + # 2. Check mounted servers via FastMCPProvider/TransformingProvider for provider in self._server._providers: - if isinstance(provider, MountedProvider): - unprefixed = provider._strip_tool_prefix(key) - if unprefixed is not None: - mounted_service = ComponentService(provider.server) - prompt = await mounted_service._enable_prompt(unprefixed) - return prompt + result = _get_mounted_server_and_key(provider, key, "tool") + if result is not None: + server, unprefixed = result + mounted_service = ComponentService(server) + prompt = await mounted_service._enable_prompt(unprefixed) + return prompt raise NotFoundError(f"Unknown prompt: {key}") async def _disable_prompt(self, key: str) -> Prompt: @@ -187,12 +222,12 @@ async def _disable_prompt(self, key: str) -> Prompt: prompt.disable() return prompt - # 2. Check mounted servers via MountedProvider + # 2. Check mounted servers via FastMCPProvider/TransformingProvider for provider in self._server._providers: - if isinstance(provider, MountedProvider): - unprefixed = provider._strip_tool_prefix(key) - if unprefixed is not None: - mounted_service = ComponentService(provider.server) - prompt = await mounted_service._disable_prompt(unprefixed) - return prompt + result = _get_mounted_server_and_key(provider, key, "tool") + if result is not None: + server, unprefixed = result + mounted_service = ComponentService(server) + prompt = await mounted_service._disable_prompt(unprefixed) + return prompt raise NotFoundError(f"Unknown prompt: {key}") diff --git a/src/fastmcp/server/providers/__init__.py b/src/fastmcp/server/providers/__init__.py index 94d0a4397e..e35581f00b 100644 --- a/src/fastmcp/server/providers/__init__.py +++ b/src/fastmcp/server/providers/__init__.py @@ -25,12 +25,12 @@ async def get_tool(self, name: str) -> Tool | None: ``` """ -from fastmcp.server.providers.base import Components, Provider, TaskComponents -from fastmcp.server.providers.mounted import MountedProvider +from fastmcp.server.providers.base import Provider +from fastmcp.server.providers.fastmcp_provider import FastMCPProvider +from fastmcp.server.providers.transforming import TransformingProvider __all__ = [ - "Components", - "MountedProvider", + "FastMCPProvider", "Provider", - "TaskComponents", + "TransformingProvider", ] diff --git a/src/fastmcp/server/providers/base.py b/src/fastmcp/server/providers/base.py index acf6a31cd5..c15bc24eeb 100644 --- a/src/fastmcp/server/providers/base.py +++ b/src/fastmcp/server/providers/base.py @@ -89,6 +89,74 @@ class Provider: exceptions are wrapped with optional detail masking. """ + def with_transforms( + self, + *, + namespace: str | None = None, + tool_renames: dict[str, str] | None = None, + ) -> Provider: + """Apply transformations to this provider's components. + + Returns a TransformingProvider that wraps this provider and applies + the specified transformations. Can be chained - each call creates a + new wrapper that composes with the previous. + + Args: + namespace: Prefix for tools/prompts ("namespace_name"), path segment + for resources ("protocol://namespace/path"). + tool_renames: Map of original_name → final_name. Tools in this map + use the specified name instead of namespace prefixing. + + Returns: + A TransformingProvider wrapping this provider. + + Example: + ```python + # Apply namespace to all components + provider = MyProvider().with_transforms(namespace="db") + # Tool "greet" becomes "db_greet" + # Resource "resource://data" becomes "resource://db/data" + + # Rename specific tools (bypasses namespace for those tools) + provider = MyProvider().with_transforms( + namespace="api", + tool_renames={"verbose_tool_name": "short"} + ) + # "verbose_tool_name" → "short" (explicit rename) + # "other_tool" → "api_other_tool" (namespace applied) + + # Stacking composes transformations + provider = ( + MyProvider() + .with_transforms(namespace="api") + .with_transforms(tool_renames={"api_foo": "bar"}) + ) + # "foo" → "api_foo" (inner) → "bar" (outer) + ``` + """ + from fastmcp.server.providers.transforming import TransformingProvider + + return TransformingProvider( + self, namespace=namespace, tool_renames=tool_renames + ) + + def with_namespace(self, namespace: str) -> Provider: + """Shorthand for with_transforms(namespace=...). + + Args: + namespace: The namespace to apply. + + Returns: + A TransformingProvider wrapping this provider. + + Example: + ```python + provider = MyProvider().with_namespace("db") + # Equivalent to: MyProvider().with_transforms(namespace="db") + ``` + """ + return self.with_transforms(namespace=namespace) + async def list_tools(self) -> Sequence[Tool]: """Return all available tools. diff --git a/src/fastmcp/server/providers/fastmcp_provider.py b/src/fastmcp/server/providers/fastmcp_provider.py new file mode 100644 index 0000000000..e1ca61855b --- /dev/null +++ b/src/fastmcp/server/providers/fastmcp_provider.py @@ -0,0 +1,223 @@ +"""FastMCPProvider for wrapping FastMCP servers as providers. + +This module provides the `FastMCPProvider` class that wraps a FastMCP server +and exposes its components through the Provider interface. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Sequence +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING, Any + +from fastmcp.exceptions import NotFoundError +from fastmcp.prompts.prompt import Prompt, PromptResult +from fastmcp.resources.resource import Resource, ResourceContent +from fastmcp.resources.template import ResourceTemplate +from fastmcp.server.providers.base import Provider, TaskComponents +from fastmcp.tools.tool import Tool, ToolResult + +if TYPE_CHECKING: + from fastmcp.prompts.prompt import FunctionPrompt + from fastmcp.resources.resource import FunctionResource + from fastmcp.resources.template import FunctionResourceTemplate + from fastmcp.server.server import FastMCP + from fastmcp.tools.tool import FunctionTool + + +class FastMCPProvider(Provider): + """Provider that wraps a FastMCP server. + + This provider enables mounting one FastMCP server onto another, exposing + the mounted server's tools, resources, and prompts through the parent + server. + + Execution methods (`call_tool`, `read_resource`, `render_prompt`) invoke + the mounted server's middleware chain. + + Example: + ```python + from fastmcp import FastMCP + from fastmcp.server.providers import FastMCPProvider + + main = FastMCP("Main") + sub = FastMCP("Sub") + + @sub.tool + def greet(name: str) -> str: + return f"Hello, {name}!" + + # Mount directly - tools accessible by original names + main.add_provider(FastMCPProvider(sub)) + + # Or with namespace + main.add_provider(FastMCPProvider(sub).with_namespace("sub")) + ``` + + Note: + Normally you would use `FastMCP.mount()` which handles proxy conversion + and creates the provider with namespace automatically. + """ + + def __init__(self, server: FastMCP[Any]): + """Initialize a FastMCPProvider. + + Args: + server: The FastMCP server to wrap. + """ + super().__init__() + self.server = server + + # ------------------------------------------------------------------------- + # Tool methods + # ------------------------------------------------------------------------- + + async def list_tools(self) -> Sequence[Tool]: + """List all tools from the mounted server.""" + return await self.server._list_tools_middleware() + + async def get_tool(self, name: str) -> Tool | None: + """Get a tool by name.""" + tools = await self.list_tools() + return next((t for t in tools if t.name == name), None) + + async def call_tool( + self, name: str, arguments: dict[str, Any] + ) -> ToolResult | None: + """Execute a tool through the mounted server's middleware chain.""" + return await self.server._call_tool_middleware(name, arguments) + + # ------------------------------------------------------------------------- + # Resource methods + # ------------------------------------------------------------------------- + + async def list_resources(self) -> Sequence[Resource]: + """List all resources from the mounted server.""" + return await self.server._list_resources_middleware() + + async def get_resource(self, uri: str) -> Resource | None: + """Get a concrete resource by URI.""" + resources = await self.list_resources() + return next((r for r in resources if str(r.uri) == uri), None) + + async def read_resource(self, uri: str) -> ResourceContent | None: + """Read a resource through the mounted server's middleware chain.""" + try: + contents = await self.server._read_resource_middleware(uri) + return contents[0] if contents else None + except NotFoundError: + return None + + # ------------------------------------------------------------------------- + # Resource template methods + # ------------------------------------------------------------------------- + + async def list_resource_templates(self) -> Sequence[ResourceTemplate]: + """List all resource templates from the mounted server.""" + return await self.server._list_resource_templates_middleware() + + async def get_resource_template(self, uri: str) -> ResourceTemplate | None: + """Get a resource template that matches the given URI.""" + templates = await self.list_resource_templates() + for template in templates: + if template.matches(uri) is not None: + return template + return None + + async def read_resource_template(self, uri: str) -> ResourceContent | None: + """Read a resource via a matching template through the mounted server.""" + # The server's middleware handles template resolution + return await self.read_resource(uri) + + # ------------------------------------------------------------------------- + # Prompt methods + # ------------------------------------------------------------------------- + + async def list_prompts(self) -> Sequence[Prompt]: + """List all prompts from the mounted server.""" + return await self.server._list_prompts_middleware() + + async def get_prompt(self, name: str) -> Prompt | None: + """Get a prompt by name.""" + prompts = await self.list_prompts() + return next((p for p in prompts if p.name == name), None) + + async def render_prompt( + self, name: str, arguments: dict[str, Any] | None + ) -> PromptResult | None: + """Render a prompt through the mounted server's middleware chain.""" + return await self.server._get_prompt_content_middleware(name, arguments) + + # ------------------------------------------------------------------------- + # Task registration + # ------------------------------------------------------------------------- + + async def get_tasks(self) -> TaskComponents: + """Return task-eligible components from the mounted server. + + Accesses the wrapped server's managers directly to avoid triggering + middleware during registration. Also recursively collects tasks from + nested providers. + """ + from fastmcp.prompts.prompt import FunctionPrompt + from fastmcp.resources.resource import FunctionResource + from fastmcp.resources.template import FunctionResourceTemplate + from fastmcp.tools.tool import FunctionTool + + tools: list[FunctionTool] = [] + resources: list[FunctionResource] = [] + templates: list[FunctionResourceTemplate] = [] + prompts: list[FunctionPrompt] = [] + + # Direct manager access (bypasses middleware) + for tool in self.server._tool_manager._tools.values(): + if isinstance(tool, FunctionTool) and tool.task_config.mode != "forbidden": + tools.append(tool) + + for resource in self.server._resource_manager._resources.values(): + if ( + isinstance(resource, FunctionResource) + and resource.task_config.mode != "forbidden" + ): + resources.append(resource) + + for template in self.server._resource_manager._templates.values(): + if ( + isinstance(template, FunctionResourceTemplate) + and template.task_config.mode != "forbidden" + ): + templates.append(template) + + for prompt in self.server._prompt_manager._prompts.values(): + if ( + isinstance(prompt, FunctionPrompt) + and prompt.task_config.mode != "forbidden" + ): + prompts.append(prompt) + + # Recursively get tasks from nested providers + for provider in self.server._providers: + nested = await provider.get_tasks() + tools.extend(nested.tools) + resources.extend(nested.resources) + templates.extend(nested.templates) + prompts.extend(nested.prompts) + + return TaskComponents( + tools=tools, resources=resources, templates=templates, prompts=prompts + ) + + # ------------------------------------------------------------------------- + # Lifecycle methods + # ------------------------------------------------------------------------- + + @asynccontextmanager + async def lifespan(self) -> AsyncIterator[None]: + """Start the mounted server's user lifespan. + + This starts only the wrapped server's user-defined lifespan, NOT its + full _lifespan_manager() (which includes Docket). The parent server's + Docket handles all background tasks. + """ + async with self.server._lifespan(self.server): + yield diff --git a/src/fastmcp/server/providers/mounted.py b/src/fastmcp/server/providers/mounted.py deleted file mode 100644 index 42809a2879..0000000000 --- a/src/fastmcp/server/providers/mounted.py +++ /dev/null @@ -1,405 +0,0 @@ -"""MountedProvider for wrapping mounted FastMCP servers. - -This module provides the `MountedProvider` class that enables mounting -one FastMCP server onto another, exposing the mounted server's tools, -resources, and prompts through the parent server with optional prefixing. -""" - -from __future__ import annotations - -import re -from collections.abc import AsyncIterator, Sequence -from contextlib import asynccontextmanager -from typing import TYPE_CHECKING, Any - -from fastmcp.exceptions import NotFoundError -from fastmcp.prompts.prompt import Prompt, PromptResult -from fastmcp.resources.resource import Resource, ResourceContent -from fastmcp.resources.template import ResourceTemplate -from fastmcp.server.providers.base import Provider, TaskComponents -from fastmcp.tools.tool import Tool, ToolResult - -if TYPE_CHECKING: - from fastmcp.prompts.prompt import FunctionPrompt - from fastmcp.resources.resource import FunctionResource - from fastmcp.resources.template import FunctionResourceTemplate - from fastmcp.server.server import FastMCP - from fastmcp.tools.tool import FunctionTool - -# Pattern for matching URIs: protocol://path -_URI_PATTERN = re.compile(r"^([^:]+://)(.*?)$") - - -def add_resource_prefix(uri: str, prefix: str) -> str: - """Add a prefix to a resource URI using path formatting (resource://prefix/path).""" - if not prefix: - return uri - match = _URI_PATTERN.match(uri) - if not match: - raise ValueError(f"Invalid URI format: {uri}. Expected protocol://path format.") - protocol, path = match.groups() - return f"{protocol}{prefix}/{path}" - - -def remove_resource_prefix(uri: str, prefix: str) -> str: - """Remove a prefix from a resource URI.""" - if not prefix: - return uri - match = _URI_PATTERN.match(uri) - if not match: - raise ValueError(f"Invalid URI format: {uri}. Expected protocol://path format.") - protocol, path = match.groups() - prefix_pattern = f"^{re.escape(prefix)}/(.*?)$" - path_match = re.match(prefix_pattern, path) - if not path_match: - return uri - return f"{protocol}{path_match.group(1)}" - - -def has_resource_prefix(uri: str, prefix: str) -> bool: - """Check if a resource URI has a specific prefix.""" - if not prefix: - return False - match = _URI_PATTERN.match(uri) - if not match: - raise ValueError(f"Invalid URI format: {uri}. Expected protocol://path format.") - _, path = match.groups() - prefix_pattern = f"^{re.escape(prefix)}/" - return bool(re.match(prefix_pattern, path)) - - -class MountedProvider(Provider): - """Provider that wraps a mounted FastMCP server. - - This provider enables mounting one FastMCP server onto another, exposing - the mounted server's tools, resources, and prompts through the parent - server with optional prefixing. - - The key benefit is that execution methods (`call_tool`, `read_resource`, - `render_prompt`) invoke the mounted server's middleware chain, enabling - full participation in the provider abstraction. - - Example: - ```python - from fastmcp import FastMCP - from fastmcp.server.providers import MountedProvider - - main = FastMCP("Main") - sub = FastMCP("Sub") - - @sub.tool - def greet(name: str) -> str: - return f"Hello, {name}!" - - # Mount with prefix - tool accessible as "sub_greet" - main.add_provider(MountedProvider(sub, prefix="sub")) - ``` - - Note: - Normally you would use `FastMCP.mount()` which handles proxy conversion - and creates the MountedProvider internally. - """ - - def __init__( - self, - server: FastMCP[Any], - prefix: str | None = None, - tool_names: dict[str, str] | None = None, - ): - """Initialize a MountedProvider. - - Args: - server: The FastMCP server to mount. - prefix: Optional prefix for tool/prompt names and resource URIs. - Tools and prompts use underscore separator: "prefix_name". - Resources use path-style: "protocol://prefix/path". - tool_names: Optional mapping of original tool names to custom names. - Overrides the default prefixed names for specific tools. - """ - super().__init__() - self.server = server - self.prefix = prefix - self.tool_names = tool_names or {} - if len(self.tool_names) != len(set(self.tool_names.values())): - raise ValueError("tool_names values must be unique") - self._reverse_tool_names = {v: k for k, v in self.tool_names.items()} - - # ------------------------------------------------------------------------- - # Helper methods for prefix handling - # ------------------------------------------------------------------------- - - def _add_tool_prefix(self, name: str) -> str: - """Add prefix to a tool or prompt name.""" - if self.prefix: - return f"{self.prefix}_{name}" - return name - - def _strip_tool_prefix(self, name: str) -> str | None: - """Strip prefix from a tool or prompt name. - - Returns: - The unprefixed name if the name matches this provider's pattern, - or None if it doesn't match (indicating another provider should handle it). - """ - # Check for tool_names override first - if name in self._reverse_tool_names: - return self._reverse_tool_names[name] - - # Check prefix pattern - if self.prefix: - expected_prefix = f"{self.prefix}_" - if name.startswith(expected_prefix): - return name[len(expected_prefix) :] - return None # Doesn't match this provider - - # No prefix means we always match - return name - - def _add_resource_prefix(self, uri: str) -> str: - """Add prefix to a resource URI.""" - if not self.prefix: - return uri - return add_resource_prefix(uri, self.prefix) - - def _strip_resource_prefix(self, uri: str) -> str | None: - """Strip prefix from a resource URI. - - Returns: - The unprefixed URI if it matches this provider's pattern, - or None if it doesn't match. - """ - if not self.prefix: - return uri - if not has_resource_prefix(uri, self.prefix): - return None - return remove_resource_prefix(uri, self.prefix) - - # ------------------------------------------------------------------------- - # Prefix helper methods for components - # ------------------------------------------------------------------------- - - def _prefix_tool(self, tool: Tool) -> Tool: - """Apply prefix to a tool. Changes name which updates .key.""" - if self.tool_names and tool.name in self.tool_names: - new_name = self.tool_names[tool.name] - else: - new_name = self._add_tool_prefix(tool.name) - if new_name != tool.name: - return tool.model_copy(update={"name": new_name}) - return tool - - def _prefix_resource(self, resource: Resource) -> Resource: - """Apply prefix to a resource URI (name is NOT prefixed).""" - if self.prefix: - new_uri = self._add_resource_prefix(str(resource.uri)) - if new_uri != str(resource.uri): - return resource.model_copy(update={"uri": new_uri}) - return resource - - def _prefix_template(self, template: ResourceTemplate) -> ResourceTemplate: - """Apply prefix to a resource template URI (name is NOT prefixed).""" - if self.prefix and template.uri_template: - new_template = self._add_resource_prefix(template.uri_template) - if new_template != template.uri_template: - return template.model_copy(update={"uri_template": new_template}) - return template - - def _prefix_prompt(self, prompt: Prompt) -> Prompt: - """Apply prefix to a prompt. Changes name which updates .key.""" - new_name = self._add_tool_prefix(prompt.name) - if new_name != prompt.name: - return prompt.model_copy(update={"name": new_name}) - return prompt - - # ------------------------------------------------------------------------- - # Tool methods - # ------------------------------------------------------------------------- - - async def list_tools(self) -> Sequence[Tool]: - """List all tools from the mounted server with prefixes applied.""" - tools = await self.server._list_tools_middleware() - return [self._prefix_tool(tool) for tool in tools] - - async def get_tool(self, name: str) -> Tool | None: - """Get a tool by name, going through middleware.""" - # Early exit if name doesn't match our prefix pattern - if self._strip_tool_prefix(name) is None: - return None - tools = await self.list_tools() - return next((t for t in tools if t.key == name), None) - - async def call_tool( - self, name: str, arguments: dict[str, Any] - ) -> ToolResult | None: - """Execute a tool through the mounted server's middleware chain.""" - unprefixed = self._strip_tool_prefix(name) - if unprefixed is None: - return None # Doesn't match this provider - - return await self.server._call_tool_middleware(unprefixed, arguments) - - # ------------------------------------------------------------------------- - # Resource methods - # ------------------------------------------------------------------------- - - async def list_resources(self) -> Sequence[Resource]: - """List all resources from the mounted server with prefixes applied.""" - resources = await self.server._list_resources_middleware() - return [self._prefix_resource(resource) for resource in resources] - - async def get_resource(self, uri: str) -> Resource | None: - """Get a concrete resource by URI, going through middleware. - - Only returns concrete resources, not resources created from templates. - This preserves the original template for task execution. - """ - # Early exit if URI doesn't match our prefix pattern - if self._strip_resource_prefix(uri) is None: - return None - - # Only check concrete resources (not templates) - resources = await self.list_resources() - return next((r for r in resources if r.key == uri), None) - - async def read_resource(self, uri: str) -> ResourceContent | None: - """Read a resource through the mounted server's middleware chain.""" - unprefixed = self._strip_resource_prefix(uri) - if unprefixed is None: - return None # Doesn't match this provider - - try: - contents = await self.server._read_resource_middleware(unprefixed) - return contents[0] if contents else None - except NotFoundError: - return None - - # ------------------------------------------------------------------------- - # Resource template methods - # ------------------------------------------------------------------------- - - async def list_resource_templates(self) -> Sequence[ResourceTemplate]: - """List all resource templates from the mounted server with prefixes applied.""" - templates = await self.server._list_resource_templates_middleware() - return [self._prefix_template(template) for template in templates] - - async def get_resource_template(self, uri: str) -> ResourceTemplate | None: - """Get a resource template that matches the given URI.""" - # For templates, we need to check if any template matches the prefixed URI - unprefixed = self._strip_resource_prefix(uri) - if unprefixed is None: - return None - - # Use middleware to include templates from nested mounted providers - templates = await self.server._list_resource_templates_middleware() - for template in templates: - if template.matches(unprefixed) is not None: - return self._prefix_template(template) - return None - - async def read_resource_template(self, uri: str) -> ResourceContent | None: - """Read a resource via a matching template through the mounted server.""" - # This is handled by read_resource since the server's middleware handles templates - return await self.read_resource(uri) - - # ------------------------------------------------------------------------- - # Prompt methods - # ------------------------------------------------------------------------- - - async def list_prompts(self) -> Sequence[Prompt]: - """List all prompts from the mounted server with prefixes applied.""" - prompts = await self.server._list_prompts_middleware() - return [self._prefix_prompt(prompt) for prompt in prompts] - - async def get_prompt(self, name: str) -> Prompt | None: - """Get a prompt by name, going through middleware.""" - # Early exit if name doesn't match our prefix pattern - if self._strip_tool_prefix(name) is None: - return None - prompts = await self.list_prompts() - return next((p for p in prompts if p.key == name), None) - - async def render_prompt( - self, name: str, arguments: dict[str, Any] | None - ) -> PromptResult | None: - """Render a prompt through the mounted server's middleware chain.""" - unprefixed = self._strip_tool_prefix(name) - if unprefixed is None: - return None # Doesn't match this provider - - return await self.server._get_prompt_content_middleware(unprefixed, arguments) - - # ------------------------------------------------------------------------- - # Task registration - # ------------------------------------------------------------------------- - - async def get_tasks(self) -> TaskComponents: - """Return task-eligible components, bypassing middleware and applying prefixes. - - This override accesses the wrapped server's managers directly to avoid - triggering middleware during registration. It also recursively collects - tasks from nested providers. - """ - from fastmcp.prompts.prompt import FunctionPrompt - from fastmcp.resources.resource import FunctionResource - from fastmcp.resources.template import FunctionResourceTemplate - from fastmcp.tools.tool import FunctionTool - - tools: list[FunctionTool] = [] - resources: list[FunctionResource] = [] - templates: list[FunctionResourceTemplate] = [] - prompts: list[FunctionPrompt] = [] - - # Direct manager access (bypasses middleware) - for tool in self.server._tool_manager._tools.values(): - if isinstance(tool, FunctionTool) and tool.task_config.mode != "forbidden": - tools.append(self._prefix_tool(tool)) # type: ignore[arg-type] - - for resource in self.server._resource_manager._resources.values(): - if ( - isinstance(resource, FunctionResource) - and resource.task_config.mode != "forbidden" - ): - resources.append(self._prefix_resource(resource)) # type: ignore[arg-type] - - for template in self.server._resource_manager._templates.values(): - if ( - isinstance(template, FunctionResourceTemplate) - and template.task_config.mode != "forbidden" - ): - templates.append(self._prefix_template(template)) # type: ignore[arg-type] - - for prompt in self.server._prompt_manager._prompts.values(): - if ( - isinstance(prompt, FunctionPrompt) - and prompt.task_config.mode != "forbidden" - ): - prompts.append(self._prefix_prompt(prompt)) # type: ignore[arg-type] - - # Recursively get tasks from nested providers and apply our prefix - for provider in self.server._providers: - nested = await provider.get_tasks() - tools.extend(self._prefix_tool(t) for t in nested.tools) # type: ignore[arg-type] - resources.extend(self._prefix_resource(r) for r in nested.resources) # type: ignore[arg-type] - templates.extend(self._prefix_template(t) for t in nested.templates) # type: ignore[arg-type] - prompts.extend(self._prefix_prompt(p) for p in nested.prompts) # type: ignore[arg-type] - - return TaskComponents( - tools=tools, resources=resources, templates=templates, prompts=prompts - ) - - # ------------------------------------------------------------------------- - # Lifecycle methods - # ------------------------------------------------------------------------- - - @asynccontextmanager - async def lifespan(self) -> AsyncIterator[None]: - """Start the mounted server's user lifespan. - - This starts only the wrapped server's user-defined lifespan, NOT its - full _lifespan_manager() (which includes Docket). The parent server's - Docket handles all background tasks. - """ - # Start the wrapped server's user lifespan only - # We pass the server instance to the user's lifespan function - async with self.server._lifespan(self.server): - yield diff --git a/src/fastmcp/server/providers/transforming.py b/src/fastmcp/server/providers/transforming.py new file mode 100644 index 0000000000..0d45f5a743 --- /dev/null +++ b/src/fastmcp/server/providers/transforming.py @@ -0,0 +1,349 @@ +"""TransformingProvider for applying component transformations. + +This module provides the `TransformingProvider` class that wraps any Provider +and applies transformations like namespace prefixes and tool renames. +""" + +from __future__ import annotations + +import re +from collections.abc import AsyncIterator, Sequence +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING, Any + +from fastmcp.prompts.prompt import Prompt, PromptResult +from fastmcp.resources.resource import Resource, ResourceContent +from fastmcp.resources.template import ResourceTemplate +from fastmcp.server.providers.base import Provider, TaskComponents +from fastmcp.tools.tool import Tool, ToolResult + +if TYPE_CHECKING: + from fastmcp.prompts.prompt import FunctionPrompt + from fastmcp.resources.resource import FunctionResource + from fastmcp.resources.template import FunctionResourceTemplate + from fastmcp.tools.tool import FunctionTool + + +# Pattern for matching URIs: protocol://path +_URI_PATTERN = re.compile(r"^([^:]+://)(.*?)$") + + +class TransformingProvider(Provider): + """Wraps any provider and applies component transformations. + + Users typically use `provider.with_transforms()` rather than instantiating + this class directly. Multiple `.with_transforms()` calls stack - each + creates a new wrapper that composes with the previous. + + Transformation rules: + - Tools/prompts with explicit renames use the rename (bypasses namespace) + - Tools/prompts without renames get namespace prefix: "namespace_name" + - Resources get path-style namespace: "protocol://namespace/path" + + Example: + ```python + # Via with_transforms() method (preferred) + provider = SomeProvider().with_transforms( + namespace="api", + tool_renames={"verbose_tool_name": "short"} + ) + + # Stacking composes transformations: + provider = ( + SomeProvider() + .with_transforms(namespace="api") + .with_transforms(tool_renames={"api_foo": "bar"}) + ) + # "foo" → "api_foo" (inner) → "bar" (outer) + ``` + """ + + def __init__( + self, + provider: Provider, + *, + namespace: str | None = None, + tool_renames: dict[str, str] | None = None, + ): + """Initialize a TransformingProvider. + + Args: + provider: The provider to wrap. + namespace: Prefix for tools/prompts, path segment for resources. + tool_renames: Map of original_name → final_name. Tools in this map + use the specified name instead of namespace prefixing. + """ + super().__init__() + self._wrapped = provider + self.namespace = namespace + self.tool_renames = tool_renames or {} + + # Validate that renames are reversible (no duplicate target names) + if len(self.tool_renames) != len(set(self.tool_renames.values())): + seen: dict[str, str] = {} + for orig, renamed in self.tool_renames.items(): + if renamed in seen: + raise ValueError( + f"tool_renames has duplicate target name {renamed!r}: " + f"both {seen[renamed]!r} and {orig!r} map to it" + ) + seen[renamed] = orig + + self._tool_renames_reverse = {v: k for k, v in self.tool_renames.items()} + + # ------------------------------------------------------------------------- + # Tool name transformation + # ------------------------------------------------------------------------- + + def _transform_tool_name(self, name: str) -> str: + """Apply transformation to tool name.""" + # Explicit rename takes precedence (bypasses namespace) + if name in self.tool_renames: + return self.tool_renames[name] + # Otherwise apply namespace + if self.namespace: + return f"{self.namespace}_{name}" + return name + + def _reverse_tool_name(self, name: str) -> str | None: + """Reverse tool name transformation, or None if no match.""" + # Check explicit renames first + if name in self._tool_renames_reverse: + return self._tool_renames_reverse[name] + # Check namespace prefix + if self.namespace: + prefix = f"{self.namespace}_" + if name.startswith(prefix): + return name[len(prefix) :] + return None + return name + + # ------------------------------------------------------------------------- + # Prompt name transformation + # ------------------------------------------------------------------------- + + def _transform_prompt_name(self, name: str) -> str: + """Apply transformation to prompt name.""" + if self.namespace: + return f"{self.namespace}_{name}" + return name + + def _reverse_prompt_name(self, name: str) -> str | None: + """Reverse prompt name transformation, or None if no match.""" + if self.namespace: + prefix = f"{self.namespace}_" + if name.startswith(prefix): + return name[len(prefix) :] + return None + return name + + # ------------------------------------------------------------------------- + # Resource URI transformation + # ------------------------------------------------------------------------- + + def _transform_resource_uri(self, uri: str) -> str: + """Apply transformation to resource URI.""" + if not self.namespace: + return uri + match = _URI_PATTERN.match(uri) + if match: + protocol, path = match.groups() + return f"{protocol}{self.namespace}/{path}" + return uri + + def _reverse_resource_uri(self, uri: str) -> str | None: + """Reverse resource URI transformation, or None if no match.""" + if not self.namespace: + return uri + match = _URI_PATTERN.match(uri) + if match: + protocol, path = match.groups() + prefix = f"{self.namespace}/" + if path.startswith(prefix): + return f"{protocol}{path[len(prefix) :]}" + return None + return None + + # ------------------------------------------------------------------------- + # Tool methods + # ------------------------------------------------------------------------- + + async def list_tools(self) -> Sequence[Tool]: + """List tools with transformations applied.""" + tools = await self._wrapped.list_tools() + return [ + t.model_copy(update={"name": self._transform_tool_name(t.name)}) + for t in tools + ] + + async def get_tool(self, name: str) -> Tool | None: + """Get tool by transformed name.""" + original = self._reverse_tool_name(name) + if original is None: + return None + tool = await self._wrapped.get_tool(original) + if tool: + return tool.model_copy(update={"name": name}) + return None + + async def call_tool( + self, name: str, arguments: dict[str, Any] + ) -> ToolResult | None: + """Call tool by transformed name.""" + original = self._reverse_tool_name(name) + if original is None: + return None + return await self._wrapped.call_tool(original, arguments) + + # ------------------------------------------------------------------------- + # Resource methods + # ------------------------------------------------------------------------- + + async def list_resources(self) -> Sequence[Resource]: + """List resources with URI transformations applied.""" + resources = await self._wrapped.list_resources() + return [ + r.model_copy(update={"uri": self._transform_resource_uri(str(r.uri))}) + for r in resources + ] + + async def get_resource(self, uri: str) -> Resource | None: + """Get resource by transformed URI.""" + original = self._reverse_resource_uri(uri) + if original is None: + return None + resource = await self._wrapped.get_resource(original) + if resource: + return resource.model_copy(update={"uri": uri}) + return None + + async def read_resource(self, uri: str) -> ResourceContent | None: + """Read resource by transformed URI.""" + original = self._reverse_resource_uri(uri) + if original is None: + return None + return await self._wrapped.read_resource(original) + + # ------------------------------------------------------------------------- + # Resource template methods + # ------------------------------------------------------------------------- + + async def list_resource_templates(self) -> Sequence[ResourceTemplate]: + """List resource templates with URI transformations applied.""" + templates = await self._wrapped.list_resource_templates() + return [ + t.model_copy( + update={"uri_template": self._transform_resource_uri(t.uri_template)} + ) + for t in templates + ] + + async def get_resource_template(self, uri: str) -> ResourceTemplate | None: + """Get resource template by transformed URI.""" + original = self._reverse_resource_uri(uri) + if original is None: + return None + template = await self._wrapped.get_resource_template(original) + if template: + return template.model_copy( + update={ + "uri_template": self._transform_resource_uri(template.uri_template) + } + ) + return None + + async def read_resource_template(self, uri: str) -> ResourceContent | None: + """Read resource template by transformed URI.""" + original = self._reverse_resource_uri(uri) + if original is None: + return None + return await self._wrapped.read_resource_template(original) + + # ------------------------------------------------------------------------- + # Prompt methods + # ------------------------------------------------------------------------- + + async def list_prompts(self) -> Sequence[Prompt]: + """List prompts with transformations applied.""" + prompts = await self._wrapped.list_prompts() + return [ + p.model_copy(update={"name": self._transform_prompt_name(p.name)}) + for p in prompts + ] + + async def get_prompt(self, name: str) -> Prompt | None: + """Get prompt by transformed name.""" + original = self._reverse_prompt_name(name) + if original is None: + return None + prompt = await self._wrapped.get_prompt(original) + if prompt: + return prompt.model_copy(update={"name": name}) + return None + + async def render_prompt( + self, name: str, arguments: dict[str, Any] | None + ) -> PromptResult | None: + """Render prompt by transformed name.""" + original = self._reverse_prompt_name(name) + if original is None: + return None + return await self._wrapped.render_prompt(original, arguments) + + # ------------------------------------------------------------------------- + # Task registration + # ------------------------------------------------------------------------- + + async def get_tasks(self) -> TaskComponents: + """Get tasks with transformations applied to all components.""" + + tasks = await self._wrapped.get_tasks() + + # Apply transforms to tools + transformed_tools: list[FunctionTool] = [] + for t in tasks.tools: + transformed_tools.append( + t.model_copy(update={"name": self._transform_tool_name(t.name)}) # type: ignore[arg-type] + ) + + # Apply transforms to resources + transformed_resources: list[FunctionResource] = [] + for r in tasks.resources: + transformed_resources.append( + r.model_copy(update={"uri": self._transform_resource_uri(str(r.uri))}) # type: ignore[arg-type] + ) + + # Apply transforms to templates + transformed_templates: list[FunctionResourceTemplate] = [] + for t in tasks.templates: + transformed_templates.append( + t.model_copy( + update={ + "uri_template": self._transform_resource_uri(t.uri_template) + } + ) # type: ignore[arg-type] + ) + + # Apply transforms to prompts + transformed_prompts: list[FunctionPrompt] = [] + for p in tasks.prompts: + transformed_prompts.append( + p.model_copy(update={"name": self._transform_prompt_name(p.name)}) # type: ignore[arg-type] + ) + + return TaskComponents( + tools=transformed_tools, + resources=transformed_resources, + templates=transformed_templates, + prompts=transformed_prompts, + ) + + # ------------------------------------------------------------------------- + # Lifecycle + # ------------------------------------------------------------------------- + + @asynccontextmanager + async def lifespan(self) -> AsyncIterator[None]: + """Delegate lifespan to wrapped provider.""" + async with self._wrapped.lifespan(): + yield diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py index fdabb4e957..8a4959f59d 100644 --- a/src/fastmcp/server/server.py +++ b/src/fastmcp/server/server.py @@ -849,7 +849,7 @@ async def get_tools(self) -> dict[str, Tool]: """Get all tools (unfiltered), including from providers, indexed by key.""" all_tools = dict(await self._tool_manager.get_tools()) - # Get tools from providers (including MountedProvider) + # Get tools from providers (including FastMCPProvider) for provider in self._providers: try: provider_tools = await provider.list_tools() @@ -939,7 +939,7 @@ async def get_resources(self) -> dict[str, Resource]: """Get all resources (unfiltered), including from providers, indexed by key.""" all_resources = dict(await self._resource_manager.get_resources()) - # Get resources from providers (including MountedProvider) + # Get resources from providers (including FastMCPProvider) for provider in self._providers: try: provider_resources = await provider.list_resources() @@ -979,7 +979,7 @@ async def get_resource_templates(self) -> dict[str, ResourceTemplate]: """Get all resource templates (unfiltered), including from providers, indexed by key.""" all_templates = dict(await self._resource_manager.get_resource_templates()) - # Get templates from providers (including MountedProvider) + # Get templates from providers (including FastMCPProvider) for provider in self._providers: try: provider_templates = await provider.list_resource_templates() @@ -1019,7 +1019,7 @@ async def get_prompts(self) -> dict[str, Prompt]: """Get all prompts (unfiltered), including from providers, indexed by key.""" all_prompts = dict(await self._prompt_manager.get_prompts()) - # Get prompts from providers (including MountedProvider) + # Get prompts from providers (including FastMCPProvider) for provider in self._providers: try: provider_prompts = await provider.list_prompts() @@ -1109,7 +1109,7 @@ def _get_additional_http_routes(self) -> list[BaseRoute]: """Get all additional HTTP routes including from providers. Returns a list of all custom HTTP routes from this server and - from all providers that have HTTP routes (e.g., MountedProvider). + from all providers that have HTTP routes (e.g., FastMCPProvider). Returns: List of Starlette BaseRoute objects @@ -2638,11 +2638,12 @@ def http_app( def mount( self, server: FastMCP[LifespanResultT], - prefix: str | None = None, + namespace: str | None = None, as_proxy: bool | None = None, tool_names: dict[str, str] | None = None, + prefix: str | None = None, # deprecated, use namespace ) -> None: - """Mount another FastMCP server on this server with an optional prefix. + """Mount another FastMCP server on this server with an optional namespace. Unlike importing (with import_server), mounting establishes a dynamic connection between servers. When a client interacts with a mounted server's objects through @@ -2650,40 +2651,53 @@ def mount( This means changes to the mounted server are immediately reflected when accessed through the parent. - When a server is mounted with a prefix: - - Tools from the mounted server are accessible with prefixed names. - Example: If server has a tool named "get_weather", it will be available as "prefix_get_weather". - - Resources are accessible with prefixed URIs. + When a server is mounted with a namespace: + - Tools from the mounted server are accessible with namespaced names. + Example: If server has a tool named "get_weather", it will be available as "namespace_get_weather". + - Resources are accessible with namespaced URIs. Example: If server has a resource with URI "weather://forecast", it will be available as - "weather://prefix/forecast". - - Templates are accessible with prefixed URI templates. + "weather://namespace/forecast". + - Templates are accessible with namespaced URI templates. Example: If server has a template with URI "weather://location/{id}", it will be available - as "weather://prefix/location/{id}". - - Prompts are accessible with prefixed names. + as "weather://namespace/location/{id}". + - Prompts are accessible with namespaced names. Example: If server has a prompt named "weather_prompt", it will be available as - "prefix_weather_prompt". + "namespace_weather_prompt". - When a server is mounted without a prefix (prefix=None), its tools, resources, templates, + When a server is mounted without a namespace (namespace=None), its tools, resources, templates, and prompts are accessible with their original names. Multiple servers can be mounted - without prefixes, and they will be tried in order until a match is found. + without namespaces, and they will be tried in order until a match is found. The mounted server's lifespan is executed when the parent server starts, and its middleware chain is invoked for all operations (tool calls, resource reads, prompts). Args: server: The FastMCP server to mount. - prefix: Optional prefix to use for the mounted server's objects. If None, + namespace: Optional namespace to use for the mounted server's objects. If None, the server's objects are accessible with their original names. as_proxy: Deprecated. Mounted servers now always have their lifespan and middleware invoked. To create a proxy server, use FastMCP.as_proxy() explicitly before mounting. tool_names: Optional mapping of original tool names to custom names. Use this - to override prefixed names. Keys are the original tool names from the + to override namespaced names. Keys are the original tool names from the mounted server. + prefix: Deprecated. Use namespace instead. """ import warnings - from fastmcp.server.providers import MountedProvider + from fastmcp.server.providers.fastmcp_provider import FastMCPProvider + + # Handle deprecated prefix parameter + if prefix is not None: + warnings.warn( + "The 'prefix' parameter is deprecated, use 'namespace' instead", + DeprecationWarning, + stacklevel=2, + ) + if namespace is None: + namespace = prefix + else: + raise ValueError("Cannot specify both 'prefix' and 'namespace'") if as_proxy is not None: warnings.warn( @@ -2700,8 +2714,12 @@ def mount( if not isinstance(server, FastMCPProxy): server = FastMCP.as_proxy(server) - # Create a MountedProvider and add it to providers - provider = MountedProvider(server, prefix, tool_names) + # Create provider with optional transformations + provider: Provider = FastMCPProvider(server) + if namespace or tool_names: + provider = provider.with_transforms( + namespace=namespace, tool_renames=tool_names + ) self._providers.append(provider) async def import_server( @@ -2746,14 +2764,20 @@ async def import_server( """ import warnings - from fastmcp.server.providers.mounted import add_resource_prefix - warnings.warn( "import_server is deprecated, use mount() instead", DeprecationWarning, stacklevel=2, ) + def add_resource_prefix(uri: str, prefix: str) -> str: + """Add prefix to resource URI: protocol://path → protocol://prefix/path.""" + match = URI_PATTERN.match(uri) + if match: + protocol, path = match.groups() + return f"{protocol}{prefix}/{path}" + return uri + # Import tools from the server for tool in (await server.get_tools()).values(): if prefix: diff --git a/tests/server/providers/__init__.py b/tests/server/providers/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/server/providers/test_fastmcp_provider.py b/tests/server/providers/test_fastmcp_provider.py new file mode 100644 index 0000000000..11b0cb9242 --- /dev/null +++ b/tests/server/providers/test_fastmcp_provider.py @@ -0,0 +1,233 @@ +"""Tests for FastMCPProvider.""" + +from fastmcp import FastMCP +from fastmcp.client import Client +from fastmcp.server.providers import FastMCPProvider + + +class TestToolOperations: + """Test tool operations through FastMCPProvider.""" + + async def test_list_tools(self): + """Test listing tools from wrapped server.""" + server = FastMCP("Test") + + @server.tool + def tool_one() -> str: + return "one" + + @server.tool + def tool_two() -> str: + return "two" + + provider = FastMCPProvider(server) + tools = await provider.list_tools() + + assert len(tools) == 2 + names = {t.name for t in tools} + assert names == {"tool_one", "tool_two"} + + async def test_get_tool(self): + """Test getting a specific tool by name.""" + server = FastMCP("Test") + + @server.tool + def my_tool() -> str: + return "result" + + provider = FastMCPProvider(server) + tool = await provider.get_tool("my_tool") + + assert tool is not None + assert tool.name == "my_tool" + + async def test_get_nonexistent_tool_returns_none(self): + """Test that getting a nonexistent tool returns None.""" + server = FastMCP("Test") + provider = FastMCPProvider(server) + + tool = await provider.get_tool("nonexistent") + assert tool is None + + async def test_call_tool_via_client(self): + """Test calling a tool through a server using the provider.""" + sub = FastMCP("Sub") + + @sub.tool + def greet(name: str) -> str: + return f"Hello, {name}!" + + main = FastMCP("Main") + main._providers.append(FastMCPProvider(sub)) + + async with Client(main) as client: + result = await client.call_tool("greet", {"name": "World"}) + assert result.data == "Hello, World!" + + +class TestResourceOperations: + """Test resource operations through FastMCPProvider.""" + + async def test_list_resources(self): + """Test listing resources from wrapped server.""" + server = FastMCP("Test") + + @server.resource("resource://one") + def resource_one() -> str: + return "one" + + @server.resource("resource://two") + def resource_two() -> str: + return "two" + + provider = FastMCPProvider(server) + resources = await provider.list_resources() + + assert len(resources) == 2 + uris = {str(r.uri) for r in resources} + assert uris == {"resource://one", "resource://two"} + + async def test_get_resource(self): + """Test getting a specific resource by URI.""" + server = FastMCP("Test") + + @server.resource("resource://data") + def my_resource() -> str: + return "content" + + provider = FastMCPProvider(server) + resource = await provider.get_resource("resource://data") + + assert resource is not None + assert str(resource.uri) == "resource://data" + + async def test_read_resource_via_client(self): + """Test reading a resource through a server using the provider.""" + sub = FastMCP("Sub") + + @sub.resource("resource://data") + def my_resource() -> str: + return "content" + + main = FastMCP("Main") + main._providers.append(FastMCPProvider(sub)) + + async with Client(main) as client: + result = await client.read_resource("resource://data") + assert result[0].text == "content" # type: ignore[attr-defined] + + +class TestResourceTemplateOperations: + """Test resource template operations through FastMCPProvider.""" + + async def test_list_resource_templates(self): + """Test listing resource templates from wrapped server.""" + server = FastMCP("Test") + + @server.resource("resource://{id}/data") + def my_template(id: str) -> str: + return f"data for {id}" + + provider = FastMCPProvider(server) + templates = await provider.list_resource_templates() + + assert len(templates) == 1 + assert templates[0].uri_template == "resource://{id}/data" + + async def test_get_resource_template(self): + """Test getting a template that matches a URI.""" + server = FastMCP("Test") + + @server.resource("resource://{id}/data") + def my_template(id: str) -> str: + return f"data for {id}" + + provider = FastMCPProvider(server) + template = await provider.get_resource_template("resource://123/data") + + assert template is not None + + async def test_read_resource_template_via_client(self): + """Test reading a resource via template through a server using the provider.""" + sub = FastMCP("Sub") + + @sub.resource("resource://{id}/data") + def my_template(id: str) -> str: + return f"data for {id}" + + main = FastMCP("Main") + main._providers.append(FastMCPProvider(sub)) + + async with Client(main) as client: + result = await client.read_resource("resource://123/data") + assert result[0].text == "data for 123" # type: ignore[attr-defined] + + +class TestPromptOperations: + """Test prompt operations through FastMCPProvider.""" + + async def test_list_prompts(self): + """Test listing prompts from wrapped server.""" + server = FastMCP("Test") + + @server.prompt + def prompt_one() -> str: + return "one" + + @server.prompt + def prompt_two() -> str: + return "two" + + provider = FastMCPProvider(server) + prompts = await provider.list_prompts() + + assert len(prompts) == 2 + names = {p.name for p in prompts} + assert names == {"prompt_one", "prompt_two"} + + async def test_get_prompt(self): + """Test getting a specific prompt by name.""" + server = FastMCP("Test") + + @server.prompt + def my_prompt() -> str: + return "content" + + provider = FastMCPProvider(server) + prompt = await provider.get_prompt("my_prompt") + + assert prompt is not None + assert prompt.name == "my_prompt" + + async def test_render_prompt_via_client(self): + """Test rendering a prompt through a server using the provider.""" + sub = FastMCP("Sub") + + @sub.prompt + def greet(name: str) -> str: + return f"Hello, {name}!" + + main = FastMCP("Main") + main._providers.append(FastMCPProvider(sub)) + + async with Client(main) as client: + result = await client.get_prompt("greet", {"name": "World"}) + assert result.messages[0].content.text == "Hello, World!" # type: ignore[attr-defined] + + +class TestServerReference: + """Test that provider maintains reference to wrapped server.""" + + def test_server_attribute(self): + """Test that provider exposes the wrapped server.""" + server = FastMCP("Test") + provider = FastMCPProvider(server) + + assert provider.server is server + + def test_server_name_accessible(self): + """Test that server name is accessible through provider.""" + server = FastMCP("MyServer") + provider = FastMCPProvider(server) + + assert provider.server.name == "MyServer" diff --git a/tests/server/providers/test_transforming_provider.py b/tests/server/providers/test_transforming_provider.py new file mode 100644 index 0000000000..0781db637d --- /dev/null +++ b/tests/server/providers/test_transforming_provider.py @@ -0,0 +1,285 @@ +"""Tests for TransformingProvider.""" + +import pytest + +from fastmcp import FastMCP +from fastmcp.client import Client +from fastmcp.server.providers import FastMCPProvider, TransformingProvider + + +class TestNamespaceTransformation: + """Test namespace prefix transformations.""" + + async def test_namespace_prefixes_tool_names(self): + """Test that namespace is applied as prefix to tool names.""" + server = FastMCP("Test") + + @server.tool + def my_tool() -> str: + return "result" + + provider = FastMCPProvider(server).with_namespace("ns") + tools = await provider.list_tools() + + assert len(tools) == 1 + assert tools[0].name == "ns_my_tool" + + async def test_namespace_prefixes_prompt_names(self): + """Test that namespace is applied as prefix to prompt names.""" + server = FastMCP("Test") + + @server.prompt + def my_prompt() -> str: + return "prompt content" + + provider = FastMCPProvider(server).with_namespace("ns") + prompts = await provider.list_prompts() + + assert len(prompts) == 1 + assert prompts[0].name == "ns_my_prompt" + + async def test_namespace_prefixes_resource_uris(self): + """Test that namespace is inserted into resource URIs.""" + server = FastMCP("Test") + + @server.resource("resource://data") + def my_resource() -> str: + return "content" + + provider = FastMCPProvider(server).with_namespace("ns") + resources = await provider.list_resources() + + assert len(resources) == 1 + assert str(resources[0].uri) == "resource://ns/data" + + async def test_namespace_prefixes_template_uris(self): + """Test that namespace is inserted into resource template URIs.""" + server = FastMCP("Test") + + @server.resource("resource://{name}/data") + def my_template(name: str) -> str: + return f"content for {name}" + + provider = FastMCPProvider(server).with_namespace("ns") + templates = await provider.list_resource_templates() + + assert len(templates) == 1 + assert templates[0].uri_template == "resource://ns/{name}/data" + + +class TestToolRenames: + """Test tool renaming functionality.""" + + async def test_tool_rename_bypasses_namespace(self): + """Test that explicit renames bypass namespace prefixing.""" + server = FastMCP("Test") + + @server.tool + def verbose_tool_name() -> str: + return "result" + + provider = FastMCPProvider(server).with_transforms( + namespace="ns", + tool_renames={"verbose_tool_name": "short"}, + ) + tools = await provider.list_tools() + + assert len(tools) == 1 + assert tools[0].name == "short" + + async def test_tool_rename_without_namespace(self): + """Test tool renaming works without namespace.""" + server = FastMCP("Test") + + @server.tool + def old_name() -> str: + return "result" + + provider = FastMCPProvider(server).with_transforms( + tool_renames={"old_name": "new_name"}, + ) + tools = await provider.list_tools() + + assert len(tools) == 1 + assert tools[0].name == "new_name" + + async def test_renamed_tool_is_callable(self): + """Test that renamed tools can be called by new name.""" + sub = FastMCP("Sub") + + @sub.tool + def original() -> str: + return "success" + + main = FastMCP("Main") + main._providers.append( + FastMCPProvider(sub).with_transforms( + tool_renames={"original": "renamed"}, + ) + ) + + async with Client(main) as client: + result = await client.call_tool("renamed", {}) + assert result.data == "success" + + async def test_duplicate_rename_targets_raises_error(self): + """Test that duplicate target names in tool_renames raises ValueError.""" + server = FastMCP("Test") + base_provider = FastMCPProvider(server) + + with pytest.raises(ValueError, match="duplicate target name"): + base_provider.with_transforms( + tool_renames={"tool_a": "same", "tool_b": "same"}, + ) + + +class TestReverseTransformation: + """Test reverse lookups for routing.""" + + async def test_reverse_tool_lookup_with_namespace(self): + """Test that tools can be looked up by transformed name.""" + server = FastMCP("Test") + + @server.tool + def my_tool() -> str: + return "result" + + provider = FastMCPProvider(server).with_namespace("ns") + tool = await provider.get_tool("ns_my_tool") + + assert tool is not None + assert tool.name == "ns_my_tool" + + async def test_reverse_tool_lookup_with_rename(self): + """Test that renamed tools can be looked up by new name.""" + server = FastMCP("Test") + + @server.tool + def original() -> str: + return "result" + + provider = FastMCPProvider(server).with_transforms( + tool_renames={"original": "renamed"}, + ) + tool = await provider.get_tool("renamed") + + assert tool is not None + assert tool.name == "renamed" + + async def test_reverse_resource_lookup_with_namespace(self): + """Test that resources can be looked up by transformed URI.""" + server = FastMCP("Test") + + @server.resource("resource://data") + def my_resource() -> str: + return "content" + + provider = FastMCPProvider(server).with_namespace("ns") + resource = await provider.get_resource("resource://ns/data") + + assert resource is not None + assert str(resource.uri) == "resource://ns/data" + + async def test_nonmatching_namespace_returns_none(self): + """Test that lookups with wrong namespace return None.""" + server = FastMCP("Test") + + @server.tool + def my_tool() -> str: + return "result" + + provider = FastMCPProvider(server).with_namespace("ns") + + # Wrong namespace prefix + assert await provider.get_tool("wrong_my_tool") is None + # No prefix at all + assert await provider.get_tool("my_tool") is None + + +class TestTransformStacking: + """Test stacking multiple transformations.""" + + async def test_stacked_namespaces_compose(self): + """Test that stacked namespaces are applied in order.""" + server = FastMCP("Test") + + @server.tool + def my_tool() -> str: + return "result" + + provider = ( + FastMCPProvider(server).with_namespace("inner").with_namespace("outer") + ) + tools = await provider.list_tools() + + assert len(tools) == 1 + assert tools[0].name == "outer_inner_my_tool" + + async def test_stacked_rename_after_namespace(self): + """Test renaming a namespaced tool.""" + server = FastMCP("Test") + + @server.tool + def my_tool() -> str: + return "result" + + provider = ( + FastMCPProvider(server) + .with_namespace("ns") + .with_transforms(tool_renames={"ns_my_tool": "short"}) + ) + tools = await provider.list_tools() + + assert len(tools) == 1 + assert tools[0].name == "short" + + async def test_stacked_transforms_are_callable(self): + """Test that stacked transforms still allow tool calls.""" + sub = FastMCP("Sub") + + @sub.tool + def my_tool() -> str: + return "success" + + main = FastMCP("Main") + main._providers.append( + FastMCPProvider(sub) + .with_namespace("ns") + .with_transforms(tool_renames={"ns_my_tool": "short"}) + ) + + async with Client(main) as client: + result = await client.call_tool("short", {}) + assert result.data == "success" + + +class TestNoTransformation: + """Test behavior when no transformations are applied.""" + + async def test_no_namespace_passthrough(self): + """Test that tools pass through unchanged without namespace.""" + server = FastMCP("Test") + + @server.tool + def my_tool() -> str: + return "result" + + provider = TransformingProvider(FastMCPProvider(server)) + tools = await provider.list_tools() + + assert len(tools) == 1 + assert tools[0].name == "my_tool" + + async def test_empty_tool_renames_passthrough(self): + """Test that empty tool_renames has no effect.""" + server = FastMCP("Test") + + @server.tool + def my_tool() -> str: + return "result" + + provider = FastMCPProvider(server).with_transforms(tool_renames={}) + tools = await provider.list_tools() + + assert len(tools) == 1 + assert tools[0].name == "my_tool" diff --git a/tests/server/test_mount.py b/tests/server/test_mount.py index b52b2ed172..947171b6b1 100644 --- a/tests/server/test_mount.py +++ b/tests/server/test_mount.py @@ -7,7 +7,7 @@ from fastmcp import FastMCP from fastmcp.client import Client from fastmcp.client.transports import FastMCPTransport, SSETransport -from fastmcp.server.providers import MountedProvider +from fastmcp.server.providers import FastMCPProvider, TransformingProvider from fastmcp.server.proxy import FastMCPProxy from fastmcp.tools.tool import Tool from fastmcp.tools.tool_transform import TransformedTool @@ -852,8 +852,10 @@ async def test_as_proxy_defaults_false(self): mcp.mount(sub, "sub") provider = mcp._providers[0] - assert isinstance(provider, MountedProvider) - assert provider.server is sub + # With namespace, we get TransformingProvider wrapping FastMCPProvider + assert isinstance(provider, TransformingProvider) + assert isinstance(provider._wrapped, FastMCPProvider) + assert provider._wrapped.server is sub async def test_as_proxy_false(self): mcp = FastMCP("Main") @@ -862,8 +864,10 @@ async def test_as_proxy_false(self): mcp.mount(sub, "sub", as_proxy=False) provider = mcp._providers[0] - assert isinstance(provider, MountedProvider) - assert provider.server is sub + # With namespace, we get TransformingProvider wrapping FastMCPProvider + assert isinstance(provider, TransformingProvider) + assert isinstance(provider._wrapped, FastMCPProvider) + assert provider._wrapped.server is sub async def test_as_proxy_true(self): mcp = FastMCP("Main") @@ -872,14 +876,16 @@ async def test_as_proxy_true(self): mcp.mount(sub, "sub", as_proxy=True) provider = mcp._providers[0] - assert isinstance(provider, MountedProvider) - assert provider.server is not sub - assert isinstance(provider.server, FastMCPProxy) + # With namespace, we get TransformingProvider wrapping FastMCPProvider + assert isinstance(provider, TransformingProvider) + assert isinstance(provider._wrapped, FastMCPProvider) + assert provider._wrapped.server is not sub + assert isinstance(provider._wrapped.server, FastMCPProxy) async def test_lifespan_server_mounted_directly(self): """Test that servers with lifespan are mounted directly (not auto-proxied). - Since MountedProvider now handles lifespan via the provider lifespan interface, + Since FastMCPProvider now handles lifespan via the provider lifespan interface, there's no need to auto-convert to a proxy. The server is mounted directly. """ @@ -894,8 +900,9 @@ async def server_lifespan(mcp: FastMCP): # Server should be mounted directly without auto-proxying provider = mcp._providers[0] - assert isinstance(provider, MountedProvider) - assert provider.server is sub + assert isinstance(provider, TransformingProvider) + assert isinstance(provider._wrapped, FastMCPProvider) + assert provider._wrapped.server is sub async def test_as_proxy_ignored_for_proxy_mounts_default(self): mcp = FastMCP("Main") @@ -905,8 +912,9 @@ async def test_as_proxy_ignored_for_proxy_mounts_default(self): mcp.mount(sub_proxy, "sub") provider = mcp._providers[0] - assert isinstance(provider, MountedProvider) - assert provider.server is sub_proxy + assert isinstance(provider, TransformingProvider) + assert isinstance(provider._wrapped, FastMCPProvider) + assert provider._wrapped.server is sub_proxy async def test_as_proxy_ignored_for_proxy_mounts_false(self): mcp = FastMCP("Main") @@ -916,8 +924,9 @@ async def test_as_proxy_ignored_for_proxy_mounts_false(self): mcp.mount(sub_proxy, "sub", as_proxy=False) provider = mcp._providers[0] - assert isinstance(provider, MountedProvider) - assert provider.server is sub_proxy + assert isinstance(provider, TransformingProvider) + assert isinstance(provider._wrapped, FastMCPProvider) + assert provider._wrapped.server is sub_proxy async def test_as_proxy_ignored_for_proxy_mounts_true(self): mcp = FastMCP("Main") @@ -927,8 +936,9 @@ async def test_as_proxy_ignored_for_proxy_mounts_true(self): mcp.mount(sub_proxy, "sub", as_proxy=True) provider = mcp._providers[0] - assert isinstance(provider, MountedProvider) - assert provider.server is sub_proxy + assert isinstance(provider, TransformingProvider) + assert isinstance(provider._wrapped, FastMCPProvider) + assert provider._wrapped.server is sub_proxy async def test_as_proxy_mounts_still_have_live_link(self): mcp = FastMCP("Main") @@ -1153,17 +1163,19 @@ async def test_mounted_servers_tracking(self): main_server.mount(sub_server1, "sub1") assert len(main_server._providers) == 1 provider1 = main_server._providers[0] - assert isinstance(provider1, MountedProvider) - assert provider1.server == sub_server1 - assert provider1.prefix == "sub1" + assert isinstance(provider1, TransformingProvider) + assert isinstance(provider1._wrapped, FastMCPProvider) + assert provider1._wrapped.server == sub_server1 + assert provider1.namespace == "sub1" # Mount second server main_server.mount(sub_server2, "sub2") assert len(main_server._providers) == 2 provider2 = main_server._providers[1] - assert isinstance(provider2, MountedProvider) - assert provider2.server == sub_server2 - assert provider2.prefix == "sub2" + assert isinstance(provider2, TransformingProvider) + assert isinstance(provider2._wrapped, FastMCPProvider) + assert provider2._wrapped.server == sub_server2 + assert provider2.namespace == "sub2" async def test_multiple_routes_same_server(self): """Test that multiple custom routes from same server are all included.""" @@ -1331,8 +1343,13 @@ def deep_tool() -> str: class TestToolNameOverrides: """Test tool and prompt name overrides in mount() (issue #2596).""" - async def test_tool_names_override_applied_in_get_tools(self): - """Test that tool_names override is reflected in get_tools().""" + async def test_tool_names_override_via_transforms(self): + """Test that tool_names renames tools via TransformingProvider. + + With TransformingProvider, tool_renames are applied to the original name + and bypass namespace prefixing. Both server introspection and client-facing + API show the transformed names consistently. + """ sub = FastMCP("Sub") @sub.tool @@ -1340,16 +1357,26 @@ def original_tool() -> str: return "test" main = FastMCP("Main") + # tool_names maps original name → final name (bypasses namespace) main.mount( sub, - prefix="prefix", + namespace="prefix", tool_names={"original_tool": "custom_name"}, ) + # Server introspection shows transformed names tools = await main.get_tools() assert "custom_name" in tools + assert "original_tool" not in tools assert "prefix_original_tool" not in tools + # Client-facing API shows the same transformed names + async with Client(main) as client: + client_tools = await client.list_tools() + tool_names = [t.name for t in client_tools] + assert "custom_name" in tool_names + assert "prefix_original_tool" not in tool_names + async def test_tool_names_override_applied_in_list_tools(self): """Test that tool_names override is reflected in list_tools().""" sub = FastMCP("Sub") @@ -1361,7 +1388,7 @@ def original_tool() -> str: main = FastMCP("Main") main.mount( sub, - prefix="prefix", + namespace="prefix", tool_names={"original_tool": "custom_name"}, ) @@ -1382,7 +1409,7 @@ def original_tool() -> str: main = FastMCP("Main") main.mount( sub, - prefix="prefix", + namespace="prefix", tool_names={"original_tool": "renamed"}, ) @@ -1390,6 +1417,17 @@ def original_tool() -> str: result = await client.call_tool("renamed", {}) assert result.data == "success" + def test_duplicate_tool_rename_targets_raises_error(self): + """Test that duplicate target names in tool_renames raises ValueError.""" + sub = FastMCP("Sub") + main = FastMCP("Main") + + with pytest.raises(ValueError, match="duplicate target name"): + main.mount( + sub, + tool_names={"tool_a": "same_name", "tool_b": "same_name"}, + ) + class TestMountedServerDocketBehavior: """Regression tests for mounted server lifecycle behavior. diff --git a/tests/server/test_server.py b/tests/server/test_server.py index 69b608fa24..1c6f9b52f8 100644 --- a/tests/server/test_server.py +++ b/tests/server/test_server.py @@ -11,11 +11,6 @@ from fastmcp.exceptions import NotFoundError from fastmcp.prompts.prompt import FunctionPrompt, Prompt from fastmcp.resources import Resource, ResourceContent, ResourceTemplate -from fastmcp.server.providers.mounted import ( - add_resource_prefix, - has_resource_prefix, - remove_resource_prefix, -) from fastmcp.tools import FunctionTool from fastmcp.tools.tool import Tool @@ -1104,151 +1099,6 @@ def test_prompt(message: str) -> str: assert prompt.meta == meta_data -class TestResourcePrefixHelpers: - @pytest.mark.parametrize( - "uri,prefix,expected", - [ - # Normal paths - ( - "resource://path/to/resource", - "prefix", - "resource://prefix/path/to/resource", - ), - # Absolute paths (with triple slash) - ("resource:///absolute/path", "prefix", "resource://prefix//absolute/path"), - # Empty prefix should return the original URI - ("resource://path/to/resource", "", "resource://path/to/resource"), - # Different protocols - ("file://path/to/file", "prefix", "file://prefix/path/to/file"), - ("http://example.com/path", "prefix", "http://prefix/example.com/path"), - # Prefixes with special characters - ( - "resource://path/to/resource", - "pre.fix", - "resource://pre.fix/path/to/resource", - ), - ( - "resource://path/to/resource", - "pre/fix", - "resource://pre/fix/path/to/resource", - ), - # Empty paths - ("resource://", "prefix", "resource://prefix/"), - ], - ) - def test_add_resource_prefix(self, uri, prefix, expected): - """Test that add_resource_prefix correctly adds prefixes to URIs.""" - result = add_resource_prefix(uri, prefix) - assert result == expected - - @pytest.mark.parametrize( - "invalid_uri", - [ - "not-a-uri", - "resource:no-slashes", - "missing-protocol", - "http:/missing-slash", - ], - ) - def test_add_resource_prefix_invalid_uri(self, invalid_uri): - """Test that add_resource_prefix raises ValueError for invalid URIs.""" - with pytest.raises(ValueError, match="Invalid URI format"): - add_resource_prefix(invalid_uri, "prefix") - - @pytest.mark.parametrize( - "uri,prefix,expected", - [ - # Normal paths - ( - "resource://prefix/path/to/resource", - "prefix", - "resource://path/to/resource", - ), - # Absolute paths (with triple slash) - ("resource://prefix//absolute/path", "prefix", "resource:///absolute/path"), - # URI without the expected prefix should return the original URI - ( - "resource://other/path/to/resource", - "prefix", - "resource://other/path/to/resource", - ), - # Empty prefix should return the original URI - ("resource://path/to/resource", "", "resource://path/to/resource"), - # Different protocols - ("file://prefix/path/to/file", "prefix", "file://path/to/file"), - # Prefixes with special characters (that need escaping in regex) - ( - "resource://pre.fix/path/to/resource", - "pre.fix", - "resource://path/to/resource", - ), - ( - "resource://pre/fix/path/to/resource", - "pre/fix", - "resource://path/to/resource", - ), - # Empty paths - ("resource://prefix/", "prefix", "resource://"), - ], - ) - def test_remove_resource_prefix(self, uri, prefix, expected): - """Test that remove_resource_prefix correctly removes prefixes from URIs.""" - result = remove_resource_prefix(uri, prefix) - assert result == expected - - @pytest.mark.parametrize( - "invalid_uri", - [ - "not-a-uri", - "resource:no-slashes", - "missing-protocol", - "http:/missing-slash", - ], - ) - def test_remove_resource_prefix_invalid_uri(self, invalid_uri): - """Test that remove_resource_prefix raises ValueError for invalid URIs.""" - with pytest.raises(ValueError, match="Invalid URI format"): - remove_resource_prefix(invalid_uri, "prefix") - - @pytest.mark.parametrize( - "uri,prefix,expected", - [ - # URI with prefix - ("resource://prefix/path/to/resource", "prefix", True), - # URI with another prefix - ("resource://other/path/to/resource", "prefix", False), - # URI with prefix as a substring but not at path start - ("resource://path/prefix/resource", "prefix", False), - # Empty prefix - ("resource://path/to/resource", "", False), - # Different protocols - ("file://prefix/path/to/file", "prefix", True), - # Prefix with special characters - ("resource://pre.fix/path/to/resource", "pre.fix", True), - # Empty paths - ("resource://prefix/", "prefix", True), - ], - ) - def test_has_resource_prefix(self, uri, prefix, expected): - """Test that has_resource_prefix correctly identifies prefixes in URIs.""" - result = has_resource_prefix(uri, prefix) - assert result == expected - - @pytest.mark.parametrize( - "invalid_uri", - [ - "not-a-uri", - "resource:no-slashes", - "missing-protocol", - "http:/missing-slash", - ], - ) - def test_has_resource_prefix_invalid_uri(self, invalid_uri): - """Test that has_resource_prefix raises ValueError for invalid URIs.""" - with pytest.raises(ValueError, match="Invalid URI format"): - has_resource_prefix(invalid_uri, "prefix") - - class TestResourcePrefixMounting: """Test resource prefixing in mounted servers.""" @@ -1297,44 +1147,6 @@ def get_template_resource(param: str): ) assert result[0].text == "Template resource with param-value" # type: ignore[attr-defined] - @pytest.mark.parametrize( - "uri,prefix,expected_match,expected_strip", - [ - # Regular resource - ( - "resource://prefix/path/to/resource", - "prefix", - True, - "resource://path/to/resource", - ), - # Absolute path - ( - "resource://prefix//absolute/path", - "prefix", - True, - "resource:///absolute/path", - ), - # Non-matching prefix - ( - "resource://other/path/to/resource", - "prefix", - False, - "resource://other/path/to/resource", - ), - # Different protocol - ("http://prefix/example.com", "prefix", True, "http://example.com"), - ], - ) - async def test_mounted_server_matching_and_stripping( - self, uri, prefix, expected_match, expected_strip - ): - """Test that resource prefix utility functions correctly match and strip resource prefixes.""" - # Test matching - assert has_resource_prefix(uri, prefix) == expected_match - - # Test stripping - assert remove_resource_prefix(uri, prefix) == expected_strip - class TestShouldIncludeComponent: def test_no_filters_returns_true(self):