feat: add resource annotations support to FastMCP (modelcontextprotocol#1468)

Author: fennb

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

@@ -13,7 +13,7 @@
     field_validator,
 )
 
-from mcp.types import Icon
+from mcp.types import Annotations, Icon
 
 
 class Resource(BaseModel, abc.ABC):
@@ -31,6 +31,7 @@ class Resource(BaseModel, abc.ABC):
         pattern=r"^[a-zA-Z0-9]+/[a-zA-Z0-9\-+.]+$",
     )
     icons: list[Icon] | None = Field(default=None, description="Optional list of icons for this resource")
+    annotations: Annotations | None = Field(default=None, description="Optional annotations for the resource")
 
     @field_validator("name", mode="before")
     @classmethod

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

@@ -10,7 +10,7 @@
 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.types import Icon
+from mcp.types import Annotations, Icon
 
 if TYPE_CHECKING:
     from mcp.server.fastmcp.server import Context
@@ -63,6 +63,7 @@ def add_template(
         description: str | None = None,
         mime_type: str | None = None,
         icons: list[Icon] | None = None,
+        annotations: Annotations | None = None,
     ) -> ResourceTemplate:
         """Add a template from a function."""
         template = ResourceTemplate.from_function(
@@ -73,6 +74,7 @@ def add_template(
             description=description,
             mime_type=mime_type,
             icons=icons,
+            annotations=annotations,
         )
         self._templates[template.uri_template] = template
         return template

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

@@ -12,7 +12,7 @@
 from mcp.server.fastmcp.resources.types import FunctionResource, Resource
 from mcp.server.fastmcp.utilities.context_injection import find_context_parameter, inject_context
 from mcp.server.fastmcp.utilities.func_metadata import func_metadata
-from mcp.types import Icon
+from mcp.types import Annotations, Icon
 
 if TYPE_CHECKING:
     from mcp.server.fastmcp.server import Context
@@ -29,6 +29,7 @@ class ResourceTemplate(BaseModel):
     description: str | None = Field(description="Description of what the resource does")
     mime_type: str = Field(default="text/plain", description="MIME type of the resource content")
     icons: list[Icon] | None = Field(default=None, description="Optional list of icons for the resource template")
+    annotations: Annotations | None = Field(default=None, description="Optional annotations for the resource template")
     fn: Callable[..., Any] = Field(exclude=True)
     parameters: dict[str, Any] = Field(description="JSON schema for function parameters")
     context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context")
@@ -43,6 +44,7 @@ def from_function(
         description: str | None = None,
         mime_type: str | None = None,
         icons: list[Icon] | None = None,
+        annotations: Annotations | None = None,
         context_kwarg: str | None = None,
     ) -> ResourceTemplate:
         """Create a template from a function."""
@@ -71,6 +73,7 @@ def from_function(
             description=description or fn.__doc__ or "",
             mime_type=mime_type or "text/plain",
             icons=icons,
+            annotations=annotations,
             fn=fn,
             parameters=parameters,
             context_kwarg=context_kwarg,
@@ -108,6 +111,7 @@ async def create_resource(
                 description=self.description,
                 mime_type=self.mime_type,
                 icons=self.icons,
+                annotations=self.annotations,
                 fn=lambda: result,  # Capture result in closure
             )
         except Exception as e:

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

@@ -14,7 +14,7 @@
 from pydantic import AnyUrl, Field, ValidationInfo, validate_call
 
 from mcp.server.fastmcp.resources.base import Resource
-from mcp.types import Icon
+from mcp.types import Annotations, Icon
 
 
 class TextResource(Resource):
@@ -82,6 +82,7 @@ def from_function(
         description: str | None = None,
         mime_type: str | None = None,
         icons: list[Icon] | None = None,
+        annotations: Annotations | None = None,
     ) -> "FunctionResource":
         """Create a FunctionResource from a function."""
         func_name = name or fn.__name__
@@ -99,6 +100,7 @@ def from_function(
             mime_type=mime_type or "text/plain",
             fn=fn,
             icons=icons,
+            annotations=annotations,
         )
 
 

src/mcp/server/fastmcp/server.py

@@ -43,7 +43,7 @@
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.context import LifespanContextT, RequestContext, RequestT
-from mcp.types import AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations
+from mcp.types import Annotations, AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations
 from mcp.types import Prompt as MCPPrompt
 from mcp.types import PromptArgument as MCPPromptArgument
 from mcp.types import Resource as MCPResource
@@ -322,6 +322,7 @@ async def list_resources(self) -> list[MCPResource]:
                 description=resource.description,
                 mimeType=resource.mime_type,
                 icons=resource.icons,
+                annotations=resource.annotations,
             )
             for resource in resources
         ]
