SEP: Prefix Filtering for Listing Resource / Templates

examples/servers/simple-resource/mcp_simple_resource/server.py

@@ -32,7 +32,9 @@ def main(port: int, transport: str) -> int:
     app = Server("mcp-simple-resource")
 
     @app.list_resources()
-    async def list_resources() -> list[types.Resource]:
+    async def list_resources(
+        request: types.ListResourcesRequest,
+    ) -> list[types.Resource]:
         return [
             types.Resource(
                 uri=FileUrl(f"file:///{name}.txt"),

src/mcp/client/session.py

@@ -221,25 +221,35 @@ async def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResul
             types.EmptyResult,
         )
 
-    async def list_resources(self, cursor: str | None = None) -> types.ListResourcesResult:
+    async def list_resources(self, prefix: str | None = None, cursor: str | None = None) -> types.ListResourcesResult:
         """Send a resources/list request."""
+        params = None
+        if cursor is not None or prefix is not None:
+            params = types.ListResourcesRequestParams(prefix=prefix, cursor=cursor)
         return await self.send_request(
             types.ClientRequest(
                 types.ListResourcesRequest(
                     method="resources/list",
-                    params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
+                    params=params,
                 )
             ),
             types.ListResourcesResult,
         )
 
-    async def list_resource_templates(self, cursor: str | None = None) -> types.ListResourceTemplatesResult:
+    async def list_resource_templates(
+        self,
+        prefix: str | None = None,
+        cursor: str | None = None,
+    ) -> types.ListResourceTemplatesResult:
         """Send a resources/templates/list request."""
+        params = None
+        if cursor is not None or prefix is not None:
+            params = types.ListResourceTemplatesRequestParams(prefix=prefix, cursor=cursor)
         return await self.send_request(
             types.ClientRequest(
                 types.ListResourceTemplatesRequest(
                     method="resources/templates/list",
-                    params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
+                    params=params,
                 )
             ),
             types.ListResourceTemplatesResult,

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

@@ -86,12 +86,18 @@ async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
 
         raise ValueError(f"Unknown resource: {uri}")
 
-    def list_resources(self) -> list[Resource]:
-        """List all registered resources."""
-        logger.debug("Listing resources", extra={"count": len(self._resources)})
-        return list(self._resources.values())
-
-    def list_templates(self) -> list[ResourceTemplate]:
-        """List all registered templates."""
-        logger.debug("Listing templates", extra={"count": len(self._templates)})
-        return list(self._templates.values())
+    def list_resources(self, prefix: str | None = None) -> list[Resource]:
+        """List all registered resources, optionally filtered by URI prefix."""
+        resources = list(self._resources.values())
+        if prefix:
+            resources = [r for r in resources if str(r.uri).startswith(prefix)]
+        logger.debug("Listing resources", extra={"count": len(resources), "prefix": prefix})
+        return resources
+
+    def list_templates(self, prefix: str | None = None) -> list[ResourceTemplate]:
+        """List all registered templates, optionally filtered by URI template prefix."""
+        templates = list(self._templates.values())
+        if prefix:
+            templates = [t for t in templates if t.matches_prefix(prefix)]
+        logger.debug("Listing templates", extra={"count": len(templates), "prefix": prefix})
+        return templates

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

@@ -63,6 +63,44 @@ def matches(self, uri: str) -> dict[str, Any] | None:
             return match.groupdict()
         return None
 
+    def matches_prefix(self, prefix: str) -> bool:
+        """Check if this template could match URIs with the given prefix."""
+
+        # First, simple check: does the template itself start with the prefix?
+        if self.uri_template.startswith(prefix):
+            return True
+
+        template_segments = self.uri_template.split("/")
+        prefix_segments = prefix.split("/")
+
+        # Handle trailing slash - it creates an empty last segment
+        has_trailing_slash = prefix.endswith("/") and prefix_segments[-1] == ""
+        if has_trailing_slash:
+            # Remove the empty segment for comparison
+            prefix_segments = prefix_segments[:-1]
+            # Template must have more segments to generate something "under" this path
+            if len(template_segments) <= len(prefix_segments):
+                return False
+        else:
+            # Without trailing slash, prefix can't have more segments than template
+            if len(prefix_segments) > len(template_segments):
+                return False
+
+        # Compare each segment
+        for i, prefix_seg in enumerate(prefix_segments):
+            template_seg = template_segments[i]
+
+            # If template segment is a parameter, it can match any value
+            if template_seg.startswith("{") and template_seg.endswith("}"):
+                continue
+
+            # If both are literals, they must match exactly
+            if template_seg != prefix_seg:
+                return False
+
+        # All prefix segments matched
+        return True
+
     async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
         """Create a resource from the template with the given parameters."""
         try:

src/mcp/server/fastmcp/server.py

@@ -21,6 +21,7 @@
 from starlette.routing import Mount, Route
 from starlette.types import Receive, Scope, Send
 
+from mcp import types
 from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
 from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier
@@ -297,10 +298,12 @@ async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[Cont
         context = self.get_context()
         return await self._tool_manager.call_tool(name, arguments, context=context, convert_result=True)
 
-    async def list_resources(self) -> list[MCPResource]:
-        """List all available resources."""
-
-        resources = self._resource_manager.list_resources()
+    async def list_resources(self, request: types.ListResourcesRequest | None = None) -> list[MCPResource]:
+        """List all available resources, optionally filtered by prefix."""
+        prefix = None
+        if request and request.params:
+            prefix = request.params.prefix
+        resources = self._resource_manager.list_resources(prefix=prefix)
         return [
             MCPResource(
                 uri=resource.uri,
@@ -312,8 +315,14 @@ async def list_resources(self) -> list[MCPResource]:
             for resource in resources
         ]
 
-    async def list_resource_templates(self) -> list[MCPResourceTemplate]:
-        templates = self._resource_manager.list_templates()
+    async def list_resource_templates(
+        self, request: types.ListResourceTemplatesRequest | None = None
+    ) -> list[MCPResourceTemplate]:
+        """List all available resource templates, optionally filtered by prefix."""
+        prefix = None
+        if request and request.params:
+            prefix = request.params.prefix
+        templates = self._resource_manager.list_templates(prefix=prefix)
         return [
             MCPResourceTemplate(
                 uriTemplate=template.uri_template,

src/mcp/server/lowlevel/server.py

@@ -259,11 +259,11 @@ async def handler(req: types.GetPromptRequest):
         return decorator
 
     def list_resources(self):
-        def decorator(func: Callable[[], Awaitable[list[types.Resource]]]):
+        def decorator(func: Callable[[types.ListResourcesRequest], Awaitable[list[types.Resource]]]):
             logger.debug("Registering handler for ListResourcesRequest")
 
-            async def handler(_: Any):
-                resources = await func()
+            async def handler(request: types.ListResourcesRequest):
+                resources = await func(request)
                 return types.ServerResult(types.ListResourcesResult(resources=resources))
 
             self.request_handlers[types.ListResourcesRequest] = handler
@@ -272,11 +272,11 @@ async def handler(_: Any):
         return decorator
 
     def list_resource_templates(self):
-        def decorator(func: Callable[[], Awaitable[list[types.ResourceTemplate]]]):
+        def decorator(func: Callable[[types.ListResourceTemplatesRequest], Awaitable[list[types.ResourceTemplate]]]):
             logger.debug("Registering handler for ListResourceTemplatesRequest")
 
-            async def handler(_: Any):
-                templates = await func()
+            async def handler(request: types.ListResourceTemplatesRequest):
+                templates = await func(request)
                 return types.ServerResult(types.ListResourceTemplatesResult(resourceTemplates=templates))
 
             self.request_handlers[types.ListResourceTemplatesRequest] = handler

src/mcp/types.py

@@ -63,6 +63,20 @@ class PaginatedRequestParams(RequestParams):
     """
 
 
+class ListResourcesRequestParams(PaginatedRequestParams):
+    """Parameters for listing resources with optional prefix filtering."""
+
+    prefix: str | None = None
+    """Optional prefix to filter resources by URI."""
+
+
+class ListResourceTemplatesRequestParams(PaginatedRequestParams):
+    """Parameters for listing resource templates with optional prefix filtering."""
+
+    prefix: str | None = None
+    """Optional prefix to filter resource templates by URI template."""
+
+
 class NotificationParams(BaseModel):
     class Meta(BaseModel):
         model_config = ConfigDict(extra="allow")
@@ -394,10 +408,11 @@ class ProgressNotification(Notification[ProgressNotificationParams, Literal["not
     params: ProgressNotificationParams
 
 
-class ListResourcesRequest(PaginatedRequest[Literal["resources/list"]]):
+class ListResourcesRequest(Request[ListResourcesRequestParams | None, Literal["resources/list"]]):
     """Sent from the client to request a list of resources the server has."""
 
     method: Literal["resources/list"]
+    params: ListResourcesRequestParams | None = None
 
 
 class Annotations(BaseModel):
@@ -461,10 +476,13 @@ class ListResourcesResult(PaginatedResult):
     resources: list[Resource]
 
 
-class ListResourceTemplatesRequest(PaginatedRequest[Literal["resources/templates/list"]]):
+class ListResourceTemplatesRequest(
+    Request[ListResourceTemplatesRequestParams | None, Literal["resources/templates/list"]]
+):
     """Sent from the client to request a list of resource templates the server has."""
 
     method: Literal["resources/templates/list"]
+    params: ListResourceTemplatesRequestParams | None = None
 
 
 class ListResourceTemplatesResult(PaginatedResult):

tests/issues/test_152_resource_mime_type.py

@@ -79,7 +79,7 @@ async def test_lowlevel_resource_mime_type():
     ]
 
     @server.list_resources()
-    async def handle_list_resources():
+    async def handle_list_resources(request: types.ListResourcesRequest):
         return test_resources
 
     @server.read_resource()

tests/server/fastmcp/resources/test_resource_manager.py

@@ -139,3 +139,97 @@ def test_list_resources(self, temp_file: Path):
         resources = manager.list_resources()
         assert len(resources) == 2
         assert resources == [resource1, resource2]
+
+    def test_list_resources_with_prefix(self, temp_file: Path):
+        """Test listing resources with prefix filtering."""
+        manager = ResourceManager()
+
+        # Add resources with different URIs
+        resource1 = FileResource(
+            uri=FileUrl("file:///data/images/test.jpg"),
+            name="test_image",
+            path=temp_file,
+        )
+        resource2 = FileResource(
+            uri=FileUrl("file:///data/docs/test.txt"),
+            name="test_doc",
+            path=temp_file,
+        )
+        resource3 = FileResource(
+            uri=FileUrl("file:///other/test.txt"),
+            name="other_test",
+            path=temp_file,
+        )
+
+        manager.add_resource(resource1)
+        manager.add_resource(resource2)
+        manager.add_resource(resource3)
+
+        # Test prefix filtering
+        data_resources = manager.list_resources(prefix="file:///data/")
+        assert len(data_resources) == 2
+        assert resource1 in data_resources
+        assert resource2 in data_resources
+
+        # More specific prefix
+        image_resources = manager.list_resources(prefix="file:///data/images/")
+        assert len(image_resources) == 1
+        assert resource1 in image_resources
+
+        # No matches
+        no_matches = manager.list_resources(prefix="file:///nonexistent/")
+        assert len(no_matches) == 0
+
+    def test_list_templates_with_prefix(self):
+        """Test listing templates with prefix filtering."""
+        manager = ResourceManager()
+
+        # Add templates with different URI patterns
+        def user_func(user_id: str) -> str:
+            return f"User {user_id}"
+
+        def post_func(user_id: str, post_id: str) -> str:
+            return f"User {user_id} Post {post_id}"
+
+        def product_func(product_id: str) -> str:
+            return f"Product {product_id}"
+
+        template1 = manager.add_template(user_func, uri_template="http://api.com/users/{user_id}", name="user_template")
+        template2 = manager.add_template(
+            post_func, uri_template="http://api.com/users/{user_id}/posts/{post_id}", name="post_template"
+        )
+        template3 = manager.add_template(
+            product_func, uri_template="http://api.com/products/{product_id}", name="product_template"
+        )
+
+        # Test listing all templates
+        all_templates = manager.list_templates()
+        assert len(all_templates) == 3
+
+        # Test prefix filtering - matches both user templates
+        user_templates = manager.list_templates(prefix="http://api.com/users/")
+        assert len(user_templates) == 2
+        assert template1 in user_templates
+        assert template2 in user_templates
+
+        # Test partial materialization - only matches post template
+        # The template users/{user_id} generates "users/123" not "users/123/"
+        # But users/{user_id}/posts/{post_id} can generate "users/123/posts/456"
+        user_123_templates = manager.list_templates(prefix="http://api.com/users/123/")
+        assert len(user_123_templates) == 1
+        assert template2 in user_123_templates  # users/{user_id}/posts/{post_id} matches
+
+        # Without trailing slash, both match
+        user_123_no_slash = manager.list_templates(prefix="http://api.com/users/123")
+        assert len(user_123_no_slash) == 2
+        assert template1 in user_123_no_slash
+        assert template2 in user_123_no_slash
+
+        # Test product prefix
+        product_templates = manager.list_templates(prefix="http://api.com/products/")
+        assert len(product_templates) == 1
+        assert template3 in product_templates
+
+        # No matches
+        no_matches = manager.list_templates(prefix="http://api.com/orders/")
+        assert len(no_matches) == 0