Add support and tests for ToolAnnotations in FastMCP and lowlevel servers

Author: bhosmer-ant

src/mcp/server/fastmcp/server.py

@@ -172,12 +172,17 @@ def _setup_handlers(self) -> None:
 
     async def list_tools(self) -> list[MCPTool]:
         """List all available tools."""
+        from mcp.types import ToolAnnotations
+
         tools = self._tool_manager.list_tools()
         return [
             MCPTool(
                 name=info.name,
                 description=info.description,
                 inputSchema=info.parameters,
+                annotations=ToolAnnotations.model_validate(info.annotations)
+                if info.annotations
+                else None,
             )
             for info in tools
         ]
@@ -246,6 +251,7 @@ def add_tool(
         fn: AnyFunction,
         name: str | None = None,
         description: str | None = None,
+        annotations: dict[str, Any] | None = None,
     ) -> None:
         """Add a tool to the server.
 
@@ -256,11 +262,17 @@ def add_tool(
             fn: The function to register as a tool
             name: Optional name for the tool (defaults to function name)
             description: Optional description of what the tool does
+            annotations: Optional annotations providing additional tool information
         """
-        self._tool_manager.add_tool(fn, name=name, description=description)
+        self._tool_manager.add_tool(
+            fn, name=name, description=description, annotations=annotations
+        )
 
     def tool(
-        self, name: str | None = None, description: str | None = None
+        self,
+        name: str | None = None,
+        description: str | None = None,
+        annotations: dict[str, Any] | None = None,
     ) -> Callable[[AnyFunction], AnyFunction]:
         """Decorator to register a tool.
 
@@ -271,6 +283,7 @@ def tool(
         Args:
             name: Optional name for the tool (defaults to function name)
             description: Optional description of what the tool does
+            annotations: Optional annotations providing additional tool information
 
         Example:
             @server.tool()
@@ -295,7 +308,9 @@ async def async_tool(x: int, context: Context) -> str:
             )
 
         def decorator(fn: AnyFunction) -> AnyFunction:
-            self.add_tool(fn, name=name, description=description)
+            self.add_tool(
+                fn, name=name, description=description, annotations=annotations
+            )
             return fn
 
         return decorator

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

@@ -30,6 +30,9 @@ class Tool(BaseModel):
     context_kwarg: str | None = Field(
         None, description="Name of the kwarg that should receive context"
     )
+    annotations: dict[str, Any] | None = Field(
+        None, description="Optional annotations for the tool"
+    )
 
     @classmethod
     def from_function(
@@ -38,6 +41,7 @@ def from_function(
         name: str | None = None,
         description: str | None = None,
         context_kwarg: str | None = None,
+        annotations: dict[str, Any] | None = None,
     ) -> Tool:
         """Create a Tool from a function."""
         from mcp.server.fastmcp import Context
@@ -73,6 +77,7 @@ def from_function(
             fn_metadata=func_arg_metadata,
             is_async=is_async,
             context_kwarg=context_kwarg,
+            annotations=annotations,
         )
 
     async def run(

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

@@ -35,9 +35,12 @@ def add_tool(
         fn: Callable[..., Any],
         name: str | None = None,
         description: str | None = None,
+        annotations: dict[str, Any] | None = None,
     ) -> Tool:
         """Add a tool to the server."""
-        tool = Tool.from_function(fn, name=name, description=description)
+        tool = Tool.from_function(
+            fn, name=name, description=description, annotations=annotations
+        )
         existing = self._tools.get(tool.name)
         if existing:
             if self.warn_on_duplicate_tools:

src/mcp/types.py

@@ -705,6 +705,54 @@ class ListToolsRequest(PaginatedRequest[RequestParams | None, Literal["tools/lis
     params: RequestParams | None = None
 
 
+class ToolAnnotations(BaseModel):
+    """
+    Additional properties describing a Tool to clients.
+
+    NOTE: all properties in ToolAnnotations are **hints**.
+    They are not guaranteed to provide a faithful description of
+    tool behavior (including descriptive properties like `title`).
+
+    Clients should never make tool use decisions based on ToolAnnotations
+    received from untrusted servers.
+    """
+
+    title: str | None = None
+    """A human-readable title for the tool."""
+
+    readOnlyHint: bool | None = None
+    """
+    If true, the tool does not modify its environment.
+    Default: false
+    """
+
+    destructiveHint: bool | None = None
+    """
+    If true, the tool may perform destructive updates to its environment.
+    If false, the tool performs only additive updates.
+    (This property is meaningful only when `readOnlyHint == false`)
+    Default: true
+    """
+
+    idempotentHint: bool | None = None
+    """
+    If true, calling the tool repeatedly with the same arguments 
+    will have no additional effect on the its environment.
+    (This property is meaningful only when `readOnlyHint == false`)
+    Default: false
+    """
+
+    openWorldHint: bool | None = None
+    """
+    If true, this tool may interact with an "open world" of external
+    entities. If false, the tool's domain of interaction is closed.
+    For example, the world of a web search tool is open, whereas that
+    of a memory tool is not.
+    Default: true
+    """
+    model_config = ConfigDict(extra="allow")
+
+
 class Tool(BaseModel):
     """Definition for a tool the client can call."""
 
@@ -714,6 +762,8 @@ class Tool(BaseModel):
     """A human-readable description of the tool."""
     inputSchema: dict[str, Any]
     """A JSON Schema object defining the expected parameters for the tool."""
+    annotations: ToolAnnotations | None = None
+    """Optional additional tool information."""
     model_config = ConfigDict(extra="allow")
 
 

tests/server/fastmcp/test_tool_manager.py

@@ -321,3 +321,43 @@ def tool_with_context(x: int, ctx: Context) -> str:
         ctx = mcp.get_context()
         with pytest.raises(ToolError, match="Error executing tool tool_with_context"):
             await manager.call_tool("tool_with_context", {"x": 42}, context=ctx)
+
+
+class TestToolAnnotations:
+    def test_tool_annotations(self):
+        """Test that tool annotations are correctly added to tools."""
+
+        def read_data(path: str) -> str:
+            """Read data from a file."""
+            return f"Data from {path}"
+
+        annotations = {
+            "title": "File Reader",
+            "readOnlyHint": True,
+            "openWorldHint": False,
+        }
+
+        manager = ToolManager()
+        tool = manager.add_tool(read_data, annotations=annotations)
+
+        assert tool.annotations is not None
+        assert tool.annotations["title"] == "File Reader"
+        assert tool.annotations["readOnlyHint"] is True
+        assert tool.annotations["openWorldHint"] is False
+
+    @pytest.mark.anyio
+    async def test_tool_annotations_in_fastmcp(self):
+        """Test that tool annotations are included in MCPTool conversion."""
+
+        app = FastMCP()
+
+        @app.tool(annotations={"title": "Echo Tool", "readOnlyHint": True})
+        def echo(message: str) -> str:
+            """Echo a message back."""
+            return message
+
+        tools = await app.list_tools()
+        assert len(tools) == 1
+        assert tools[0].annotations is not None
+        assert tools[0].annotations.title == "Echo Tool"
+        assert tools[0].annotations.readOnlyHint is True

tests/server/test_lowlevel_tool_annotations.py

@@ -0,0 +1,125 @@
+"""Tests for tool annotations in low-level server."""
+
+import anyio
+import pytest
+
+from mcp.client.session import ClientSession
+from mcp.server import Server
+from mcp.server.lowlevel import NotificationOptions
+from mcp.server.models import InitializationOptions
+from mcp.server.session import ServerSession
+from mcp.shared.session import RequestResponder
+from mcp.types import (
+    Tool,
+    ToolAnnotations,
+    JSONRPCMessage,
+    ListToolsRequest,
+    ListToolsResult,
+    InitializeRequestParams,
+    Implementation,
+    ClientCapabilities,
+    ServerRequest,
+    ClientResult,
+    ServerNotification,
+)
+
+
+@pytest.mark.anyio
+async def test_lowlevel_server_tool_annotations():
+    """Test that tool annotations work in low-level server."""
+    server = Server("test")
+
+    # Create a tool with annotations
+    @server.list_tools()
+    async def list_tools():
+        return [
+            Tool(
+                name="echo",
+                description="Echo a message back",
+                inputSchema={
+                    "type": "object",
+                    "properties": {
+                        "message": {"type": "string"},
+                    },
+                    "required": ["message"],
+                },
+                annotations=ToolAnnotations(
+                    title="Echo Tool", 
+                    readOnlyHint=True,
+                ),
+            )
+        ]
+
+    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream[
+        JSONRPCMessage
+    ](10)
+    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream[
+        JSONRPCMessage
+    ](10)
+
+    # Track results for assertion
+    tools_result = None
+
+    # Message handler for client
+    async def message_handler(
+        message: RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception,
+    ) -> None:
+        nonlocal tools_result
+        if isinstance(message, Exception):
+            raise message
+        if isinstance(message, RequestResponder):
+            result = message.message.result
+            if isinstance(result, dict) and "tools" in result:
+                tools_result = ListToolsResult.model_validate(result)
+
+    # Server task
+    async def run_server():
+        async with ServerSession(
+            client_to_server_receive,
+            server_to_client_send,
+            InitializationOptions(
+                server_name="test-server",
+                server_version="1.0.0",
+                capabilities=server.get_capabilities(
+                    notification_options=NotificationOptions(),
+                    experimental_capabilities={},
+                ),
+            ),
+        ) as server_session:
+            async with anyio.create_task_group() as tg:
+
+                async def handle_messages():
+                    async for message in server_session.incoming_messages:
+                        await server._handle_message(
+                            message, server_session, {}, False
+                        )
+
+                tg.start_soon(handle_messages)
+                await anyio.sleep_forever()
+
+    # Run the test
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(run_server)
+        
+        async with ClientSession(
+            server_to_client_receive,
+            client_to_server_send,
+            message_handler=message_handler,
+        ) as client_session:
+            # Initialize the session
+            await client_session.initialize()
+            
+            # List tools
+            tools_result = await client_session.list_tools()
+            
+            
+            # Cancel the server task
+            tg.cancel_scope.cancel()
+    
+    # Verify results
+    assert tools_result is not None
+    assert len(tools_result.tools) == 1
+    assert tools_result.tools[0].name == "echo"
+    assert tools_result.tools[0].annotations is not None
+    assert tools_result.tools[0].annotations.title == "Echo Tool"
+    assert tools_result.tools[0].annotations.readOnlyHint is True