@@ -336,6 +337,7 @@ async def list_resource_templates(self) -> list[MCPResourceTemplate]:
                 description=template.description,
                 mimeType=template.mime_type,
                 icons=template.icons,
+                annotations=template.annotations,
             )
             for template in templates
         ]
@@ -497,6 +499,7 @@ def resource(
         description: str | None = None,
         mime_type: str | None = None,
         icons: list[Icon] | None = None,
+        annotations: Annotations | None = None,
     ) -> Callable[[AnyFunction], AnyFunction]:
         """Decorator to register a function as a resource.
 
@@ -572,6 +575,7 @@ def decorator(fn: AnyFunction) -> AnyFunction:
                     description=description,
                     mime_type=mime_type,
                     icons=icons,
+                    annotations=annotations,
                 )
             else:
                 # Register as regular resource
@@ -583,6 +587,7 @@ def decorator(fn: AnyFunction) -> AnyFunction:
                     description=description,
                     mime_type=mime_type,
                     icons=icons,
+                    annotations=annotations,
                 )
                 self.add_resource(resource)
             return fn

tests/server/fastmcp/resources/test_resource_template.py

@@ -4,7 +4,9 @@
 import pytest
 from pydantic import BaseModel
 
+from mcp.server.fastmcp import FastMCP
 from mcp.server.fastmcp.resources import FunctionResource, ResourceTemplate
+from mcp.types import Annotations
 
 
 class TestResourceTemplate:
@@ -186,3 +188,73 @@ def get_data(value: str) -> CustomData:
         assert isinstance(resource, FunctionResource)
         content = await resource.read()
         assert content == '"hello"'
+
+
+class TestResourceTemplateAnnotations:
+    """Test annotations on resource templates."""
+
+    def test_template_with_annotations(self):
+        """Test creating a template with annotations."""
+
+        def get_user_data(user_id: str) -> str:
+            return f"User {user_id}"
+
+        annotations = Annotations(priority=0.9)
+
+        template = ResourceTemplate.from_function(
+            fn=get_user_data, uri_template="resource://users/{user_id}", annotations=annotations
+        )
+
+        assert template.annotations is not None
+        assert template.annotations.priority == 0.9
+
+    def test_template_without_annotations(self):
+        """Test that annotations are optional for templates."""
+
+        def get_user_data(user_id: str) -> str:
+            return f"User {user_id}"
+
+        template = ResourceTemplate.from_function(fn=get_user_data, uri_template="resource://users/{user_id}")
+
+        assert template.annotations is None
+
+    @pytest.mark.anyio
+    async def test_template_annotations_in_fastmcp(self):
+        """Test template annotations via FastMCP decorator."""
+
+        mcp = FastMCP()
+
+        @mcp.resource("resource://dynamic/{id}", annotations=Annotations(audience=["user"], priority=0.7))
+        def get_dynamic(id: str) -> str:
+            """A dynamic annotated resource."""
+            return f"Data for {id}"
+
+        templates = await mcp.list_resource_templates()
+        assert len(templates) == 1
+        assert templates[0].annotations is not None
+        assert templates[0].annotations.audience == ["user"]
+        assert templates[0].annotations.priority == 0.7
+
+    @pytest.mark.anyio
+    async def test_template_created_resources_inherit_annotations(self):
+        """Test that resources created from templates inherit annotations."""
+
+        def get_item(item_id: str) -> str:
+            return f"Item {item_id}"
+
+        annotations = Annotations(priority=0.6)
+
+        template = ResourceTemplate.from_function(
+            fn=get_item, uri_template="resource://items/{item_id}", annotations=annotations
+        )
+
+        # Create a resource from the template
+        resource = await template.create_resource("resource://items/123", {"item_id": "123"})
+
+        # The resource should inherit the template's annotations
+        assert resource.annotations is not None
+        assert resource.annotations.priority == 0.6
+
+        # Verify the resource works correctly
+        content = await resource.read()
+        assert content == "Item 123"

tests/server/fastmcp/resources/test_resources.py

@@ -1,7 +1,9 @@
 import pytest
 from pydantic import AnyUrl
 
+from mcp.server.fastmcp import FastMCP
 from mcp.server.fastmcp.resources import FunctionResource, Resource
+from mcp.types import Annotations
 
 
 class TestResourceValidation:
@@ -99,3 +101,95 @@ class ConcreteResource(Resource):
 
         with pytest.raises(TypeError, match="abstract method"):
             ConcreteResource(uri=AnyUrl("test://test"), name="test")  # type: ignore
+
+
+class TestResourceAnnotations:
+    """Test annotations on resources."""
+
+    def test_resource_with_annotations(self):
+        """Test creating a resource with annotations."""
+
+        def get_data() -> str:
+            return "data"
+
+        annotations = Annotations(audience=["user"], priority=0.8)
+
+        resource = FunctionResource.from_function(fn=get_data, uri="resource://test", annotations=annotations)
+
+        assert resource.annotations is not None
+        assert resource.annotations.audience == ["user"]
+        assert resource.annotations.priority == 0.8
+
+    def test_resource_without_annotations(self):
+        """Test that annotations are optional."""
+
+        def get_data() -> str:
+            return "data"
+
+        resource = FunctionResource.from_function(fn=get_data, uri="resource://test")
+
+        assert resource.annotations is None
+
+    @pytest.mark.anyio
+    async def test_resource_annotations_in_fastmcp(self):
+        """Test resource annotations via FastMCP decorator."""
+
+        mcp = FastMCP()
+
+        @mcp.resource("resource://annotated", annotations=Annotations(audience=["assistant"], priority=0.5))
+        def get_annotated() -> str:
+            """An annotated resource."""
+            return "annotated data"
+
+        resources = await mcp.list_resources()
+        assert len(resources) == 1
+        assert resources[0].annotations is not None
+        assert resources[0].annotations.audience == ["assistant"]
+        assert resources[0].annotations.priority == 0.5
+
+    @pytest.mark.anyio
+    async def test_resource_annotations_with_both_audiences(self):
+        """Test resource with both user and assistant audience."""
+
+        mcp = FastMCP()
+
+        @mcp.resource("resource://both", annotations=Annotations(audience=["user", "assistant"], priority=1.0))
+        def get_both() -> str:
+            return "for everyone"
+
+        resources = await mcp.list_resources()
+        assert resources[0].annotations is not None
+        assert resources[0].annotations.audience == ["user", "assistant"]
+        assert resources[0].annotations.priority == 1.0
+
+
+class TestAnnotationsValidation:
+    """Test validation of annotation values."""
+
+    def test_priority_validation(self):
+        """Test that priority is validated to be between 0.0 and 1.0."""
+
+        # Valid priorities
+        Annotations(priority=0.0)
+        Annotations(priority=0.5)
+        Annotations(priority=1.0)
+
+        # Invalid priorities should raise validation error
+        with pytest.raises(Exception):  # Pydantic validation error
+            Annotations(priority=-0.1)
+
+        with pytest.raises(Exception):
+            Annotations(priority=1.1)
+
+    def test_audience_validation(self):
+        """Test that audience only accepts valid roles."""
+
+        # Valid audiences
+        Annotations(audience=["user"])
+        Annotations(audience=["assistant"])
+        Annotations(audience=["user", "assistant"])
+        Annotations(audience=[])
+
+        # Invalid roles should raise validation error
+        with pytest.raises(Exception):  # Pydantic validation error
+            Annotations(audience=["invalid_role"])  # type: ignore