add authorizer plugin to enable fine granded authorization checks on tools/resources/prompts

Author: davemssavage

src/mcp/server/fastmcp/authorizer.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+import abc
+from typing import TYPE_CHECKING, Any
+from pydantic import AnyUrl
+
+from mcp.shared.context import LifespanContextT, RequestT
+
+if TYPE_CHECKING:
+    from mcp.server.fastmcp.server import Context
+    from mcp.server.session import ServerSessionT
+
+
+class Authorizer:
+    __metaclass__ = abc.ABCMeta
+
+    @abc.abstractmethod
+    def permit_get_tool(self, name: str) -> bool:
+        """Check if the specified tool can be retrieved from the associated mcp server"""
+        return False
+    
+    @abc.abstractmethod
+    def permit_list_tool(self, name: str) -> bool:
+        """Check if the specified tool can be listed from the associated mcp server"""
+        return False
+
+    @abc.abstractmethod
+    def permit_call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any],
+        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
+    ) -> bool:
+        """Check if the specified tool can be called from the associated mcp server"""
+        return False
+
+    @abc.abstractmethod
+    def permit_get_resource(self, resource: AnyUrl | str) -> bool:
+        """Check if the specified resource can be retrieved from the associated mcp server"""
+        return False
+
+    @abc.abstractmethod
+    def permit_create_resource(self, uri: str, params: dict[str, Any]) -> bool:
+        """Check if the specified resource can be created on the associated mcp server"""
+        return False
+
+    @abc.abstractmethod
+    def permit_list_resource(self, resource: AnyUrl | str) -> bool:
+        """Check if the specified resource can be listed from the associated mcp server"""
+        return False
+
+    @abc.abstractmethod
+    def permit_list_template(self, resource: AnyUrl | str) -> bool:
+        """Check if the specified template can be listed from the associated mcp server"""
+        return False
+    
+    @abc.abstractmethod
+    def permit_get_prompt(self, name: str) -> bool:
+        """Check if the specified prompt can be retrieved from the associated mcp server"""
+        return False
+        
+    @abc.abstractmethod
+    def permit_list_prompt(self, name: str) -> bool:
+        """Check if the specified prompt can be listed from the associated mcp server"""
+        return False
+    
+    @abc.abstractmethod
+    def permit_render_prompt(self, name: str,  arguments: dict[str, Any] | None = None) -> bool:
+        """Check if the specified prompt can be rendered from the associated mcp server"""
+        return False
+    
+class AllAllAuthorizer(Authorizer):
+    def permit_get_tool(self, name: str) -> bool:
+        return True
+    
+    def permit_list_tool(self, name: str) -> bool:
+        return True
+
+    def permit_call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any],
+        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
+    ) -> bool:
+        return True
+
+    def permit_get_resource(self, resource: AnyUrl | str) -> bool:
+        return True
+
+    def permit_create_resource(self, uri: str, params: dict[str, Any]) -> bool:
+        return True
+    
+    def permit_list_resource(self, resource: AnyUrl | str) -> bool:
+        return True
+
+    def permit_list_template(self, resource: AnyUrl | str) -> bool:
+        return True
+    
+    def permit_get_prompt(self, name: str) -> bool:
+        return True
+    
+    def permit_list_prompt(self, name: str) -> bool:
+        return True
+
+    def permit_render_prompt(self, name: str,  arguments: dict[str, Any] | None = None) -> bool:
+        return True
+

src/mcp/server/fastmcp/prompts/manager.py

@@ -2,6 +2,7 @@
 
 from typing import Any
 
+from mcp.server.fastmcp.authorizer import AllAllAuthorizer, Authorizer
 from mcp.server.fastmcp.prompts.base import Message, Prompt
 from mcp.server.fastmcp.utilities.logging import get_logger
 
@@ -11,17 +12,25 @@
 class PromptManager:
     """Manages FastMCP prompts."""
 
-    def __init__(self, warn_on_duplicate_prompts: bool = True):
+    def __init__(
+        self,
+        warn_on_duplicate_prompts: bool = True,
+        authorizer: Authorizer = AllAllAuthorizer(),
+    ):
         self._prompts: dict[str, Prompt] = {}
+        self._authorizer = authorizer
         self.warn_on_duplicate_prompts = warn_on_duplicate_prompts
 
     def get_prompt(self, name: str) -> Prompt | None:
         """Get prompt by name."""
