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/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(
+                    args={"old_arg": ArgTransform(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."""

src/fastmcp/server/providers/proxy.py

@@ -43,17 +43,14 @@
 from fastmcp.server.server import FastMCP
 from fastmcp.server.tasks.config import TaskConfig
 from fastmcp.tools.tool import Tool, ToolResult
-from fastmcp.tools.tool_transform import (
-    ToolTransformConfig,
-    apply_transformations_to_tools,
-)
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.logging import get_logger
 
 if TYPE_CHECKING:
     from pathlib import Path
 
     from fastmcp.client.transports import ClientTransport
+    from fastmcp.tools.tool_transform import ToolTransformConfig
 
 logger = get_logger(__name__)
 
@@ -455,20 +452,16 @@ class ProxyProvider(Provider):
     def __init__(
         self,
         client_factory: ClientFactoryT,
-        *,
-        tool_transformations: dict[str, ToolTransformConfig] | None = None,
     ):
         """Initialize a ProxyProvider.
 
         Args:
             client_factory: A callable that returns a Client instance when called.
                            This gives you full control over session creation and reuse.
                            Can be either a synchronous or asynchronous function.
-            tool_transformations: Optional tool transformations to apply to proxy tools.
         """
         super().__init__()
         self.client_factory = client_factory
-        self.tool_transformations = tool_transformations or {}
 
     async def _get_client(self) -> Client:
         """Gets a client instance by calling the sync or async factory."""
@@ -487,16 +480,9 @@ async def list_tools(self) -> Sequence[Tool]:
             client = await self._get_client()
             async with client:
                 mcp_tools = await client.list_tools()
-                tools = {
-                    t.name: ProxyTool.from_mcp_tool(self.client_factory, t)
-                    for t in mcp_tools
-                }
-                # Apply tool transformations if configured
-                if self.tool_transformations:
-                    tools = apply_transformations_to_tools(
-                        tools, self.tool_transformations
-                    )
-                return list(tools.values())
+                return [
+                    ProxyTool.from_mcp_tool(self.client_factory, t) for t in mcp_tools
+                ]
         except McpError as e:
             if e.error.code == METHOD_NOT_FOUND:
                 return []
@@ -657,6 +643,7 @@ def __init__(
         self,
         *,
         client_factory: ClientFactoryT,
+        tool_transforms: dict[str, ToolTransformConfig] | None = None,
         **kwargs,
     ):
         """Initialize the proxy server.
@@ -668,15 +655,16 @@ def __init__(
             client_factory: A callable that returns a Client instance when called.
                            This gives you full control over session creation and reuse.
                            Can be either a synchronous or asynchronous function.
+            tool_transforms: Optional dict of tool_name to ToolTransformConfig for
+                schema modifications (arg renames, hidden args, etc.)
             **kwargs: Additional settings for the FastMCP server.
         """
-        # Extract tool_transformations before passing to parent
-        tool_transformations = kwargs.pop("tool_transformations", None)
         super().__init__(**kwargs)
         self.client_factory = client_factory
-        self.add_provider(
-            ProxyProvider(client_factory, tool_transformations=tool_transformations)
-        )
+        provider: Provider = ProxyProvider(client_factory)
+        if tool_transforms:
+            provider = provider.with_transforms(tool_transforms=tool_transforms)
+        self.add_provider(provider)
 
 
 # -----------------------------------------------------------------------------

src/fastmcp/server/providers/transforming.py

@@ -16,6 +16,7 @@
 from fastmcp.resources.template import ResourceTemplate
 from fastmcp.server.providers.base import Provider
 from fastmcp.tools.tool import Tool
+from fastmcp.tools.tool_transform import ToolTransformConfig
 from fastmcp.utilities.components import FastMCPComponent
 
 if TYPE_CHECKING:
@@ -37,13 +38,17 @@ class TransformingProvider(Provider):
         - 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"
+        - tool_transforms apply schema modifications (arg renames, hidden args, etc.)
 
     Example:
         ```python
         # Via with_transforms() method (preferred)
         provider = SomeProvider().with_transforms(
             namespace="api",
-            tool_renames={"verbose_tool_name": "short"}
+            tool_renames={"verbose_tool_name": "short"},
+            tool_transforms={"short": ToolTransformConfig(
+                args={"old_arg": ArgTransform(name="new_arg")}
+            )}
         )
 
         # Stacking composes transformations:
@@ -62,6 +67,7 @@ def __init__(
         *,
         namespace: str | None = None,
         tool_renames: dict[str, str] | None = None,
+        tool_transforms: dict[str, ToolTransformConfig] | None = None,
     ):
         """Initialize a TransformingProvider.
 
@@ -70,11 +76,15 @@ def __init__(
             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.
+            tool_transforms: Map of tool_name → ToolTransformConfig for schema
+                modifications (arg renames, hidden args, custom functions, etc.).
+                Applied after namespace/rename transformations.
         """
         super().__init__()
         self._wrapped: Provider = provider
         self.namespace = namespace
         self.tool_renames = tool_renames or {}
+        self.tool_transforms = tool_transforms or {}
 
         # Validate that renames are reversible (no duplicate target names)
         if len(self.tool_renames) != len(set(self.tool_renames.values())):
@@ -89,6 +99,13 @@ def __init__(
 
         self._tool_renames_reverse = {v: k for k, v in self.tool_renames.items()}
 
+        # Build reverse mapping for tool_transforms that rename tools
+        # Maps: final_name -> pre_transform_name (the key in tool_transforms)
+        self._tool_transforms_name_reverse: dict[str, str] = {}
+        for pre_transform_name, config in self.tool_transforms.items():
+            if config.name is not None:
+                self._tool_transforms_name_reverse[config.name] = pre_transform_name
+
     # -------------------------------------------------------------------------
     # Tool name transformation
     # -------------------------------------------------------------------------
@@ -166,22 +183,44 @@ def _reverse_resource_uri(self, uri: str) -> str | None:
     # Tool methods
     # -------------------------------------------------------------------------
 
+    def _apply_tool_transform(self, tool: Tool, name: str) -> Tool:
+        """Apply namespace/rename and schema transforms to a tool."""
+        # First apply namespace/rename
+        t = tool.model_copy(update={"name": name})
+
+        # Then apply schema transform if present
+        if name in self.tool_transforms:
+            t = self.tool_transforms[name].apply(t)
+
+        return t
+
     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)})
+            self._apply_tool_transform(t, 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)
+        # First check if name was transformed by tool_transforms (e.g., renamed)
+        # If so, get the pre-transform name (the key in tool_transforms)
+        pre_transform_name = self._tool_transforms_name_reverse.get(name, name)
+
+        # Now reverse the namespace/rename transformation
+        original = self._reverse_tool_name(pre_transform_name)
         if original is None:
             return None
+
         tool = await self._wrapped.get_tool(original)
         if tool:
-            return tool.model_copy(update={"name": name})
+            # Apply transforms using the pre_transform_name (the key in tool_transforms)
+            transformed = self._apply_tool_transform(tool, pre_transform_name)
+            # Only return if requested name matches the final transformed name
+            # This prevents accessing tools by their pre-transform name
+            if transformed.name == name:
+                return transformed
         return None
 
     # -------------------------------------------------------------------------
@@ -266,11 +305,8 @@ async def get_tasks(self) -> Sequence[FastMCPComponent]:
 
         for component in await self._wrapped.get_tasks():
             if isinstance(component, Tool):
-                transformed.append(
-                    component.model_copy(
-                        update={"name": self._transform_tool_name(component.name)}
-                    )
-                )
+                name = self._transform_tool_name(component.name)
+                transformed.append(self._apply_tool_transform(component, name))
             elif isinstance(component, ResourceTemplate):
                 transformed.append(
                     component.model_copy(