Refactor server to wrap mcp calls

src/fastmcp/exceptions.py

@@ -33,3 +33,7 @@ class ClientError(Exception):
 
 class NotFoundError(Exception):
     """Object not found."""
+
+
+class DisabledError(Exception):
+    """Object is disabled."""

src/fastmcp/server/proxy.py

@@ -241,20 +241,20 @@ class FastMCPProxy(FastMCP):
                 prompts[prompt_proxy.name] = prompt_proxy
         return prompts
 
-    async def _mcp_call_tool(
+    async def _call_tool(
         self, key: str, arguments: dict[str, Any]
     ) -> list[TextContent | ImageContent | EmbeddedResource]:
         try:
-            result = await super()._mcp_call_tool(key, arguments)
+            result = await super()._call_tool(key, arguments)
             return result
         except NotFoundError:
             async with self.client:
                 result = await self.client.call_tool(key, arguments)
             return result
 
-    async def _mcp_read_resource(self, uri: AnyUrl | str) -> list[ReadResourceContents]:
+    async def _read_resource(self, uri: AnyUrl | str) -> list[ReadResourceContents]:
         try:
-            result = await super()._mcp_read_resource(uri)
+            result = await super()._read_resource(uri)
             return result
         except NotFoundError:
             async with self.client:
@@ -270,11 +270,11 @@ class FastMCPProxy(FastMCP):
                 ReadResourceContents(content=content, mime_type=resource[0].mimeType)
             ]
 
