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

tests/server/fastmcp/resources/test_resource_template.py

@@ -4,7 +4,9 @@ from typing import Any
 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 @@ class TestResourceTemplate:
         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 TestResourceValidation:
 
         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