feat: Use AnyUrl type for URI fields and update managers

- Change Tool and Prompt classes to use AnyUrl for uri field
- Update all manager get/call methods to accept AnyUrl | str
- Add method overloads for better type hints
- Update tests to use AnyUrl objects
- Update filter_by_uri_paths to use Sequence for covariance

Author: Tapan Chugh

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

@@ -5,7 +5,7 @@
 from typing import Any, Literal
 
 import pydantic_core
-from pydantic import BaseModel, Field, TypeAdapter, validate_call
+from pydantic import AnyUrl, BaseModel, Field, TypeAdapter, validate_call
 
 from mcp.types import PROMPT_SCHEME, ContentBlock, TextContent
 
@@ -58,7 +58,7 @@ class Prompt(BaseModel):
     """A prompt template that can be rendered with parameters."""
 
     name: str = Field(description="Name of the prompt")
-    uri: str = Field(description="URI of the prompt")
+    uri: AnyUrl = Field(description="URI of the prompt")
     title: str | None = Field(None, description="Human-readable title of the prompt")
     description: str | None = Field(None, description="Description of what the prompt does")
     arguments: list[PromptArgument] | None = Field(None, description="Arguments that can be passed to the prompt")
@@ -67,7 +67,7 @@ class Prompt(BaseModel):
     def __init__(self, **data: Any) -> None:
         """Initialize Prompt, generating URI from name if not provided."""
         if "uri" not in data and "name" in data:
-            data["uri"] = f"{PROMPT_SCHEME}/{data['name']}"
+            data["uri"] = AnyUrl(f"{PROMPT_SCHEME}/{data['name']}")
         super().__init__(**data)
 
     @classmethod

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

@@ -1,6 +1,8 @@
 """Prompt management functionality."""
 
-from typing import Any
+from typing import Any, overload
+
+from pydantic import AnyUrl
 
 from mcp.server.fastmcp.prompts.base import Message, Prompt
 from mcp.server.fastmcp.uri_utils import filter_by_uri_paths, normalize_to_prompt_uri
@@ -20,15 +22,28 @@ def _normalize_to_uri(self, name_or_uri: str) -> str:
         """Convert name to URI if needed."""
         return normalize_to_prompt_uri(name_or_uri)
 
-    def get_prompt(self, name: str) -> Prompt | None:
+    @overload
+    def get_prompt(self, name_or_uri: str) -> Prompt | None:
+        """Get prompt by name."""
+        ...
+
+    @overload
+    def get_prompt(self, name_or_uri: AnyUrl) -> Prompt | None:
+        """Get prompt by URI."""
+        ...
+
+    def get_prompt(self, name_or_uri: AnyUrl | str) -> Prompt | None:
         """Get prompt by name or URI."""
-        uri = self._normalize_to_uri(name)
+        if isinstance(name_or_uri, AnyUrl):
+            return self._prompts.get(str(name_or_uri))
+        uri = self._normalize_to_uri(name_or_uri)
         return self._prompts.get(uri)
 
-    def list_prompts(self, uri_paths: list[str] | None = None) -> list[Prompt]:
+    def list_prompts(self, uri_paths: list[AnyUrl] | None = None) -> list[Prompt]:
         """List all registered prompts, optionally filtered by URI paths."""
         prompts = list(self._prompts.values())
-        prompts = filter_by_uri_paths(prompts, uri_paths, lambda p: p.uri)
+        if uri_paths:
+            prompts = filter_by_uri_paths(prompts, uri_paths)
         logger.debug("Listing prompts", extra={"count": len(prompts), "uri_paths": uri_paths})
         return prompts
 