-    async def _mcp_get_prompt(
+    async def _get_prompt(
         self, name: str, arguments: dict[str, Any] | None = None
     ) -> GetPromptResult:
         try:
-            result = await super()._mcp_get_prompt(name, arguments)
+            result = await super()._get_prompt(name, arguments)
             return result
         except NotFoundError:
             async with self.client:

src/fastmcp/server/server.py

@@ -44,7 +44,7 @@ from starlette.routing import BaseRoute, Route
 import fastmcp
 import fastmcp.server
 import fastmcp.settings
-from fastmcp.exceptions import NotFoundError
+from fastmcp.exceptions import DisabledError, NotFoundError
 from fastmcp.prompts import Prompt, PromptManager
 from fastmcp.prompts.prompt import FunctionPrompt
 from fastmcp.resources import Resource, ResourceManager
@@ -291,6 +291,12 @@ class FastMCP(Generic[LifespanResultT]):
             self._cache.set("resources", resources)
         return resources
 
+    async def get_resource(self, key: str) -> Resource:
+        resources = await self.get_resources()
+        if key not in resources:
+            raise NotFoundError(f"Unknown resource: {key}")
+        return resources[key]
+
     async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
         """Get all registered resource templates, indexed by registered key."""
         if (
@@ -311,6 +317,12 @@ class FastMCP(Generic[LifespanResultT]):
             self._cache.set("resource_templates", templates)
         return templates
 
+    async def get_resource_template(self, key: str) -> ResourceTemplate:
+        templates = await self.get_resource_templates()
+        if key not in templates:
+            raise NotFoundError(f"Unknown resource template: {key}")
+        return templates[key]
+
     async def get_prompts(self) -> dict[str, Prompt]:
         """
         List all available prompts.
@@ -330,6 +342,12 @@ class FastMCP(Generic[LifespanResultT]):
             self._cache.set("prompts", prompts)
         return prompts
 
+    async def get_prompt(self, key: str) -> Prompt:
+        prompts = await self.get_prompts()
+        if key not in prompts:
+            raise NotFoundError(f"Unknown prompt: {key}")
+        return prompts[key]
+
     def custom_route(
         self,
         path: str,
@@ -381,7 +399,9 @@ class FastMCP(Generic[LifespanResultT]):
 
         """
         tools = await self.get_tools()
-        return [tool.to_mcp_tool(name=key) for key, tool in tools.items()]
+        return [
+            tool.to_mcp_tool(name=key) for key, tool in tools.items() if tool.enabled
+        ]
 
     async def _mcp_list_resources(self) -> list[MCPResource]:
         """
@@ -391,7 +411,9 @@ class FastMCP(Generic[LifespanResultT]):
         """
         resources = await self.get_resources()
         return [
-            resource.to_mcp_resource(uri=key) for key, resource in resources.items()
+            resource.to_mcp_resource(uri=key)
+            for key, resource in resources.items()
+            if resource.enabled
         ]
 
     async def _mcp_list_resource_templates(self) -> list[MCPResourceTemplate]:
@@ -404,6 +426,7 @@ class FastMCP(Generic[LifespanResultT]):
         return [
             template.to_mcp_template(uriTemplate=key)
             for key, template in templates.items()
+            if template.enabled
         ]
 
     async def _mcp_list_prompts(self) -> list[MCPPrompt]:
@@ -413,12 +436,19 @@ class FastMCP(Generic[LifespanResultT]):
 
         """
         prompts = await self.get_prompts()
-        return [prompt.to_mcp_prompt(name=key) for key, prompt in prompts.items()]
+        return [
+            prompt.to_mcp_prompt(name=key)
+            for key, prompt in prompts.items()
+            if prompt.enabled
+        ]
 
     async def _mcp_call_tool(
         self, key: str, arguments: dict[str, Any]
     ) -> list[TextContent | ImageContent | EmbeddedResource]:
-        """Handle MCP 'callTool' requests.
+        """
+        Handle MCP 'callTool' requests.
+
+        Delegates to _call_tool, which should be overridden by FastMCP subclasses.
 
         Args:
             key: The name of the tool to call
@@ -431,43 +461,109 @@ class FastMCP(Generic[LifespanResultT]):
 
         # Create and use context for the entire call
         with fastmcp.server.context.Context(fastmcp=self):
-            # Get tool, checking first from our tools, then from the mounted servers
-            if self._tool_manager.has_tool(key):
-                return await self._tool_manager.call_tool(key, arguments)
+            try:
+                return await self._call_tool(key, arguments)
+            except DisabledError:
+                # convert to NotFoundError to avoid leaking tool presence
+                raise NotFoundError(f"Unknown tool: {key}")
+            except NotFoundError:
+                # standardize NotFound message
+                raise NotFoundError(f"Unknown tool: {key}")
+
+    async def _call_tool(
+        self, key: str, arguments: dict[str, Any]
+    ) -> list[TextContent | ImageContent | EmbeddedResource]:
+        """
+        Call a tool with raw MCP arguments. FastMCP subclasses should override
+        this method, not _mcp_call_tool.
 
-            # Check mounted servers to see if they have the tool
-            for server in self._mounted_servers.values():
-                if server.match_tool(key):
-                    tool_key = server.strip_tool_prefix(key)
-                    return await server.server._mcp_call_tool(tool_key, arguments)
+        Args:
+            key: The name of the tool to call arguments: Arguments to pass to
+            the tool
 
-            raise NotFoundError(f"Unknown tool: {key}")
+        Returns:
+            List of MCP Content objects containing the tool results
+        """
+
+        # Get tool, checking first from our tools, then from the mounted servers
+        if self._tool_manager.has_tool(key):
+            tool = self._tool_manager.get_tool(key)
+            if not tool.enabled:
+                raise DisabledError(f"Tool {key!r} is disabled")
+            return await self._tool_manager.call_tool(key, arguments)
+
+        # Check mounted servers to see if they have the tool
+        for server in self._mounted_servers.values():
+            if server.match_tool(key):
+                tool_key = server.strip_tool_prefix(key)
+                return await server.server._call_tool(tool_key, arguments)
+
+        raise NotFoundError(f"Unknown tool: {key!r}")
 
     async def _mcp_read_resource(self, uri: AnyUrl | str) -> list[ReadResourceContents]:
+        """
+        Handle MCP 'readResource' requests.
+
+        Delegates to _read_resource, which should be overridden by FastMCP subclasses.
+        """
+        logger.debug("Read resource: %s", uri)
+
+        with fastmcp.server.context.Context(fastmcp=self):
+            try:
+                return await self._read_resource(uri)
+            except DisabledError:
+                # convert to NotFoundError to avoid leaking resource presence
+                raise NotFoundError(f"Unknown resource: {str(uri)!r}")
+            except NotFoundError:
+                # standardize NotFound message
+                raise NotFoundError(f"Unknown resource: {str(uri)!r}")
+
+    async def _read_resource(self, uri: AnyUrl | str) -> list[ReadResourceContents]:
         """
         Read a resource by URI, in the format expected by the low-level MCP
         server.
         """
-        with fastmcp.server.context.Context(fastmcp=self):
-            if self._resource_manager.has_resource(uri):
-                resource = await self._resource_manager.get_resource(uri)
-                content = await self._resource_manager.read_resource(uri)
-                return [
-                    ReadResourceContents(
-                        content=content,
-                        mime_type=resource.mime_type,
-                    )
-                ]
+        if self._resource_manager.has_resource(uri):
+            resource = await self._resource_manager.get_resource(uri)
+            if not resource.enabled:
+                raise DisabledError(f"Resource {str(uri)!r} is disabled")
+            content = await self._resource_manager.read_resource(uri)
+            return [
+                ReadResourceContents(
+                    content=content,
+                    mime_type=resource.mime_type,
+                )
+            ]
+        else:
+            for server in self._mounted_servers.values():
+                if server.match_resource(str(uri)):
+                    new_uri = server.strip_resource_prefix(str(uri))
+                    return await server.server._mcp_read_resource(new_uri)
             else:
-                for server in self._mounted_servers.values():
-                    if server.match_resource(str(uri)):
-                        new_uri = server.strip_resource_prefix(str(uri))
-                        return await server.server._mcp_read_resource(new_uri)
-                else:
-                    raise NotFoundError(f"Unknown resource: {uri}")
+                raise NotFoundError(f"Unknown resource: {uri}")
 
     async def _mcp_get_prompt(
         self, name: str, arguments: dict[str, Any] | None = None
+    ) -> GetPromptResult:
+        """
+        Handle MCP 'getPrompt' requests.
+
+        Delegates to _get_prompt, which should be overridden by FastMCP subclasses.
+        """
+        logger.debug("Get prompt: %s with %s", name, arguments)
+
+        with fastmcp.server.context.Context(fastmcp=self):
+            try:
+                return await self._get_prompt(name, arguments)
+            except DisabledError:
+                # convert to NotFoundError to avoid leaking prompt presence
+                raise NotFoundError(f"Unknown prompt: {name}")
+            except NotFoundError:
+                # standardize NotFound message
+                raise NotFoundError(f"Unknown prompt: {name}")
+
+    async def _get_prompt(
+        self, name: str, arguments: dict[str, Any] | None = None
     ) -> GetPromptResult:
         """Handle MCP 'getPrompt' requests.
 
@@ -480,19 +576,20 @@ class FastMCP(Generic[LifespanResultT]):
         """
         logger.debug("Get prompt: %s with %s", name, arguments)
 
-        # Create and use context for the entire call
-        with fastmcp.server.context.Context(fastmcp=self):
-            # Get prompt, checking first from our prompts, then from the mounted servers
-            if self._prompt_manager.has_prompt(name):
-                return await self._prompt_manager.render_prompt(name, arguments)
+        # Get prompt, checking first from our prompts, then from the mounted servers
+        if self._prompt_manager.has_prompt(name):
+            prompt = self._prompt_manager.get_prompt(name)
+            if not prompt.enabled:
+                raise DisabledError(f"Prompt {name!r} is disabled")
+            return await self._prompt_manager.render_prompt(name, arguments)
 
-            # Check mounted servers to see if they have the prompt
-            for server in self._mounted_servers.values():
-                if server.match_prompt(name):
-                    prompt_name = server.strip_prompt_prefix(name)
-                    return await server.server._mcp_get_prompt(prompt_name, arguments)
+        # Check mounted servers to see if they have the prompt
+        for server in self._mounted_servers.values():
+            if server.match_prompt(name):
+                prompt_name = server.strip_prompt_prefix(name)
+                return await server.server._mcp_get_prompt(prompt_name, arguments)
 
-            raise NotFoundError(f"Unknown prompt: {name}")
+        raise NotFoundError(f"Unknown prompt: {name}")
 
     def add_tool(self, tool: Tool) -> None:
         """Add a tool to the server.