Spaces:
Running
Running
File size: 7,233 Bytes
c2a9892 d4fceae c2a9892 a853a78 c2a9892 506c09a c2a9892 506c09a c2a9892 a853a78 c2a9892 a853a78 c2a9892 506c09a c2a9892 a853a78 c2a9892 a853a78 c2a9892 aa46c09 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 | from typing import Any
from mcp.types import ToolAnnotations
from fastmcp import Client, FastMCP
from fastmcp.tools.tool import Tool
async def test_tool_annotations_in_tool_manager():
"""Test that tool annotations are correctly stored in the tool manager."""
mcp = FastMCP("Test Server")
@mcp.tool(
annotations=ToolAnnotations(
title="Echo Tool",
readOnlyHint=True,
openWorldHint=False,
)
)
def echo(message: str) -> str:
"""Echo back the message provided."""
return message
# Check internal tool objects directly
tools_dict = await mcp._tool_manager.get_tools()
tools = list(tools_dict.values())
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
assert tools[0].annotations.openWorldHint is False
async def test_tool_annotations_in_mcp_protocol():
"""Test that tool annotations are correctly propagated to MCP tools list."""
mcp = FastMCP("Test Server")
@mcp.tool(
annotations=ToolAnnotations(
title="Echo Tool",
readOnlyHint=True,
openWorldHint=False,
)
)
def echo(message: str) -> str:
"""Echo back the message provided."""
return message
# Check via MCP protocol
mcp_tools = await mcp._mcp_list_tools()
assert len(mcp_tools) == 1
assert mcp_tools[0].annotations is not None
assert mcp_tools[0].annotations.title == "Echo Tool"
assert mcp_tools[0].annotations.readOnlyHint is True
assert mcp_tools[0].annotations.openWorldHint is False
async def test_tool_annotations_in_client_api():
"""Test that tool annotations are correctly accessible via client API."""
mcp = FastMCP("Test Server")
@mcp.tool(
annotations=ToolAnnotations(
title="Echo Tool",
readOnlyHint=True,
openWorldHint=False,
)
)
def echo(message: str) -> str:
"""Echo back the message provided."""
return message
# Check via client API
async with Client(mcp) as client:
tools_result = await client.list_tools()
assert len(tools_result) == 1
assert tools_result[0].name == "echo"
assert tools_result[0].annotations is not None
assert tools_result[0].annotations.title == "Echo Tool"
assert tools_result[0].annotations.readOnlyHint is True
assert tools_result[0].annotations.openWorldHint is False
async def test_provide_tool_annotations_as_dict_to_decorator():
"""Test that tool annotations are correctly accessible via client API."""
mcp = FastMCP("Test Server")
@mcp.tool(
annotations={
"title": "Echo Tool",
"readOnlyHint": True,
"openWorldHint": False,
}
)
def echo(message: str) -> str:
"""Echo back the message provided."""
return message
# Check via client API
async with Client(mcp) as client:
tools_result = await client.list_tools()
assert len(tools_result) == 1
assert tools_result[0].name == "echo"
assert tools_result[0].annotations is not None
assert tools_result[0].annotations.title == "Echo Tool"
assert tools_result[0].annotations.readOnlyHint is True
assert tools_result[0].annotations.openWorldHint is False
async def test_direct_tool_annotations_in_tool_manager():
"""Test direct ToolAnnotations object is correctly stored in tool manager."""
mcp = FastMCP("Test Server")
annotations = ToolAnnotations(
title="Direct Tool",
readOnlyHint=False,
destructiveHint=True,
idempotentHint=False,
openWorldHint=True,
)
@mcp.tool(annotations=annotations)
def modify(data: dict[str, Any]) -> dict[str, Any]:
"""Modify the data provided."""
return {"modified": True, **data}
# Check internal tool objects directly
tools_dict = await mcp._tool_manager.get_tools()
tools = list(tools_dict.values())
assert len(tools) == 1
assert tools[0].annotations is not None
assert tools[0].annotations.title == "Direct Tool"
assert tools[0].annotations.readOnlyHint is False
assert tools[0].annotations.destructiveHint is True
assert tools[0].annotations.idempotentHint is False
assert tools[0].annotations.openWorldHint is True
async def test_direct_tool_annotations_in_client_api():
"""Test direct ToolAnnotations object is correctly accessible via client API."""
mcp = FastMCP("Test Server")
annotations = ToolAnnotations(
title="Direct Tool",
readOnlyHint=False,
destructiveHint=True,
idempotentHint=False,
openWorldHint=True,
)
@mcp.tool(annotations=annotations)
def modify(data: dict[str, Any]) -> dict[str, Any]:
"""Modify the data provided."""
return {"modified": True, **data}
# Check via client API
async with Client(mcp) as client:
tools_result = await client.list_tools()
assert len(tools_result) == 1
assert tools_result[0].name == "modify"
assert tools_result[0].annotations is not None
assert tools_result[0].annotations.title == "Direct Tool"
assert tools_result[0].annotations.readOnlyHint is False
assert tools_result[0].annotations.destructiveHint is True
async def test_add_tool_method_annotations():
"""Test that tool annotations work with add_tool method."""
mcp = FastMCP("Test Server")
def create_item(name: str, value: int) -> dict[str, Any]:
"""Create a new item."""
return {"name": name, "value": value}
tool = Tool.from_function(
create_item,
name="create_item",
annotations=ToolAnnotations(
title="Create Item",
readOnlyHint=False,
destructiveHint=False,
),
)
mcp.add_tool(tool)
# Check internal tool objects directly
tools_dict = await mcp._tool_manager.get_tools()
tools = list(tools_dict.values())
assert len(tools) == 1
assert tools[0].annotations is not None
assert tools[0].annotations.title == "Create Item"
assert tools[0].annotations.readOnlyHint is False
assert tools[0].annotations.destructiveHint is False
async def test_tool_functionality_with_annotations():
"""Test that tool functionality is preserved when using annotations."""
mcp = FastMCP("Test Server")
def create_item(name: str, value: int) -> dict[str, Any]:
"""Create a new item."""
return {"name": name, "value": value}
tool = Tool.from_function(
create_item,
name="create_item",
annotations=ToolAnnotations(
title="Create Item",
readOnlyHint=False,
destructiveHint=False,
),
)
mcp.add_tool(tool)
# Use the tool to verify functionality is preserved
async with Client(mcp) as client:
result = await client.call_tool(
"create_item", {"name": "test_item", "value": 42}
)
assert result.data == {"name": "test_item", "value": 42}
|