pass context to all permit methods to allow checking other request params such as http headers

davemssavage · davemssavage · commit 092d769f3455 · 2025-06-28T09:33:15.000Z
diff --git a/src/mcp/server/fastmcp/authorizer.py b/src/mcp/server/fastmcp/authorizer.py
@@ -4,24 +4,28 @@
 from typing import TYPE_CHECKING, Any
 
 from pydantic import AnyUrl
+from starlette.requests import Request
 
-from mcp.shared.context import LifespanContextT, RequestT
+from mcp.server.session import ServerSession
 
 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:
+    def permit_get_tool(self, name: str, context: Context[ServerSession, object, Request] | None = None) -> 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:
+    def permit_list_tool(
+        self,
+        name: str,
+        context: Context[ServerSession, object, Request] | None = None,
+    ) -> bool:
         """Check if the specified tool can be listed from the associated mcp server"""
         return False
 
@@ -30,79 +34,105 @@ def permit_call_tool(
         self,
         name: str,
         arguments: dict[str, Any],
-        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
+        context: Context[ServerSession, object, Request] | 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:
+    def permit_get_resource(
+        self, resource: AnyUrl | str, context: Context[ServerSession, object, Request] | None = None
+    ) -> 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:
+    def permit_create_resource(
+        self, uri: str, params: dict[str, Any], context: Context[ServerSession, object, Request] | None = None
+    ) -> 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:
+    def permit_list_resource(
+        self, resource: AnyUrl | str, context: Context[ServerSession, object, Request] | None = None
+    ) -> 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:
+    def permit_list_template(
+        self, resource: AnyUrl | str, context: Context[ServerSession, object, Request] | None = None
+    ) -> 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:
+    def permit_get_prompt(self, name: str, context: Context[ServerSession, object, Request] | None = None) -> 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:
+    def permit_list_prompt(self, name: str, context: Context[ServerSession, object, Request] | None = None) -> 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:
+    def permit_render_prompt(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        context: Context[ServerSession, object, Request] | 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:
+    def permit_get_tool(self, name: str, context: Context[ServerSession, object, Request] | None = None) -> bool:
         return True
 
-    def permit_list_tool(self, name: str) -> bool:
+    def permit_list_tool(self, name: str, context: Context[ServerSession, object, Request] | None = None) -> bool:
         return True
 
     def permit_call_tool(
         self,
         name: str,
         arguments: dict[str, Any],
-        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
+        context: Context[ServerSession, object, Request] | None = None,
     ) -> bool:
         return True
 
-    def permit_get_resource(self, resource: AnyUrl | str) -> bool:
+    def permit_get_resource(
+        self, resource: AnyUrl | str, context: Context[ServerSession, object, Request] | None = None
+    ) -> bool:
         return True
 
-    def permit_create_resource(self, uri: str, params: dict[str, Any]) -> bool:
+    def permit_create_resource(
+        self, uri: str, params: dict[str, Any], context: Context[ServerSession, object, Request] | None = None
+    ) -> bool:
         return True
 
-    def permit_list_resource(self, resource: AnyUrl | str) -> bool:
+    def permit_list_resource(
+        self, resource: AnyUrl | str, context: Context[ServerSession, object, Request] | None = None
+    ) -> bool:
         return True
 
-    def permit_list_template(self, resource: AnyUrl | str) -> bool:
+    def permit_list_template(
+        self, resource: AnyUrl | str, context: Context[ServerSession, object, Request] | None = None
+    ) -> bool:
         return True
 
-    def permit_get_prompt(self, name: str) -> bool:
+    def permit_get_prompt(self, name: str, context: Context[ServerSession, object, Request] | None = None) -> bool:
         return True
 
-    def permit_list_prompt(self, name: str) -> bool:
+    def permit_list_prompt(self, name: str, context: Context[ServerSession, object, Request] | None = None) -> bool:
         return True
 
-    def permit_render_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> bool:
+    def permit_render_prompt(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        context: Context[ServerSession, object, Request] | None = None,
+    ) -> bool:
         return True
diff --git a/src/mcp/server/fastmcp/prompts/manager.py b/src/mcp/server/fastmcp/prompts/manager.py
@@ -1,10 +1,18 @@
 """Prompt management functionality."""
 
-from typing import Any
+from __future__ import annotations as _annotations
+
+from typing import TYPE_CHECKING, Any
+
+from starlette.requests import Request
 
 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
+from mcp.server.session import ServerSession
+
+if TYPE_CHECKING:
+    from mcp.server.fastmcp.server import Context
 
 logger = get_logger(__name__)
 
@@ -21,16 +29,16 @@ def __init__(
         self._authorizer = authorizer
         self.warn_on_duplicate_prompts = warn_on_duplicate_prompts
 
-    def get_prompt(self, name: str) -> Prompt | None:
+    def get_prompt(self, name: str, context: Context[ServerSession, object, Request] | None = None) -> Prompt | None:
         """Get prompt by name."""
-        if self._authorizer.permit_get_prompt(name):
+        if self._authorizer.permit_get_prompt(name, context):
             return self._prompts.get(name)
         else:
             return None
 
-    def list_prompts(self) -> list[Prompt]:
+    def list_prompts(self, context: Context[ServerSession, object, Request] | None = None) -> list[Prompt]:
         """List all registered prompts."""
-        return [prompt for name, prompt in self._prompts.items() if self._authorizer.permit_list_prompt(name)]
+        return [prompt for name, prompt in self._prompts.items() if self._authorizer.permit_list_prompt(name, context)]
 
     def add_prompt(
         self,
@@ -48,12 +56,17 @@ def add_prompt(
         self._prompts[prompt.name] = prompt
         return prompt
 
-    async def render_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> list[Message]:
+    async def render_prompt(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        context: Context[ServerSession, object, Request] | None = None,
+    ) -> list[Message]:
         """Render a prompt by name with arguments."""
         prompt = self.get_prompt(name)
         if not prompt:
             raise ValueError(f"Unknown prompt: {name}")
-        if self._authorizer.permit_render_prompt(name, arguments):
+        if self._authorizer.permit_render_prompt(name, arguments, context):
             return await prompt.render(arguments)
         else:
             raise ValueError(f"Unknown prompt: {name}")
diff --git a/src/mcp/server/fastmcp/resources/resource_manager.py b/src/mcp/server/fastmcp/resources/resource_manager.py
@@ -1,14 +1,21 @@
 """Resource manager functionality."""
 
+from __future__ import annotations as _annotations
+
 from collections.abc import Callable
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from pydantic import AnyUrl
+from starlette.requests import Request
 
 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
+from mcp.server.session import ServerSession
+
+if TYPE_CHECKING:
+    from mcp.server.fastmcp.server import Context
 
 logger = get_logger(__name__)
 
@@ -73,14 +80,16 @@ def add_template(
         self._templates[template.uri_template] = template
         return template
 
-    async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
+    async def get_resource(
+        self, uri: AnyUrl | str, context: Context[ServerSession, object, Request] | None = None
+    ) -> Resource | None:
         """Get resource by URI, checking concrete resources first, then templates."""
         uri_str = str(uri)
         logger.debug("Getting resource", extra={"uri": uri_str})
 
         # First check concrete resources
         if resource := self._resources.get(uri_str):
-            if self._authorizer.permit_get_resource(uri_str):
+            if self._authorizer.permit_get_resource(uri_str, context):
                 return resource
             else:
                 raise ValueError(f"Unknown resource: {uri}")
@@ -98,12 +107,16 @@ async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
 
         raise ValueError(f"Unknown resource: {uri}")
 
-    def list_resources(self) -> list[Resource]:
+    def list_resources(self, context: Context[ServerSession, object, Request] | None = None) -> list[Resource]:
         """List all registered resources."""
         logger.debug("Listing resources", extra={"count": len(self._resources)})
-        return [resource for uri, resource in self._resources.items() if self._authorizer.permit_list_resource(uri)]
+        return [
+            resource for uri, resource in self._resources.items() if self._authorizer.permit_list_resource(uri, context)
+        ]
 
-    def list_templates(self) -> list[ResourceTemplate]:
+    def list_templates(self, context: Context[ServerSession, object, Request] | None = None) -> list[ResourceTemplate]:
         """List all registered templates."""
         logger.debug("Listing templates", extra={"count": len(self._templates)})
-        return [template for uri, template in self._templates.items() if self._authorizer.permit_list_template(uri)]
+        return [
+            template for uri, template in self._templates.items() if self._authorizer.permit_list_template(uri, context)
+        ]
diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py
@@ -257,7 +257,8 @@ def _setup_handlers(self) -> None:
 
     async def list_tools(self) -> list[MCPTool]:
         """List all available tools."""
-        tools = self._tool_manager.list_tools()
+        context = self.get_context()
+        tools = self._tool_manager.list_tools(context)
         return [
             MCPTool(
                 name=info.name,
@@ -289,8 +290,8 @@ async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[Cont
 
     async def list_resources(self) -> list[MCPResource]:
         """List all available resources."""
-
-        resources = self._resource_manager.list_resources()
+        context = self.get_context()
+        resources = self._resource_manager.list_resources(context)
         return [
             MCPResource(
                 uri=resource.uri,
@@ -303,7 +304,8 @@ async def list_resources(self) -> list[MCPResource]:
         ]
 
     async def list_resource_templates(self) -> list[MCPResourceTemplate]:
-        templates = self._resource_manager.list_templates()
+        context = self.get_context()
+        templates = self._resource_manager.list_templates(context)
         return [
             MCPResourceTemplate(
                 uriTemplate=template.uri_template,
@@ -316,8 +318,8 @@ async def list_resource_templates(self) -> list[MCPResourceTemplate]:
 
     async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]:
         """Read a resource by URI."""
-
-        resource = await self._resource_manager.get_resource(uri)
+        context = self.get_context()
+        resource = await self._resource_manager.get_resource(uri, context)
         if not resource:
             raise ResourceError(f"Unknown resource: {uri}")
 
@@ -924,9 +926,9 @@ async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) ->
             lifespan=lambda app: self.session_manager.run(),
         )
 
-    async def list_prompts(self) -> list[MCPPrompt]:
+    async def list_prompts(self, context: Context[ServerSession, object, Request] | None = None) -> list[MCPPrompt]:
         """List all available prompts."""
-        prompts = self._prompt_manager.list_prompts()
+        prompts = self._prompt_manager.list_prompts(context)
         return [
             MCPPrompt(
                 name=prompt.name,
@@ -944,10 +946,15 @@ async def list_prompts(self) -> list[MCPPrompt]:
             for prompt in prompts
         ]
 
-    async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> GetPromptResult:
+    async def get_prompt(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        context: Context[ServerSession, object, Request] | None = None,
+    ) -> GetPromptResult:
         """Get a prompt by name with arguments."""
         try:
-            messages = await self._prompt_manager.render_prompt(name, arguments)
+            messages = await self._prompt_manager.render_prompt(name, arguments, context)
 
             return GetPromptResult(messages=pydantic_core.to_jsonable_python(messages))
         except Exception as e:
diff --git a/src/mcp/server/fastmcp/tools/tool_manager.py b/src/mcp/server/fastmcp/tools/tool_manager.py
diff --git a/tests/server/fastmcp/test_tool_manager.py b/tests/server/fastmcp/test_tool_manager.py
