refactor: Replace single prefix filter with multi-path URI filters

Author: Tapan Chugh

src/mcp/.types.py.~undo-tree~

@@ -221,11 +221,13 @@ async def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResul
             types.EmptyResult,
         )
 
-    async def list_resources(self, prefix: str | None = None, cursor: str | None = None) -> types.ListResourcesResult:
+    async def list_resources(
+        self, filters: types.ListFilters | 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.ListRequestParams(prefix=prefix, cursor=cursor)
+        if cursor is not None or filters is not None:
+            params = types.ListRequestParams(filters=filters, cursor=cursor)
         return await self.send_request(
             types.ClientRequest(
                 types.ListResourcesRequest(
@@ -238,13 +240,13 @@ async def list_resources(self, prefix: str | None = None, cursor: str | None = N
 
     async def list_resource_templates(
         self,
-        prefix: str | None = None,
+        filters: types.ListFilters | 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.ListRequestParams(prefix=prefix, cursor=cursor)
+        if cursor is not None or filters is not None:
+            params = types.ListRequestParams(filters=filters, cursor=cursor)
         return await self.send_request(
             types.ClientRequest(
                 types.ListResourceTemplatesRequest(
@@ -342,11 +344,13 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) -
             except SchemaError as e:
                 raise RuntimeError(f"Invalid schema for tool {name}: {e}")
 
-    async def list_prompts(self, prefix: str | None = None, cursor: str | None = None) -> types.ListPromptsResult:
+    async def list_prompts(
+        self, filters: types.ListFilters | None = None, cursor: str | None = None
+    ) -> types.ListPromptsResult:
         """Send a prompts/list request."""
         params = None
-        if cursor is not None or prefix is not None:
-            params = types.ListRequestParams(prefix=prefix, cursor=cursor)
+        if cursor is not None or filters is not None:
+            params = types.ListRequestParams(filters=filters, cursor=cursor)
         return await self.send_request(
             types.ClientRequest(
                 types.ListPromptsRequest(
@@ -394,11 +398,13 @@ async def complete(
             types.CompleteResult,
         )
 
-    async def list_tools(self, prefix: str | None = None, cursor: str | None = None) -> types.ListToolsResult:
+    async def list_tools(
+        self, filters: types.ListFilters | None = None, cursor: str | None = None
+    ) -> types.ListToolsResult:
         """Send a tools/list request."""
         params = None
-        if cursor is not None or prefix is not None:
-            params = types.ListRequestParams(prefix=prefix, cursor=cursor)
+        if cursor is not None or filters is not None:
+            params = types.ListRequestParams(filters=filters, cursor=cursor)
         result = await self.send_request(
             types.ClientRequest(
                 types.ListToolsRequest(

src/mcp/client/session.py

@@ -3,7 +3,7 @@
 from typing import Any
 
 from mcp.server.fastmcp.prompts.base import Message, Prompt
-from mcp.server.fastmcp.uri_utils import filter_by_prefix, normalize_to_prompt_uri
+from mcp.server.fastmcp.uri_utils import filter_by_uri_paths, normalize_to_prompt_uri
 from mcp.server.fastmcp.utilities.logging import get_logger
 
 logger = get_logger(__name__)
@@ -25,11 +25,11 @@ def get_prompt(self, name: str) -> Prompt | None:
         uri = self._normalize_to_uri(name)
         return self._prompts.get(uri)
 
-    def list_prompts(self, prefix: str | None = None) -> list[Prompt]:
-        """List all registered prompts, optionally filtered by URI prefix."""
+    def list_prompts(self, uri_paths: list[str] | None = None) -> list[Prompt]:
+        """List all registered prompts, optionally filtered by URI paths."""
         prompts = list(self._prompts.values())
-        prompts = filter_by_prefix(prompts, prefix, lambda p: p.uri)
-        logger.debug("Listing prompts", extra={"count": len(prompts), "prefix": prefix})
+        prompts = filter_by_uri_paths(prompts, uri_paths, lambda p: p.uri)
+        logger.debug("Listing prompts", extra={"count": len(prompts), "uri_paths": uri_paths})
         return prompts
 
     def add_prompt(

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

@@ -1,7 +1,7 @@
 """Prompt management functionality."""
 
 from mcp.server.fastmcp.prompts.base import Prompt
-from mcp.server.fastmcp.uri_utils import filter_by_prefix, normalize_to_prompt_uri
+from mcp.server.fastmcp.uri_utils import filter_by_uri_paths, normalize_to_prompt_uri
 from mcp.server.fastmcp.utilities.logging import get_logger
 
 logger = get_logger(__name__)
@@ -34,9 +34,9 @@ def get_prompt(self, name: str) -> Prompt | None:
         uri = self._normalize_to_uri(name)
         return self._prompts.get(uri)
 
-    def list_prompts(self, prefix: str | None = None) -> list[Prompt]:
-        """List all registered prompts, optionally filtered by URI prefix."""
+    def list_prompts(self, uri_paths: list[str] | None = None) -> list[Prompt]:
+        """List all registered prompts, optionally filtered by URI paths."""
         prompts = list(self._prompts.values())
-        prompts = filter_by_prefix(prompts, prefix, lambda p: p.uri)
-        logger.debug("Listing prompts", extra={"count": len(prompts), "prefix": prefix})
+        prompts = filter_by_uri_paths(prompts, uri_paths, lambda p: p.uri)
+        logger.debug("Listing prompts", extra={"count": len(prompts), "uri_paths": uri_paths})
         return prompts

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

@@ -7,7 +7,7 @@
 
 from mcp.server.fastmcp.resources.base import Resource
 from mcp.server.fastmcp.resources.templates import ResourceTemplate
-from mcp.server.fastmcp.uri_utils import filter_by_prefix
+from mcp.server.fastmcp.uri_utils import filter_by_uri_paths
 from mcp.server.fastmcp.utilities.logging import get_logger
 
 logger = get_logger(__name__)
@@ -87,20 +87,26 @@ async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
 
         raise ValueError(f"Unknown resource: {uri}")
 
-    def list_resources(self, prefix: str | None = None) -> list[Resource]:
-        """List all registered resources, optionally filtered by URI prefix."""
+    def list_resources(self, uri_paths: list[str] | None = None) -> list[Resource]:
+        """List all registered resources, optionally filtered by URI paths."""
         resources = list(self._resources.values())
-        resources = filter_by_prefix(resources, prefix, lambda r: r.uri)
-        logger.debug("Listing resources", extra={"count": len(resources), "prefix": prefix})
+        resources = filter_by_uri_paths(resources, uri_paths, lambda r: r.uri)
+        logger.debug("Listing resources", extra={"count": len(resources), "uri_paths": uri_paths})
         return resources
 
-    def list_templates(self, prefix: str | None = None) -> list[ResourceTemplate]:
-        """List all registered templates, optionally filtered by URI template prefix."""
+    def list_templates(self, uri_paths: list[str] | None = None) -> list[ResourceTemplate]:
+        """List all registered templates, optionally filtered by URI paths."""
         templates = list(self._templates.values())
-        if prefix:
-            # Ensure prefix ends with / for proper path matching
-            if not prefix.endswith("/"):
-                prefix = prefix + "/"
-            templates = [t for t in templates if t.matches_prefix(prefix)]
-        logger.debug("Listing templates", extra={"count": len(templates), "prefix": prefix})
+        if uri_paths:
+            filtered: list[ResourceTemplate] = []
+            for template in templates:
+                for prefix in uri_paths:
+                    # Ensure prefix ends with / for proper path matching
+                    if not prefix.endswith("/"):
+                        prefix = prefix + "/"
+                    if template.matches_prefix(prefix):
+                        filtered.append(template)
+                        break
+            templates = filtered
+        logger.debug("Listing templates", extra={"count": len(templates), "uri_paths": uri_paths})
         return templates

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

@@ -268,11 +268,11 @@ def _setup_handlers(self) -> None:
         self._mcp_server.list_resource_templates()(self.list_resource_templates)
 
     async def list_tools(self, request: types.ListToolsRequest | None = None) -> list[MCPTool]:
-        """List all available tools, optionally filtered by prefix."""
-        prefix = None
-        if request and request.params:
-            prefix = request.params.prefix
-        tools = self._tool_manager.list_tools(prefix=prefix)
+        """List all available tools, optionally filtered by URI paths."""
+        uri_paths = None
+        if request and request.params and request.params.filters:
+            uri_paths = request.params.filters.uri_paths
+        tools = self._tool_manager.list_tools(uri_paths=uri_paths)
         return [
             MCPTool(
                 name=info.name,
@@ -303,11 +303,11 @@ async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[Cont
         return await self._tool_manager.call_tool(name, arguments, context=context, convert_result=True)
 
     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)
+        """List all available resources, optionally filtered by URI paths."""
+        uri_paths = None
+        if request and request.params and request.params.filters:
+            uri_paths = request.params.filters.uri_paths
+        resources = self._resource_manager.list_resources(uri_paths=uri_paths)
         return [
             MCPResource(
                 uri=resource.uri,
@@ -322,11 +322,11 @@ async def list_resources(self, request: types.ListResourcesRequest | None = None
     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)
+        """List all available resource templates, optionally filtered by URI paths."""
+        uri_paths = None
+        if request and request.params and request.params.filters:
+            uri_paths = request.params.filters.uri_paths
+        templates = self._resource_manager.list_templates(uri_paths=uri_paths)
         return [
             MCPResourceTemplate(
                 uriTemplate=template.uri_template,
@@ -970,11 +970,11 @@ def streamable_http_app(self) -> Starlette:
         )
 
     async def list_prompts(self, request: types.ListPromptsRequest | None = None) -> list[MCPPrompt]:
-        """List all available prompts, optionally filtered by prefix."""
-        prefix = None
-        if request and request.params:
-            prefix = request.params.prefix
-        prompts = self._prompt_manager.list_prompts(prefix=prefix)
+        """List all available prompts, optionally filtered by URI paths."""
+        uri_paths = None
+        if request and request.params and request.params.filters:
+            uri_paths = request.params.filters.uri_paths
+        prompts = self._prompt_manager.list_prompts(uri_paths=uri_paths)
         return [
             MCPPrompt(
                 name=prompt.name,

src/mcp/server/fastmcp/server.py

@@ -5,7 +5,7 @@
 
 from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.fastmcp.tools.base import Tool
-from mcp.server.fastmcp.uri_utils import filter_by_prefix, normalize_to_tool_uri
+from mcp.server.fastmcp.uri_utils import filter_by_uri_paths, normalize_to_tool_uri
 from mcp.server.fastmcp.utilities.logging import get_logger
 from mcp.shared.context import LifespanContextT, RequestT
 from mcp.types import ToolAnnotations
@@ -44,11 +44,11 @@ def get_tool(self, name: str) -> Tool | None:
         uri = self._normalize_to_uri(name)
         return self._tools.get(uri)
 
-    def list_tools(self, prefix: str | None = None) -> list[Tool]:
-        """List all registered tools, optionally filtered by URI prefix."""
+    def list_tools(self, uri_paths: list[str] | None = None) -> list[Tool]:
+        """List all registered tools, optionally filtered by URI paths."""
         tools = list(self._tools.values())
-        tools = filter_by_prefix(tools, prefix, lambda t: t.uri)
-        logger.debug("Listing tools", extra={"count": len(tools), "prefix": prefix})
+        tools = filter_by_uri_paths(tools, uri_paths, lambda t: t.uri)
+        logger.debug("Listing tools", extra={"count": len(tools), "uri_paths": uri_paths})
         return tools
 
     def add_tool(

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

@@ -35,30 +35,35 @@ def normalize_to_prompt_uri(name_or_uri: str) -> str:
     return normalize_to_uri(name_or_uri, PROMPT_SCHEME)
 
 
-def filter_by_prefix(items: list[T], prefix: str | None, uri_getter: Callable[[T], AnyUrl | str]) -> list[T]:
-    """Filter items by URI prefix.
+def filter_by_uri_paths(
+    items: list[T], uri_paths: list[str] | None, uri_getter: Callable[[T], AnyUrl | str]
+) -> list[T]:
+    """Filter items by multiple URI path prefixes.
 
     Args:
         items: List of items to filter
-        prefix: Optional prefix to filter by. If None, returns all items.
+        uri_paths: Optional list of URI path prefixes to filter by. If None or empty, returns all items.
         uri_getter: Function to extract URI from an item
 
     Returns:
-        Filtered list of items
+        Filtered list of items matching any of the provided prefixes
     """
-    if not prefix:
+    if not uri_paths:
         return items
 
-    # Filter items where the URI starts with the prefix
+    # Filter items where the URI matches any of the prefixes
     filtered: list[T] = []
     for item in items:
         uri = str(uri_getter(item))
-        if uri.startswith(prefix):
-            # If prefix ends with a separator, we already have a proper boundary
-            if prefix.endswith(("/", "?", "#")):
-                filtered.append(item)
-            # Otherwise check if it's an exact match or if the next character is a separator
-            elif len(uri) == len(prefix) or uri[len(prefix)] in ("/", "?", "#"):
-                filtered.append(item)
+        for prefix in uri_paths:
+            if uri.startswith(prefix):
+                # If prefix ends with a separator, we already have a proper boundary
+                if prefix.endswith(("/", "?", "#")):
+                    filtered.append(item)
+                    break
+                # Otherwise check if it's an exact match or if the next character is a separator
+                elif len(uri) == len(prefix) or uri[len(prefix)] in ("/", "?", "#"):
+                    filtered.append(item)
+                    break
 
     return filtered

src/mcp/server/fastmcp/uri_utils.py

@@ -60,9 +60,16 @@ class Meta(BaseModel):
     meta: Meta | None = Field(alias="_meta", default=None)
 
 
+class ListFilters(BaseModel):
+    """Filters for list operations."""
+
+    uri_paths: list[str] | None = None
+    """Optional list of absolute URI path prefixes to filter results."""
+
+
 class ListRequestParams(RequestParams):
-    prefix: str | None = None
-    """Optional prefix to filter results by URI."""
+    filters: ListFilters | None = None
+    """Optional filters to apply to the list results."""
 
     cursor: Cursor | None = None
     """

src/mcp/types.py

@@ -104,38 +104,45 @@ def question_age() -> str:
         all_prompts = manager.list_prompts()
         assert len(all_prompts) == 4
 
-        # Test prefix filtering - greeting prompts
-        greeting_prompts = manager.list_prompts(prefix=f"{PROMPT_SCHEME}/greeting/")
+        # Test uri_paths filtering - greeting prompts
+        greeting_prompts = manager.list_prompts(uri_paths=[f"{PROMPT_SCHEME}/greeting/"])
         assert len(greeting_prompts) == 2
         assert all(str(p.uri).startswith(f"{PROMPT_SCHEME}/greeting/") for p in greeting_prompts)
         assert hello_prompt in greeting_prompts
         assert goodbye_prompt in greeting_prompts
 
-        # Test prefix filtering - question prompts
-        question_prompts = manager.list_prompts(prefix=f"{PROMPT_SCHEME}/question/")
+        # Test uri_paths filtering - question prompts
+        question_prompts = manager.list_prompts(uri_paths=[f"{PROMPT_SCHEME}/question/"])
         assert len(question_prompts) == 2
         assert all(str(p.uri).startswith(f"{PROMPT_SCHEME}/question/") for p in question_prompts)
         assert name_prompt in question_prompts
         assert age_prompt in question_prompts
 
         # Test exact URI match
-        hello_prompts = manager.list_prompts(prefix=f"{PROMPT_SCHEME}/greeting/hello")
+        hello_prompts = manager.list_prompts(uri_paths=[f"{PROMPT_SCHEME}/greeting/hello"])
         assert len(hello_prompts) == 1
         assert hello_prompts[0] == hello_prompt
 
         # Test partial prefix doesn't match
-        no_partial = manager.list_prompts(prefix=f"{PROMPT_SCHEME}/greeting/h")
+        no_partial = manager.list_prompts(uri_paths=[f"{PROMPT_SCHEME}/greeting/h"])
         assert len(no_partial) == 0  # Won't match because next char is 'e' not a separator
 
         # Test no matches
-        no_matches = manager.list_prompts(prefix=f"{PROMPT_SCHEME}/nonexistent")
+        no_matches = manager.list_prompts(uri_paths=[f"{PROMPT_SCHEME}/nonexistent"])
         assert len(no_matches) == 0
 
         # Test with trailing slash
-        greeting_prompts_slash = manager.list_prompts(prefix=f"{PROMPT_SCHEME}/greeting/")
+        greeting_prompts_slash = manager.list_prompts(uri_paths=[f"{PROMPT_SCHEME}/greeting/"])
         assert len(greeting_prompts_slash) == 2
         assert greeting_prompts_slash == greeting_prompts
 
+        # Test multiple uri_paths
+        greeting_and_question = manager.list_prompts(
+            uri_paths=[f"{PROMPT_SCHEME}/greeting/", f"{PROMPT_SCHEME}/question/"]
+        )
+        assert len(greeting_and_question) == 4
+        assert all(p in greeting_and_question for p in all_prompts)
+
     @pytest.mark.anyio
     async def test_render_prompt(self):
         """Test rendering a prompt."""