Add mirrored component support for proxy servers

Prevents enable/disable operations on mirrored components (tools, resources, prompts retrieved from proxy servers). Users must create local copies with .copy() and add them to their server to modify them. Local components take precedence over mirrored ones.

Closes #1102

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

docs/docs.json

@@ -81,13 +81,13 @@
                     "icon": "stars",
                     "pages": [
                       "servers/context",
+                      "servers/proxy",
+                      "servers/composition",
                       "servers/elicitation",
                       "servers/logging",
                       "servers/progress",
                       "servers/sampling",
-                      "servers/middleware",
-                      "servers/composition",
-                      "servers/proxy"
+                      "servers/middleware"
                     ]
                   },
                   {

docs/servers/composition.mdx

@@ -1,6 +1,6 @@
 ---
 title: Server Composition
-sidebarTitle: Composition
+sidebarTitle: Server Composition
 description: Combine multiple FastMCP servers into a single, larger application using mounting and importing.
 icon: puzzle-piece
 ---

docs/servers/proxy.mdx

@@ -65,6 +65,10 @@ This single setup gives you:
 - Session isolation to prevent context mixing
 - Full compatibility with all MCP clients
 
+You can also pass a FastMCP [client transport](/clients/transports) (or parameter that can be inferred to a transport) to `as_proxy()`. This will automatically create a `ProxyClient` instance for you.
+
+Finally, you can pass a regular FastMCP `Client` instance to `as_proxy()`. This will work for many use cases, but may break if advanced MCP features like sampling or elicitation are invoked by the server. 
+
 ## Session Isolation & Concurrency
 
 <VersionBadge version="2.10.3" />
@@ -240,23 +244,35 @@ composite_proxy = FastMCP.as_proxy(config, name="Composite Proxy")
 # - weather://weather/icons/sunny, calendar://calendar/events/today
 ```
 
-## Alternative Approaches
+## Mirrored Components
 
-The examples above show the recommended approach using `ProxyClient` or transport strings. For advanced use cases, you can also work directly with the underlying classes.
+When you access tools, resources, or prompts from a proxy server, they are "mirrored" from the remote server. Mirrored components cannot be modified directly directly since they reflect the state of the remote server. For example, you can not simply "disable" a mirrored component.
 
-### Using Regular Client
+However, you can create a copy of a mirrored component and store it as a new locally-defined component. Local components always take precedence over mirored ones because the proxy server will check its own registry before it attempts to engage the remote server.
 
-You can pass a regular `Client` instance to `as_proxy()`. The proxy will automatically create an appropriate session strategy:
+Therefore, to enable or disable a proxy tool, resource, or prompt, you should first create a local copy and add it to your own server. Here's an example of how to do that for a tool:
 
 ```python
-from fastmcp import FastMCP, Client
+# Create your own server
+my_server = FastMCP("MyServer")
 
-# Using regular Client (session strategy auto-detected)
-client = Client("backend_server.py")
-proxy = FastMCP.as_proxy(client)
+# Get a proxy server
+proxy = FastMCP.as_proxy("backend_server.py")
+
+# Add mirrored components to your server
+async with proxy:
+    mirrored_tool = await proxy.get_tool("useful_tool")
+    
+    # Create a local copy that you can modify
+    local_tool = mirrored_tool.copy()
+    
+    # Add the local copy to your server
+    my_server.add_tool(local_tool)
+    
+    # Now you can disable YOUR copy
+    local_tool.disable()
 ```
 
-This approach provides session isolation but doesn't include advanced MCP feature forwarding (sampling, elicitation, etc.) unless you configure handlers manually.
 
 ## `FastMCPProxy` Class
 
@@ -308,4 +324,5 @@ def custom_client_factory():
     return client
 
 proxy = FastMCPProxy(client_factory=custom_client_factory)
-``` 
+```
+

src/fastmcp/cli/install/claude_code.py

@@ -1,5 +1,6 @@
 """Claude Code integration for FastMCP install using Cyclopts."""
 
+import shutil
 import subprocess
 import sys
 from pathlib import Path
@@ -16,22 +17,50 @@ logger = get_logger(__name__)
 
 
 def find_claude_command() -> str | None:
-    """Find the Claude Code CLI command."""
-    # Check the default installation location
-    default_path = Path.home() / ".claude" / "local" / "claude"
-    if default_path.exists():
+    """Find the Claude Code CLI command.
+
+    Checks common installation locations since 'claude' is often a shell alias
+    that doesn't work with subprocess calls.
+    """
+    # First try shutil.which() in case it's a real executable in PATH
+    claude_in_path = shutil.which("claude")
+    if claude_in_path:
         try:
             result = subprocess.run(
-                [str(default_path), "--version"],
+                [claude_in_path, "--version"],
                 check=True,
                 capture_output=True,
                 text=True,
             )
             if "Claude Code" in result.stdout:
-                return str(default_path)
+                return claude_in_path
         except (subprocess.CalledProcessError, FileNotFoundError):
             pass
 
+    # Check common installation locations (aliases don't work with subprocess)
+    potential_paths = [
+        # Default Claude Code installation location (after migration)
+        Path.home() / ".claude" / "local" / "claude",
+        # npm global installation on macOS/Linux (default)
+        Path("/usr/local/bin/claude"),
+        # npm global installation with custom prefix
+        Path.home() / ".npm-global" / "bin" / "claude",
+    ]
+
+    for path in potential_paths:
+        if path.exists():
+            try:
+                result = subprocess.run(
+                    [str(path), "--version"],
+                    check=True,
+                    capture_output=True,
+                    text=True,
+                )
+                if "Claude Code" in result.stdout:
+                    return str(path)
+            except (subprocess.CalledProcessError, FileNotFoundError):
+                continue
+
     return None
 
 

src/fastmcp/server/proxy.py

@@ -36,6 +36,7 @@ from fastmcp.server.dependencies import get_context
 from fastmcp.server.server import FastMCP
 from fastmcp.tools.tool import Tool, ToolResult
 from fastmcp.tools.tool_manager import ToolManager
+from fastmcp.utilities.components import MirroredComponent
 from fastmcp.utilities.logging import get_logger
 
 if TYPE_CHECKING:
@@ -226,7 +227,7 @@ class ProxyPromptManager(PromptManager):
                 return result
 
 
-class ProxyTool(Tool):
+class ProxyTool(Tool, MirroredComponent):
     """
     A Tool that represents and executes a tool on a remote server.
     """
@@ -245,6 +246,7 @@ class ProxyTool(Tool):
             parameters=mcp_tool.inputSchema,
             annotations=mcp_tool.annotations,
             output_schema=mcp_tool.outputSchema,
+            _mirrored=True,
         )
 
     async def run(
@@ -266,7 +268,7 @@ class ProxyTool(Tool):
         )
 
 
-class ProxyResource(Resource):
+class ProxyResource(Resource, MirroredComponent):
     """
     A Resource that represents and reads a resource from a remote server.
     """
@@ -298,6 +300,7 @@ class ProxyResource(Resource):
             name=mcp_resource.name,
             description=mcp_resource.description,
             mime_type=mcp_resource.mimeType or "text/plain",
+            _mirrored=True,
         )
 
     async def read(self) -> str | bytes:
@@ -315,7 +318,7 @@ class ProxyResource(Resource):
             raise ResourceError(f"Unsupported content type: {type(result[0])}")
 
 
-class ProxyTemplate(ResourceTemplate):
+class ProxyTemplate(ResourceTemplate, MirroredComponent):
     """
     A ResourceTemplate that represents and creates resources from a remote server template.
     """
@@ -336,6 +339,7 @@ class ProxyTemplate(ResourceTemplate):
             description=mcp_template.description,
             mime_type=mcp_template.mimeType or "text/plain",
             parameters={},  # Remote templates don't have local parameters
+            _mirrored=True,
         )
 
     async def create_resource(
@@ -371,7 +375,7 @@ class ProxyTemplate(ResourceTemplate):
         )
 
 
-class ProxyPrompt(Prompt):
+class ProxyPrompt(Prompt, MirroredComponent):
     """
     A Prompt that represents and renders a prompt from a remote server.
     """
@@ -400,6 +404,7 @@ class ProxyPrompt(Prompt):
             name=mcp_prompt.name,
             description=mcp_prompt.description,
             arguments=arguments,
+            _mirrored=True,
         )
 
     async def render(self, arguments: dict[str, Any]) -> list[PromptMessage]:

src/fastmcp/server/server.py

@@ -759,7 +759,7 @@ class FastMCP(Generic[LifespanResultT]):
         )
         return await self._apply_middleware(mw_context, _handler)
 
-    def add_tool(self, tool: Tool) -> None:
+    def add_tool(self, tool: Tool) -> Tool:
         """Add a tool to the server.
 
         The tool function can optionally request a Context object by adding a parameter
@@ -767,6 +767,9 @@ class FastMCP(Generic[LifespanResultT]):
 
         Args:
             tool: The Tool instance to register
+
+        Returns:
+            The tool instance that was added to the server.
         """
         self._tool_manager.add_tool(tool)
         self._cache.clear()
@@ -780,6 +783,8 @@ class FastMCP(Generic[LifespanResultT]):
         except RuntimeError:
             pass  # No context available
 
+        return tool
+
     def remove_tool(self, name: str) -> None:
         """Remove a tool from the server.
 
@@ -958,13 +963,15 @@ class FastMCP(Generic[LifespanResultT]):
             enabled=enabled,
         )
 
-    def add_resource(self, resource: Resource) -> None:
+    def add_resource(self, resource: Resource) -> Resource:
         """Add a resource to the server.
 
         Args:
             resource: A Resource instance to add
-        """
 
+        Returns:
+            The resource instance that was added to the server.
+        """
         self._resource_manager.add_resource(resource)
         self._cache.clear()
 
@@ -977,11 +984,16 @@ class FastMCP(Generic[LifespanResultT]):
         except RuntimeError:
             pass  # No context available
 
-    def add_template(self, template: ResourceTemplate) -> None:
+        return resource
+
+    def add_template(self, template: ResourceTemplate) -> ResourceTemplate:
         """Add a resource template to the server.
 
         Args:
             template: A ResourceTemplate instance to add
+
+        Returns:
+            The template instance that was added to the server.
         """
         self._resource_manager.add_template(template)
 
@@ -994,6 +1006,8 @@ class FastMCP(Generic[LifespanResultT]):
         except RuntimeError:
             pass  # No context available
 
+        return template
+
     def add_resource_fn(
         self,
         fn: AnyFunction,
@@ -1159,11 +1173,14 @@ class FastMCP(Generic[LifespanResultT]):
 
         return decorator
 
-    def add_prompt(self, prompt: Prompt) -> None:
+    def add_prompt(self, prompt: Prompt) -> Prompt:
         """Add a prompt to the server.
 
         Args:
             prompt: A Prompt instance to add
+
+        Returns:
+            The prompt instance that was added to the server.
         """
         self._prompt_manager.add_prompt(prompt)
         self._cache.clear()
@@ -1177,6 +1194,8 @@ class FastMCP(Generic[LifespanResultT]):
         except RuntimeError:
             pass  # No context available
 
+        return prompt
+
     @overload
     def prompt(
         self,

src/fastmcp/utilities/components.py

@@ -77,3 +77,46 @@ class FastMCPComponent(FastMCPBaseModel):
     def disable(self) -> None:
         """Disable the component."""
         self.enabled = False
+
+    def copy(self) -> Self:
+        """Create a copy of the component."""
+        return self.model_copy()
+
+
+class MirroredComponent(FastMCPComponent):
+    """Base class for components that are mirrored from a remote server.
+
+    Mirrored components cannot be enabled or disabled directly. Call copy() first
+    to create a local version you can modify.
+    """
+
+    _mirrored: bool = PrivateAttr(default=False)
+
+    def __init__(self, *, _mirrored: bool = False, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self._mirrored = _mirrored
+
+    def enable(self) -> None:
+        """Enable the component."""
+        if self._mirrored:
+            raise RuntimeError(
+                f"Cannot enable mirrored component '{self.name}'. "
+                f"Create a local copy first with {self.name}.copy() and add it to your server."
+            )
+        super().enable()
+
+    def disable(self) -> None:
+        """Disable the component."""
+        if self._mirrored:
+            raise RuntimeError(
+                f"Cannot disable mirrored component '{self.name}'. "
+                f"Create a local copy first with {self.name}.copy() and add it to your server."
+            )
+        super().disable()
+
+    def copy(self) -> Self:
+        """Create a copy of the component that can be modified."""
+        # Create a copy and mark it as not mirrored
+        copied = self.model_copy()
+        copied._mirrored = False
+        return copied

tests/server/proxy/test_proxy_server.py

@@ -516,3 +516,161 @@ async def test_proxy_handles_multiple_concurrent_tasks_correctly(
     assert list(results["tools"]) == Contains(
         "greet", "add", "error_tool", "tool_without_description"
     )
+
+
+class TestMirroredComponents:
+    """Test mirrored component functionality - components retrieved from proxy servers."""
+
+    async def test_mirrored_tool_cannot_be_enabled(self, proxy_server):
+        """Test that mirrored tools cannot be enabled directly."""
+        tools = await proxy_server.get_tools()
+        mirrored_tool = tools["greet"]
+
+        # Verify it's mirrored
+        assert mirrored_tool._mirrored is True
+
+        # Should raise error when trying to enable
+        with pytest.raises(RuntimeError, match="Cannot enable mirrored component"):
+            mirrored_tool.enable()
+
+    async def test_mirrored_tool_cannot_be_disabled(self, proxy_server):
+        """Test that mirrored tools cannot be disabled directly."""
+        tools = await proxy_server.get_tools()
+        mirrored_tool = tools["greet"]
+
+        # Verify it's mirrored
+        assert mirrored_tool._mirrored is True
+
+        # Should raise error when trying to disable
+        with pytest.raises(RuntimeError, match="Cannot disable mirrored component"):
+            mirrored_tool.disable()
+
+    async def test_mirrored_resource_cannot_be_enabled(self, proxy_server):
+        """Test that mirrored resources cannot be enabled directly."""
+        resources = await proxy_server.get_resources()
+        mirrored_resource = resources["resource://wave"]
+
+        # Verify it's mirrored
+        assert mirrored_resource._mirrored is True
+
+        # Should raise error when trying to enable
+        with pytest.raises(RuntimeError, match="Cannot enable mirrored component"):
+            mirrored_resource.enable()
+
+    async def test_mirrored_resource_cannot_be_disabled(self, proxy_server):
+        """Test that mirrored resources cannot be disabled directly."""
+        resources = await proxy_server.get_resources()
+        mirrored_resource = resources["resource://wave"]
+
+        # Verify it's mirrored
+        assert mirrored_resource._mirrored is True
+
+        # Should raise error when trying to disable
+        with pytest.raises(RuntimeError, match="Cannot disable mirrored component"):
+            mirrored_resource.disable()
+
+    async def test_mirrored_prompt_cannot_be_enabled(self, proxy_server):
+        """Test that mirrored prompts cannot be enabled directly."""
+        prompts = await proxy_server.get_prompts()
+        mirrored_prompt = prompts["welcome"]
+
+        # Verify it's mirrored
+        assert mirrored_prompt._mirrored is True
+
+        # Should raise error when trying to enable
+        with pytest.raises(RuntimeError, match="Cannot enable mirrored component"):
+            mirrored_prompt.enable()
+
+    async def test_mirrored_prompt_cannot_be_disabled(self, proxy_server):
+        """Test that mirrored prompts cannot be disabled directly."""
+        prompts = await proxy_server.get_prompts()
+        mirrored_prompt = prompts["welcome"]
+
+        # Verify it's mirrored
+        assert mirrored_prompt._mirrored is True
+
+        # Should raise error when trying to disable
+        with pytest.raises(RuntimeError, match="Cannot disable mirrored component"):
+            mirrored_prompt.disable()
+
+    async def test_copy_creates_non_mirrored_component(self, proxy_server):
+        """Test that copy() creates a non-mirrored component that can be modified."""
+        tools = await proxy_server.get_tools()
+        mirrored_tool = tools["greet"]
+
+        # Create a copy
+        local_tool = mirrored_tool.copy()
+
+        # Copy should not be mirrored
+        assert local_tool._mirrored is False
+
+        # Should be able to enable/disable the copy
+        local_tool.enable()
+        assert local_tool.enabled is True
+
+        local_tool.disable()
+        assert local_tool.enabled is False
+
+    async def test_local_component_takes_precedence_over_mirrored(self, proxy_server):
+        """Test that local components take precedence over mirrored ones."""
+        # Get the mirrored tool
+        tools = await proxy_server.get_tools()
+        mirrored_tool = tools["greet"]
+
+        # Create a local copy and add it
+        local_tool = mirrored_tool.copy()
+        proxy_server.add_tool(local_tool)
+
+        # Disable the local copy
+        local_tool.disable()
+
+        # The local disabled tool should take precedence
+        updated_tools = await proxy_server.get_tools()
+        final_tool = updated_tools["greet"]
+
+        # Should be the local tool (not mirrored) and disabled
+        assert final_tool is local_tool
+        assert final_tool._mirrored is False
+        assert final_tool.enabled is False
+
+    async def test_error_messages_mention_copy_method(self, proxy_server):
+        """Test that error messages guide users to use copy() method."""
+        tools = await proxy_server.get_tools()
+        mirrored_tool = tools["greet"]
+
+        # Check enable error message
+        with pytest.raises(RuntimeError) as exc_info:
+            mirrored_tool.enable()
+        assert "copy()" in str(exc_info.value)
+
+        # Check disable error message
+        with pytest.raises(RuntimeError) as exc_info:
+            mirrored_tool.disable()
+        assert "copy()" in str(exc_info.value)
+
+    async def test_client_cannot_call_disabled_proxy_tool(self, proxy_server):
+        """Test that clients cannot call a tool when local copy is disabled."""
+        # Get the mirrored tool
+        tools = await proxy_server.get_tools()
+        mirrored_tool = tools["greet"]
+
+        # Verify the tool works initially
+        async with Client(proxy_server) as client:
+            result = await client.call_tool("greet", {"name": "Alice"})
+            assert result.data == "Hello, Alice!"
+
+        # Create a local copy and disable it
+        local_tool = mirrored_tool.copy()
+        proxy_server.add_tool(local_tool)
+        local_tool.disable()
+
+        # Client should now get "Unknown tool" error
+        async with Client(proxy_server) as client:
+            with pytest.raises(ToolError, match="Unknown tool"):
+                await client.call_tool("greet", {"name": "Alice"})
+
+        # Tool should not appear in tool list either
+        async with Client(proxy_server) as client:
+            tools_list = await client.list_tools()
+            tool_names = [tool.name for tool in tools_list]
+            assert "greet" not in tool_names