Add ToolAnnotations support in FastMCP and lowlevel servers (#482)

tests/server/fastmcp/test_tool_manager.py

@@ -9,6 +9,7 @@ from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.fastmcp.tools import ToolManager
 from mcp.server.session import ServerSessionT
 from mcp.shared.context import LifespanContextT
+from mcp.types import ToolAnnotations
 
 
 class TestAddTools:
@@ -321,3 +322,43 @@ class TestContextHandling:
         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 = ToolAnnotations(
+            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=ToolAnnotations(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,111 @@
+"""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 (
+    ClientResult,
+    JSONRPCMessage,
+    ServerNotification,
+    ServerRequest,
+    Tool,
+    ToolAnnotations,
+)
+
+
+@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)
+
+    # Message handler for client
+    async def message_handler(
+        message: RequestResponder[ServerRequest, ClientResult]
+        | ServerNotification
+        | Exception,
+    ) -> None:
+        if isinstance(message, Exception):
+            raise message
+
+    # 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