feat: Convert tools and prompts to URI-based storage

- Add uri field to FastMCP Tool and Prompt classes with automatic generation
- Change ToolManager and PromptManager to use URIs as primary keys
- Implement backward compatibility with automatic name-to-URI conversion
- Update server to use stored URIs instead of generating them
- Fix tests to handle URI-based warning messages

This aligns tools and prompts with the resource storage pattern, where URIs
are the primary identifiers. Names are automatically converted to URIs
(e.g., 'my_tool' -> 'tool://my_tool') for backward compatibility.

Author: Tapan Chugh

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

@@ -1,14 +1,12 @@
 """Base classes for FastMCP prompts."""
 
 import inspect
-import json
 from collections.abc import Awaitable, Callable, Sequence
 from typing import Any, Literal
 
 import pydantic_core
 from pydantic import BaseModel, Field, TypeAdapter, validate_call
 
-from mcp.server.fastmcp.resources import PromptResource
 from mcp.types import ContentBlock, TextContent
 
 
@@ -56,12 +54,22 @@ class PromptArgument(BaseModel):
     required: bool = Field(default=False, description="Whether the argument is required")
 
 
-class Prompt(PromptResource):
+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")
+    title: str | None = Field(None, description="Human-readable title of the prompt")
+    description: str = Field(description="Description of what the prompt does")
     arguments: list[PromptArgument] | None = Field(None, description="Arguments that can be passed to the prompt")
     fn: Callable[..., PromptResult | Awaitable[PromptResult]] = Field(exclude=True)
 