@@ -40,19 +55,29 @@ def add_prompt(
         logger.debug(f"Adding prompt: {prompt.name} with URI: {prompt.uri}")
 
         # Check for duplicates
-        existing = self._prompts.get(prompt.uri)
+        existing = self._prompts.get(str(prompt.uri))
         if existing:
             if self.warn_on_duplicate_prompts:
                 logger.warning(f"Prompt already exists: {prompt.uri}")
             return existing
 
-        self._prompts[prompt.uri] = prompt
+        self._prompts[str(prompt.uri)] = prompt
         return prompt
 
-    async def render_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> list[Message]:
+    @overload
+    async def render_prompt(self, name_or_uri: str, arguments: dict[str, Any] | None = None) -> list[Message]:
         """Render a prompt by name with arguments."""
-        prompt = self.get_prompt(name)
+        ...
+
+    @overload
+    async def render_prompt(self, name_or_uri: AnyUrl, arguments: dict[str, Any] | None = None) -> list[Message]:
+        """Render a prompt by URI with arguments."""
+        ...
+
+    async def render_prompt(self, name_or_uri: AnyUrl | str, arguments: dict[str, Any] | None = None) -> list[Message]:
+        """Render a prompt by name or URI with arguments."""
+        prompt = self.get_prompt(name_or_uri)
         if not prompt:
-            raise ValueError(f"Unknown prompt: {name}")
+            raise ValueError(f"Unknown prompt: {name_or_uri}")
 
         return await prompt.render(arguments)

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

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

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

@@ -87,24 +87,26 @@ async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
 
         raise ValueError(f"Unknown resource: {uri}")
 
-    def list_resources(self, uri_paths: list[str] | None = None) -> list[Resource]:
+    def list_resources(self, uri_paths: list[AnyUrl] | None = None) -> list[Resource]:
         """List all registered resources, optionally filtered by URI paths."""
         resources = list(self._resources.values())
-        resources = filter_by_uri_paths(resources, uri_paths, lambda r: r.uri)
+        if uri_paths:
+            resources = filter_by_uri_paths(resources, uri_paths)
         logger.debug("Listing resources", extra={"count": len(resources), "uri_paths": uri_paths})
         return resources
 
-    def list_templates(self, uri_paths: list[str] | None = None) -> list[ResourceTemplate]:
+    def list_templates(self, uri_paths: list[AnyUrl] | None = None) -> list[ResourceTemplate]:
         """List all registered templates, optionally filtered by URI paths."""
         templates = list(self._templates.values())
         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):
+                    prefix_str = str(prefix)
+                    if not prefix_str.endswith("/"):
+                        prefix_str = prefix_str + "/"
+                    if template.matches_prefix(prefix_str):
                         filtered.append(template)
                         break
             templates = filtered

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

@@ -6,7 +6,7 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Any, get_origin
 
-from pydantic import BaseModel, Field
+from pydantic import AnyUrl, BaseModel, Field
 
 from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata, func_metadata
@@ -22,7 +22,7 @@ class Tool(BaseModel):
     """Internal tool registration info."""
 
     name: str = Field(description="Name of the tool")
-    uri: str = Field(description="URI of the tool")
+    uri: AnyUrl = Field(description="URI of the tool")
     title: str | None = Field(None, description="Human-readable title of the tool")
     description: str = Field(description="Description of what the tool does")
     fn: Callable[..., Any] = Field(exclude=True)
@@ -37,7 +37,7 @@ class Tool(BaseModel):
     def __init__(self, **data: Any) -> None:
         """Initialize Tool, generating URI from name if not provided."""
         if "uri" not in data and "name" in data:
-            data["uri"] = f"{TOOL_SCHEME}/{data['name']}"
+            data["uri"] = AnyUrl(f"{TOOL_SCHEME}/{data['name']}")
         super().__init__(**data)
 
     @cached_property

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

@@ -1,7 +1,9 @@
 from __future__ import annotations as _annotations
 
 from collections.abc import Callable
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, overload
+
+from pydantic import AnyUrl
 
 from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.fastmcp.tools.base import Tool
@@ -39,15 +41,28 @@ def _normalize_to_uri(self, name_or_uri: str) -> str:
         """Convert name to URI if needed."""
         return normalize_to_tool_uri(name_or_uri)
 
-    def get_tool(self, name: str) -> Tool | None:
+    @overload
+    def get_tool(self, name_or_uri: str) -> Tool | None:
+        """Get tool by name."""
+        ...
+
+    @overload
+    def get_tool(self, name_or_uri: AnyUrl) -> Tool | None:
+        """Get tool by URI."""
+        ...
+
+    def get_tool(self, name_or_uri: AnyUrl | str) -> Tool | None:
         """Get tool by name or URI."""
-        uri = self._normalize_to_uri(name)
+        if isinstance(name_or_uri, AnyUrl):
+            return self._tools.get(str(name_or_uri))
+        uri = self._normalize_to_uri(name_or_uri)
         return self._tools.get(uri)
 
