Add transform system for modifying components in provider chains

src/fastmcp/client/transports.py

@@ -1037,8 +1037,8 @@ def _create_proxy(
         client = ProxyClient(transport=transport, timeout=timeout)
         proxy = create_proxy(
             client,
+            tool_transforms=tool_transforms,
             name=f"Proxy-{name}",
-            tool_transformations=tool_transforms,
             include_tags=include_tags,
             exclude_tags=exclude_tags,
         )

src/fastmcp/mcp_config.py

@@ -108,8 +108,8 @@ def _to_server_and_underlying_transport(
 
         wrapped_mcp_server = create_proxy(
             client,
+            tool_transforms=self.tools,
             name=server_name,
-            tool_transformations=self.tools,
             include_tags=self.include_tags,
             exclude_tags=self.exclude_tags,
         )

src/fastmcp/server/providers/aggregate.py

@@ -0,0 +1,191 @@
+"""AggregateProvider for combining multiple providers into one.
+
+This module provides `AggregateProvider` which presents multiple providers
+as a single unified provider. Used internally by FastMCP for applying
+server-level transforms across all providers.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import AsyncIterator, Sequence
+from contextlib import AsyncExitStack, asynccontextmanager
+from typing import TypeVar
+
+from fastmcp.exceptions import NotFoundError
+from fastmcp.prompts.prompt import Prompt
+from fastmcp.resources.resource import Resource
+from fastmcp.resources.template import ResourceTemplate
+from fastmcp.server.providers.base import Provider
+from fastmcp.tools.tool import Tool
+from fastmcp.utilities.components import FastMCPComponent
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class AggregateProvider(Provider):
+    """Presents multiple providers as a single provider.
+
+    Components are aggregated from all providers. For get_* operations,
+    providers are queried in parallel and the first non-None result is returned.
+
+    Errors from individual providers are logged and skipped (graceful degradation).
+    This matches the behavior of FastMCP's original provider iteration.
+    """
+
+    def __init__(self, providers: Sequence[Provider]) -> None:
+        """Initialize with a sequence of providers.
+
+        Args:
+            providers: The providers to aggregate. Queried in order for lookups.
+        """
+        super().__init__()
+        self._providers = list(providers)
+
+    def _collect_list_results(
+        self, results: list[Sequence[T] | BaseException], operation: str
+    ) -> list[T]:
+        """Collect successful list results, logging any exceptions."""
+        collected: list[T] = []
+        for i, result in enumerate(results):
+            if isinstance(result, BaseException):
+                logger.debug(
+                    f"Error during {operation} from provider "
+                    f"{self._providers[i]}: {result}"
+                )
+                continue
+            collected.extend(result)
+        return collected
+
+    def _get_first_result(
+        self, results: list[T | None | BaseException], operation: str
+    ) -> T | None:
+        """Get first successful non-None result, logging non-NotFoundError exceptions."""
+        for i, result in enumerate(results):
+            if isinstance(result, BaseException):
+                # NotFoundError is expected - don't log it
+                if not isinstance(result, NotFoundError):
+                    logger.debug(
+                        f"Error during {operation} from provider "
+                        f"{self._providers[i]}: {result}"
+                    )
+                continue
+            if result is not None:
+                return result
+        return None
+
+    def __repr__(self) -> str:
+        return f"AggregateProvider(providers={self._providers!r})"
+
+    # -------------------------------------------------------------------------
+    # Tools
+    # -------------------------------------------------------------------------
+
+    async def list_tools(self) -> Sequence[Tool]:
+        """List all tools from all providers."""
+        results = await asyncio.gather(
+            *[p.list_tools() for p in self._providers], return_exceptions=True
+        )
+        return self._collect_list_results(results, "list_tools")
+
+    async def get_tool(self, name: str) -> Tool | None:
+        """Get tool by name from first provider that has it."""
+        results = await asyncio.gather(
+            *[p.get_tool(name) for p in self._providers], return_exceptions=True
+        )
+        return self._get_first_result(results, f"get_tool({name!r})")
+
+    # -------------------------------------------------------------------------
+    # Resources
+    # -------------------------------------------------------------------------
+
+    async def list_resources(self) -> Sequence[Resource]:
+        """List all resources from all providers."""
+        results = await asyncio.gather(
+            *[p.list_resources() for p in self._providers], return_exceptions=True
+        )
+        return self._collect_list_results(results, "list_resources")
+
+    async def get_resource(self, uri: str) -> Resource | None:
+        """Get resource by URI from first provider that has it."""
+        results = await asyncio.gather(
+            *[p.get_resource(uri) for p in self._providers], return_exceptions=True
+        )
+        return self._get_first_result(results, f"get_resource({uri!r})")
+
+    # -------------------------------------------------------------------------
+    # Resource Templates
+    # -------------------------------------------------------------------------
+
+    async def list_resource_templates(self) -> Sequence[ResourceTemplate]:
+        """List all resource templates from all providers."""
+        results = await asyncio.gather(
+            *[p.list_resource_templates() for p in self._providers],
+            return_exceptions=True,
+        )
+        return self._collect_list_results(results, "list_resource_templates")
+
+    async def get_resource_template(self, uri: str) -> ResourceTemplate | None:
+        """Get resource template by URI from first provider that has it."""
+        results = await asyncio.gather(
+            *[p.get_resource_template(uri) for p in self._providers],
+            return_exceptions=True,
+        )
+        return self._get_first_result(results, f"get_resource_template({uri!r})")
+
+    # -------------------------------------------------------------------------
+    # Prompts
+    # -------------------------------------------------------------------------
+
+    async def list_prompts(self) -> Sequence[Prompt]:
+        """List all prompts from all providers."""
+        results = await asyncio.gather(
+            *[p.list_prompts() for p in self._providers], return_exceptions=True
+        )
+        return self._collect_list_results(results, "list_prompts")
+
+    async def get_prompt(self, name: str) -> Prompt | None:
+        """Get prompt by name from first provider that has it."""
+        results = await asyncio.gather(
+            *[p.get_prompt(name) for p in self._providers], return_exceptions=True
+        )
+        return self._get_first_result(results, f"get_prompt({name!r})")
+
+    # -------------------------------------------------------------------------
+    # Components
+    # -------------------------------------------------------------------------
+
+    async def get_component(
+        self, key: str
+    ) -> Tool | Resource | ResourceTemplate | Prompt | None:
+        """Get component by key from first provider that has it."""
+        results = await asyncio.gather(
+            *[p.get_component(key) for p in self._providers], return_exceptions=True
+        )
+        return self._get_first_result(results, f"get_component({key!r})")
+
+    # -------------------------------------------------------------------------
+    # Tasks
+    # -------------------------------------------------------------------------
+
+    async def get_tasks(self) -> Sequence[FastMCPComponent]:
+        """Get all task-eligible components from all providers."""
+        results = await asyncio.gather(
+            *[p.get_tasks() for p in self._providers], return_exceptions=True
+        )
+        return self._collect_list_results(results, "get_tasks")
+
+    # -------------------------------------------------------------------------
+    # Lifecycle
+    # -------------------------------------------------------------------------
+
+    @asynccontextmanager
+    async def lifespan(self) -> AsyncIterator[None]:
+        """Combine lifespans of all providers."""
+        async with AsyncExitStack() as stack:
+            for provider in self._providers:
+                await stack.enter_async_context(provider.lifespan())
+            yield

src/fastmcp/server/providers/base.py

@@ -30,6 +30,7 @@ async def get_tool(self, name: str) -> Tool | None:
 
 from collections.abc import AsyncIterator, Sequence
 from contextlib import asynccontextmanager
+from typing import TYPE_CHECKING
 
 from fastmcp.prompts.prompt import Prompt
 from fastmcp.resources.resource import Resource
@@ -39,6 +40,9 @@ async def get_tool(self, name: str) -> Tool | None:
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.visibility import VisibilityFilter
 
+if TYPE_CHECKING:
+    from fastmcp.tools.tool_transform import ToolTransformConfig
+
 
 class Provider:
     """Base class for dynamic component providers.
@@ -69,6 +73,7 @@ def with_transforms(
         *,
         namespace: str | None = None,
         tool_renames: dict[str, str] | None = None,
+        tool_transforms: dict[str, ToolTransformConfig] | None = None,
     ) -> Provider:
         """Apply transformations to this provider's components.
 
@@ -81,6 +86,9 @@ def with_transforms(
                 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.
+            tool_transforms: Map of tool_name → ToolTransformConfig for schema
+                modifications (arg renames, hidden args, custom functions, etc.).
+                Applied after namespace/rename transformations.
 
         Returns:
             A TransformingProvider wrapping this provider.
@@ -100,6 +108,13 @@ def with_transforms(
             # "verbose_tool_name" → "short" (explicit rename)
             # "other_tool" → "api_other_tool" (namespace applied)
 
+            # Apply schema transforms to modify tool arguments
+            provider = MyProvider().with_transforms(
+                tool_transforms={"my_tool": ToolTransformConfig(
+                    arguments={"old_arg": ArgTransformConfig(name="new_arg")}
+                )}
+            )
+
             # Stacking composes transformations
             provider = (
                 MyProvider()
@@ -112,7 +127,10 @@ def with_transforms(
         from fastmcp.server.providers.transforming import TransformingProvider
 
         return TransformingProvider(
-            self, namespace=namespace, tool_renames=tool_renames
+            self,
+            namespace=namespace,
+            tool_renames=tool_renames,
+            tool_transforms=tool_transforms,
         )
 
     def with_namespace(self, namespace: str) -> Provider:

src/fastmcp/server/providers/filesystem.py

@@ -104,7 +104,6 @@ def _load_components(self) -> None:
         # Clear existing components if reloading
         if self._loaded:
             self._components.clear()
-            self._tool_transformations.clear()
 
         result = discover_and_import(self._root)
 

src/fastmcp/server/providers/local_provider.py

@@ -42,10 +42,6 @@ def greet(name: str) -> str:
 from fastmcp.server.providers.base import Provider
 from fastmcp.server.tasks.config import TaskConfig
 from fastmcp.tools.tool import FunctionTool, Tool
-from fastmcp.tools.tool_transform import (
-    ToolTransformConfig,
-    apply_transformations_to_tools,
-)
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.types import NotSet, NotSetT
@@ -112,7 +108,6 @@ def __init__(
         self._on_duplicate = on_duplicate
         # Unified component storage - keyed by prefixed key (e.g., "tool:name", "resource:uri")
         self._components: dict[str, FastMCPComponent] = {}
-        self._tool_transformations: dict[str, ToolTransformConfig] = {}
 
     def _send_list_changed_notification(self, component: FastMCPComponent) -> None:
         """Send a list changed notification for the component type."""
@@ -215,58 +210,28 @@ def remove_prompt(self, name: str) -> None:
         """Remove a prompt from this provider's storage."""
         self._remove_component(Prompt.make_key(name))
 
-    # =========================================================================
-    # Tool transformation methods
-    # =========================================================================
-
-    def add_tool_transformation(
-        self, tool_name: str, transformation: ToolTransformConfig
-    ) -> None:
-        """Add a tool transformation.
-
-        Args:
-            tool_name: The name of the tool to transform.
-            transformation: The transformation configuration.
-        """
-        self._tool_transformations[tool_name] = transformation
-
-    def get_tool_transformation(self, tool_name: str) -> ToolTransformConfig | None:
-        """Get a tool transformation.
-
-        Args:
-            tool_name: The name of the tool.
-
-        Returns:
-            The transformation config, or None if not found.
-        """
-        return self._tool_transformations.get(tool_name)
-
-    def remove_tool_transformation(self, tool_name: str) -> None:
-        """Remove a tool transformation.
-
-        Args:
-            tool_name: The name of the tool.
-        """
-        if tool_name in self._tool_transformations:
-            del self._tool_transformations[tool_name]
-
     # =========================================================================
     # Provider interface implementation
     # =========================================================================
 
     async def list_tools(self) -> Sequence[Tool]:
-        """Return all visible tools with transformations applied."""
-        tools = {k: v for k, v in self._components.items() if isinstance(v, Tool)}
-        transformed = apply_transformations_to_tools(
-            tools=tools,
-            transformations=self._tool_transformations,
-        )
-        return [t for t in transformed.values() if self._is_component_enabled(t)]
+        """Return all visible tools."""
+        return [
+            v
+            for v in self._components.values()
+            if isinstance(v, Tool) and self._is_component_enabled(v)
+        ]
 
     async def get_tool(self, name: str) -> Tool | None:
-        """Get a tool by name, with transformations applied."""
-        tools = await self.list_tools()
-        return next((t for t in tools if t.name == name), None)
+        """Get a tool by name."""
+        tool = self._get_component(Tool.make_key(name))
+        if (
+            tool is not None
+            and isinstance(tool, Tool)
+            and self._is_component_enabled(tool)
+        ):
+            return tool
+        return None
 
     async def list_resources(self) -> Sequence[Resource]:
         """Return all visible resources."""