-        return self._prompts.get(name)
+        if self._authorizer.permit_get_prompt(name):
+            return self._prompts.get(name)
+        else:
+            return None
 
     def list_prompts(self) -> list[Prompt]:
         """List all registered prompts."""
-        return list(self._prompts.values())
+        return [prompt for name, prompt in self._prompts.items() if self._authorizer.permit_list_prompt(name)]
 
     def add_prompt(
         self,
@@ -44,5 +53,7 @@ async def render_prompt(self, name: str, arguments: dict[str, Any] | None = None
         prompt = self.get_prompt(name)
         if not prompt:
             raise ValueError(f"Unknown prompt: {name}")
-
-        return await prompt.render(arguments)
+        if self._authorizer.permit_render_prompt(name, arguments):
+            return await prompt.render(arguments)
+        else:
+            raise ValueError(f"Unknown prompt: {name}")

src/mcp/server/fastmcp/resources/resource_manager.py

@@ -5,6 +5,7 @@
 
 from pydantic import AnyUrl
 
+from mcp.server.fastmcp.authorizer import AllAllAuthorizer, Authorizer
 from mcp.server.fastmcp.resources.base import Resource
 from mcp.server.fastmcp.resources.templates import ResourceTemplate
 from mcp.server.fastmcp.utilities.logging import get_logger
@@ -15,10 +16,15 @@
 class ResourceManager:
     """Manages FastMCP resources."""
 
-    def __init__(self, warn_on_duplicate_resources: bool = True):
+    def __init__(
+        self,
+        warn_on_duplicate_resources: bool = True,
+        authorizer: Authorizer = AllAllAuthorizer(),
+    ):
         self._resources: dict[str, Resource] = {}
         self._templates: dict[str, ResourceTemplate] = {}
         self.warn_on_duplicate_resources = warn_on_duplicate_resources
+        self._authorizer = authorizer
 
     def add_resource(self, resource: Resource) -> Resource:
         """Add a resource to the manager.
@@ -74,13 +80,19 @@ async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
 
         # First check concrete resources
         if resource := self._resources.get(uri_str):
-            return resource
+            if self._authorizer.permit_get_resource(uri_str):
+                return resource
+            else:
+                raise ValueError(f"Unknown resource: {uri}")
 
         # Then check templates
         for template in self._templates.values():
             if params := template.matches(uri_str):
                 try:
-                    return await template.create_resource(uri_str, params)
+                    if self._authorizer.permit_create_resource(uri_str, params):
+                        return await template.create_resource(uri_str, params)
+                    else:
+                        raise ValueError(f"Unknown resource: {uri}")
                 except Exception as e:
                     raise ValueError(f"Error creating resource from template: {e}")
 
@@ -89,9 +101,9 @@ async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
     def list_resources(self) -> list[Resource]:
         """List all registered resources."""
         logger.debug("Listing resources", extra={"count": len(self._resources)})
-        return list(self._resources.values())
+        return [resource for uri, resource in self._resources.items() if self._authorizer.permit_list_resource(uri)]
 
     def list_templates(self) -> list[ResourceTemplate]:
         """List all registered templates."""
         logger.debug("Listing templates", extra={"count": len(self._templates)})
-        return list(self._templates.values())
+        return [template for uri, template in self._templates.items() if self._authorizer.permit_list_template(uri)]

src/mcp/server/fastmcp/server.py

@@ -33,6 +33,7 @@
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier
 from mcp.server.auth.settings import AuthSettings
 from mcp.server.elicitation import ElicitationResult, ElicitSchemaModelT, elicit_with_validation
+from mcp.server.fastmcp.authorizer import AllAllAuthorizer, Authorizer
 from mcp.server.fastmcp.exceptions import ResourceError
 from mcp.server.fastmcp.prompts import Prompt, PromptManager
 from mcp.server.fastmcp.resources import FunctionResource, Resource, ResourceManager
@@ -120,6 +121,8 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     # Transport security settings (DNS rebinding protection)
     transport_security: TransportSecuritySettings | None = None
 
+    authorizer: Authorizer = AllAllAuthorizer()
+
 
 def lifespan_wrapper(
     app: FastMCP,
@@ -152,9 +155,19 @@ def __init__(
             instructions=instructions,
             lifespan=(lifespan_wrapper(self, self.settings.lifespan) if self.settings.lifespan else default_lifespan),
         )
-        self._tool_manager = ToolManager(tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools)
-        self._resource_manager = ResourceManager(warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources)
-        self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)
+        self._tool_manager = ToolManager(
+            tools=tools,
+            warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools,
+            authorizer=self.settings.authorizer,
+        )
+        self._resource_manager = ResourceManager(
+            warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources,
+            authorizer=self.settings.authorizer,
+        )
+        self._prompt_manager = PromptManager(
+            warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts,
+            authorizer=self.settings.authorizer,
+        )
         # Validate auth configuration
         if self.settings.auth is not None:
             if auth_server_provider and token_verifier:

src/mcp/server/fastmcp/tools/tool_manager.py

@@ -3,6 +3,7 @@
 from collections.abc import Callable
 from typing import TYPE_CHECKING, Any
 
+from mcp.server.fastmcp.authorizer import AllAllAuthorizer, Authorizer
 from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.fastmcp.tools.base import Tool
 from mcp.server.fastmcp.utilities.logging import get_logger
@@ -24,6 +25,7 @@ def __init__(
         warn_on_duplicate_tools: bool = True,
         *,
         tools: list[Tool] | None = None,
+        authorizer: Authorizer = AllAllAuthorizer(),
     ):
         self._tools: dict[str, Tool] = {}
         if tools is not None:
@@ -32,15 +34,19 @@ def __init__(
                     logger.warning(f"Tool already exists: {tool.name}")
                 self._tools[tool.name] = tool
 
-        self.warn_on_duplicate_tools = warn_on_duplicate_tools
+        self.warn_on_duplicate_tools = (warn_on_duplicate_tools,)
+        self._authorizer = authorizer
 
     def get_tool(self, name: str) -> Tool | None:
         """Get tool by name."""
-        return self._tools.get(name)
+        if self._authorizer.permit_get_tool(name):
+            return self._tools.get(name)
+        else:
+            return None
 
     def list_tools(self) -> list[Tool]:
         """List all registered tools."""
-        return list(self._tools.values())
+        return [tool for name, tool in self._tools.items() if self._authorizer.permit_list_tool(name)]
 
     def add_tool(
         self,
@@ -67,8 +73,8 @@ async def call_tool(
         context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
     ) -> Any:
         """Call a tool by name with arguments."""
-        tool = self.get_tool(name)
-        if not tool:
+        tool = self._tools.get(name)
+        if not tool or not self._authorizer.permit_call_tool(name, arguments, context):
             raise ToolError(f"Unknown tool: {name}")
 
         return await tool.run(arguments, context=context)

tests/server/fastmcp/test_tool_manager.py

@@ -5,6 +5,7 @@
 from pydantic import BaseModel
 
 from mcp.server.fastmcp import Context, FastMCP
+from mcp.server.fastmcp.authorizer import Authorizer
 from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.fastmcp.tools import Tool, ToolManager
 from mcp.server.fastmcp.utilities.func_metadata import ArgModelBase, FuncMetadata
@@ -171,7 +172,7 @@ def f(x: int) -> int:
 
         manager = ToolManager()
         manager.add_tool(f)
-        manager.warn_on_duplicate_tools = False
+        manager.warn_on_duplicate_tools = False  # type: ignore
         with caplog.at_level(logging.WARNING):
             manager.add_tool(f)
             assert "Tool already exists: f" not in caplog.text
@@ -311,6 +312,30 @@ def name_shrimp(tank: MyShrimpTank, ctx: Context) -> list[str]:
         )
         assert result == ["rex", "gertrude"]
 
+    @pytest.mark.anyio
+    async def test_call_tool_not_permitted(self):
+        async def double(n: int) -> int:
+            """Double a number."""
+            return n * 2
+
+        class TestAuthorizer(Authorizer):
+            allow: bool = True
+
+            def permit_list_tool(self, name):
+                return self.allow
+
+            def permit_call_tool(self, name, arguments, context=None):
+                return self.allow
+
+        authorizer = TestAuthorizer()
+        manager = ToolManager(authorizer=authorizer)
+        manager.add_tool(double)
+        result = await manager.call_tool("double", {"n": 5})
+        assert result == 10
+        authorizer.allow = False
+        with pytest.raises(ToolError, match="Unknown tool: double"):
+            await manager.call_tool("double", {"n": 5})
+
 
 class TestToolSchema:
     @pytest.mark.anyio