-    def list_tools(self, uri_paths: list[str] | None = None) -> list[Tool]:
+    def list_tools(self, uri_paths: list[AnyUrl] | None = None) -> list[Tool]:
         """List all registered tools, optionally filtered by URI paths."""
         tools = list(self._tools.values())
-        tools = filter_by_uri_paths(tools, uri_paths, lambda t: t.uri)
+        if uri_paths:
+            tools = filter_by_uri_paths(tools, uri_paths)
         logger.debug("Listing tools", extra={"count": len(tools), "uri_paths": uri_paths})
         return tools
 
@@ -77,16 +92,38 @@ def add_tool(
         self._tools[str(tool.uri)] = tool
         return tool
 
+    @overload
     async def call_tool(
         self,
-        name: str,
+        name_or_uri: str,
         arguments: dict[str, Any],
         context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
         convert_result: bool = False,
     ) -> Any:
         """Call a tool by name with arguments."""
-        tool = self.get_tool(name)
+        ...
+
+    @overload
+    async def call_tool(
+        self,
+        name_or_uri: AnyUrl,
+        arguments: dict[str, Any],
+        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
+        convert_result: bool = False,
+    ) -> Any:
+        """Call a tool by URI with arguments."""
+        ...
+
+    async def call_tool(
+        self,
+        name_or_uri: AnyUrl | str,
+        arguments: dict[str, Any],
+        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
+        convert_result: bool = False,
+    ) -> Any:
+        """Call a tool by name or URI with arguments."""
+        tool = self.get_tool(name_or_uri)
         if not tool:
-            raise ToolError(f"Unknown tool: {name}")
+            raise ToolError(f"Unknown tool: {name_or_uri}")
 
         return await tool.run(arguments, context=context, convert_result=convert_result)

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

@@ -1,13 +1,13 @@
 """Common URI utilities for FastMCP."""
 
-from collections.abc import Callable
-from typing import TypeVar
+from collections.abc import Sequence
+from typing import Protocol, TypeVar, runtime_checkable
 
 from pydantic import AnyUrl
 
 from mcp.types import PROMPT_SCHEME, TOOL_SCHEME
 
-T = TypeVar("T")
+T = TypeVar("T", bound="HasUri")
 
 
 def normalize_to_uri(name_or_uri: str, scheme: str) -> str:
@@ -35,34 +35,37 @@ def normalize_to_prompt_uri(name_or_uri: str) -> str:
     return normalize_to_uri(name_or_uri, PROMPT_SCHEME)
 
 
-def filter_by_uri_paths(
-    items: list[T], uri_paths: list[str] | None, uri_getter: Callable[[T], AnyUrl | str]
-) -> list[T]:
+@runtime_checkable
+class HasUri(Protocol):
+    """Protocol for objects that have a URI attribute."""
+
+    uri: AnyUrl
+
+
+def filter_by_uri_paths(items: Sequence[T], uri_paths: Sequence[AnyUrl]) -> list[T]:
     """Filter items by multiple URI path prefixes.
 
     Args:
-        items: List of items to filter
-        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
+        items: List of items that have a 'uri' attribute
+        uri_paths: List of URI path prefixes to filter by.
 
     Returns:
         Filtered list of items matching any of the provided prefixes
     """
-    if not uri_paths:
-        return items
 
     # Filter items where the URI matches any of the prefixes
     filtered: list[T] = []
     for item in items:
-        uri = str(uri_getter(item))
+        uri = str(item.uri)
         for prefix in uri_paths:
-            if uri.startswith(prefix):
+            prefix_str = str(prefix)
+            if uri.startswith(prefix_str):
                 # If prefix ends with a separator, we already have a proper boundary
-                if prefix.endswith(("/", "?", "#")):
+                if prefix_str.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 ("/", "?", "#"):
+                elif len(uri) == len(prefix_str) or uri[len(prefix_str)] in ("/", "?", "#"):
                     filtered.append(item)
                     break
 

src/mcp/server/fastmcp/uri_utils.py

@@ -63,7 +63,7 @@ class Meta(BaseModel):
 class ListFilters(BaseModel):
     """Filters for list operations."""
 
-    uri_paths: list[str] | None = None
+    uri_paths: list[AnyUrl] | None = None
     """Optional list of absolute URI path prefixes to filter results."""