+    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://{data['name']}"
+        super().__init__(**data)
+
     @classmethod
     def from_function(
         cls,
@@ -110,28 +118,6 @@ def from_function(
             fn=fn,
         )
 
-    async def read(self) -> str | bytes:
-        """Read the prompt template/documentation as JSON."""
-        prompt_info: dict[str, Any] = {
-            "name": self.name,
-            "title": self.title,
-            "description": self.description,
-            "uri": str(self.uri),
-        }
-
-        # Include arguments if available
-        if self.arguments:
-            prompt_info["arguments"] = [
-                {
-                    "name": arg.name,
-                    "description": arg.description,
-                    "required": arg.required,
-                }
-                for arg in self.arguments
-            ]
-
-        return json.dumps(prompt_info, indent=2)
-
     async def render(self, arguments: dict[str, Any] | None = None) -> list[Message]:
         """Render the prompt with arguments."""
         # Validate required arguments

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

@@ -13,20 +13,27 @@ def __init__(self, warn_on_duplicate_prompts: bool = True):
         self._prompts: dict[str, Prompt] = {}
         self.warn_on_duplicate_prompts = warn_on_duplicate_prompts
 
+    def _normalize_to_uri(self, name_or_uri: str) -> str:
+        """Convert name to URI if needed."""
+        if name_or_uri.startswith("prompt://"):
+            return name_or_uri
+        return f"prompt://{name_or_uri}"
+
     def add_prompt(self, prompt: Prompt) -> Prompt:
         """Add a prompt to the manager."""
-        logger.debug(f"Adding prompt: {prompt.name}")
-        existing = self._prompts.get(prompt.name)
+        logger.debug(f"Adding prompt: {prompt.name} with URI: {prompt.uri}")
+        existing = self._prompts.get(prompt.uri)
         if existing:
             if self.warn_on_duplicate_prompts:
-                logger.warning(f"Prompt already exists: {prompt.name}")
+                logger.warning(f"Prompt already exists: {prompt.uri}")
             return existing
-        self._prompts[prompt.name] = prompt
+        self._prompts[prompt.uri] = prompt
         return prompt
 
     def get_prompt(self, name: str) -> Prompt | None:
-        """Get prompt by name."""
-        return self._prompts.get(name)
+        """Get prompt by name or URI."""
+        uri = self._normalize_to_uri(name)
+        return self._prompts.get(uri)
 
     def list_prompts(self) -> list[Prompt]:
         """List all registered prompts."""

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

@@ -1,8 +1,6 @@
 from .base import Resource
-from .prompt_resource import PromptResource
 from .resource_manager import ResourceManager
 from .templates import ResourceTemplate
-from .tool_resource import ToolResource
 from .types import (
     BinaryResource,
     DirectoryResource,
@@ -22,6 +20,4 @@
     "DirectoryResource",
     "ResourceTemplate",
     "ResourceManager",
-    "ToolResource",
-    "PromptResource",
 ]

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

@@ -87,26 +87,11 @@ 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.
-
-        Note: Tool and prompt resources (with tool:// and prompt:// URIs) are excluded
-        by default to maintain API compatibility.
-        """
-        all_resources = list(self._resources.values())
-
-        # Filter out tool and prompt resources to maintain API compatibility
-        resources = [
-            r for r in all_resources if not (str(r.uri).startswith("tool://") or str(r.uri).startswith("prompt://"))
-        ]
-
-        # Apply prefix filter if provided
+        """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={"total_count": len(all_resources), "filtered_count": len(resources), "prefix": prefix},
-        )
+        logger.debug("Listing resources", extra={"count": len(resources), "prefix": prefix})
         return resources
 
     def list_templates(self, prefix: str | None = None) -> list[ResourceTemplate]:

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

@@ -275,6 +275,7 @@ async def list_tools(self) -> list[MCPTool]:
                 name=info.name,
                 title=info.title,
                 description=info.description,
+                uri=info.uri,
                 inputSchema=info.parameters,
                 outputSchema=info.output_schema,
                 annotations=info.annotations,
@@ -973,6 +974,7 @@ async def list_prompts(self) -> list[MCPPrompt]:
                 name=prompt.name,
                 title=prompt.title,
                 description=prompt.description,
+                uri=prompt.uri,
                 arguments=[
                     MCPPromptArgument(
                         name=arg.name,

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

@@ -2,15 +2,13 @@
 
 import functools
 import inspect
-import json
 from collections.abc import Callable
 from functools import cached_property
 from typing import TYPE_CHECKING, Any, get_origin
 
-from pydantic import Field
+from pydantic import BaseModel, Field
 
 from mcp.server.fastmcp.exceptions import ToolError
-from mcp.server.fastmcp.resources import ToolResource
 from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata, func_metadata
 from mcp.types import ToolAnnotations
 
@@ -20,9 +18,13 @@
     from mcp.shared.context import LifespanContextT, RequestT
 
 
-class Tool(ToolResource):
+class Tool(BaseModel):
     """Internal tool registration info."""
 
+    name: str = Field(description="Name of the tool")
+    uri: str = 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)
     parameters: dict[str, Any] = Field(description="JSON schema for tool parameters")
     fn_metadata: FuncMetadata = Field(
@@ -32,6 +34,12 @@ class Tool(ToolResource):
     context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context")
     annotations: ToolAnnotations | None = Field(None, description="Optional annotations for the tool")
 
+    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://{data['name']}"
+        super().__init__(**data)
+
     @cached_property
     def output_schema(self) -> dict[str, Any] | None:
         return self.fn_metadata.output_schema
@@ -86,26 +94,6 @@ def from_function(
             annotations=annotations,
         )
 
-    async def read(self) -> str | bytes:
-        """Read the tool schema/documentation as JSON."""
-        tool_info = {
-            "name": self.name,
-            "title": self.title,
-            "description": self.description,
-            "uri": str(self.uri),
-            "parameters": self.parameters,
-        }
-
-        # Include output schema if available
-        if self.output_schema:
-            tool_info["output_schema"] = self.output_schema
-
-        # Include annotations if available
-        if self.annotations:
-            tool_info["annotations"] = self.annotations.model_dump(exclude_none=True)
-
-        return json.dumps(tool_info, indent=2)
-
     async def run(
         self,
         arguments: dict[str, Any],

src/mcp/server/fastmcp/server.py

@@ -28,15 +28,22 @@ def __init__(
         self._tools: dict[str, Tool] = {}
         if tools is not None:
             for tool in tools:
-                if warn_on_duplicate_tools and tool.name in self._tools:
-                    logger.warning(f"Tool already exists: {tool.name}")
-                self._tools[tool.name] = tool
+                if warn_on_duplicate_tools and tool.uri in self._tools:
+                    logger.warning(f"Tool already exists: {tool.uri}")
+                self._tools[tool.uri] = tool
 
         self.warn_on_duplicate_tools = warn_on_duplicate_tools
 
+    def _normalize_to_uri(self, name_or_uri: str) -> str:
+        """Convert name to URI if needed."""
+        if name_or_uri.startswith("tool://"):
+            return name_or_uri
+        return f"tool://{name_or_uri}"
+
     def get_tool(self, name: str) -> Tool | None:
-        """Get tool by name."""
-        return self._tools.get(name)
+        """Get tool by name or URI."""
+        uri = self._normalize_to_uri(name)
+        return self._tools.get(uri)
 
     def list_tools(self) -> list[Tool]:
         """List all registered tools."""
@@ -60,12 +67,12 @@ def add_tool(
             annotations=annotations,
             structured_output=structured_output,
         )
-        existing = self._tools.get(tool.name)
+        existing = self._tools.get(tool.uri)
         if existing:
             if self.warn_on_duplicate_tools:
-                logger.warning(f"Tool already exists: {tool.name}")
+                logger.warning(f"Tool already exists: {tool.uri}")
             return existing
-        self._tools[tool.name] = tool
+        self._tools[tool.uri] = tool
         return tool
 
     async def call_tool(

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

@@ -1,7 +1,7 @@
 from collections.abc import Callable
 from typing import Annotated, Any, Generic, Literal, TypeAlias, TypeVar
 
-from pydantic import BaseModel, ConfigDict, Field, FileUrl, RootModel
+from pydantic import BaseModel, ConfigDict, Field, FileUrl, RootModel, model_validator
 from pydantic.networks import AnyUrl, UrlConstraints
 from typing_extensions import deprecated
 
@@ -445,6 +445,14 @@ class Resource(BaseMetadata):
     """
     model_config = ConfigDict(extra="allow")
 
+    @model_validator(mode="after")
+    def validate_uri_scheme(self) -> "Resource":
+        """Ensure resource URI doesn't use reserved schemes."""
+        uri_str = str(self.uri)
+        if uri_str.startswith(("tool://", "prompt://")):
+            raise ValueError(f"Resource URI cannot use reserved schemes 'tool://' or 'prompt://', got: {self.uri}")
+        return self
+
 
 class ResourceTemplate(BaseMetadata):
     """A template description for resources available on the server."""
@@ -642,6 +650,8 @@ class PromptArgument(BaseModel):
 class Prompt(BaseMetadata):
     """A prompt or prompt template that the server offers."""
 
+    uri: Annotated[AnyUrl, UrlConstraints(allowed_schemes=["prompt"], host_required=False)]
+    """URI for the prompt. Must use 'prompt' scheme."""
     description: str | None = None
     """An optional description of what this prompt provides."""
     arguments: list[PromptArgument] | None = None
@@ -861,6 +871,8 @@ class ToolAnnotations(BaseModel):
 class Tool(BaseMetadata):
     """Definition for a tool the client can call."""
 
+    uri: Annotated[AnyUrl, UrlConstraints(allowed_schemes=["tool"], host_required=False)]
+    """URI for the tool. Must use 'tool' scheme."""
     description: str | None = None
     """A human-readable description of the tool."""
     inputSchema: dict[str, Any]