Skip to content

Add ToolAnnotations support in FastMCP and lowlevel servers #482

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 8 commits into from
Apr 30, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions src/mcp/server/fastmcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
GetPromptResult,
ImageContent,
TextContent,
ToolAnnotations,
)
from mcp.types import Prompt as MCPPrompt
from mcp.types import PromptArgument as MCPPromptArgument
Expand Down Expand Up @@ -176,6 +177,7 @@ async def list_tools(self) -> list[MCPTool]:
name=info.name,
description=info.description,
inputSchema=info.parameters,
annotations=info.annotations,
)
for info in tools
]
Expand Down Expand Up @@ -244,6 +246,7 @@ def add_tool(
fn: AnyFunction,
name: str | None = None,
description: str | None = None,
annotations: ToolAnnotations | None = None,
) -> None:
"""Add a tool to the server.

Expand All @@ -254,11 +257,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 ToolAnnotations 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: ToolAnnotations | None = None,
) -> Callable[[AnyFunction], AnyFunction]:
"""Decorator to register a tool.

Expand All @@ -269,6 +278,7 @@ def tool(
Args:
name: Optional name for the tool (defaults to function name)
description: Optional description of what the tool does
annotations: Optional ToolAnnotations providing additional tool information

Example:
@server.tool()
Expand All @@ -293,7 +303,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
Expand Down
8 changes: 7 additions & 1 deletion src/mcp/server/fastmcp/tools/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

from mcp.server.fastmcp.exceptions import ToolError
from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata, func_metadata
from mcp.types import ToolAnnotations

if TYPE_CHECKING:
from mcp.server.fastmcp.server import Context
Expand All @@ -30,6 +31,9 @@ class Tool(BaseModel):
context_kwarg: str | None = Field(
None, description="Name of the kwarg that should receive context"
)
annotations: ToolAnnotations | None = Field(
None, description="Optional annotations for the tool"
)

@classmethod
def from_function(
Expand All @@ -38,9 +42,10 @@ def from_function(
name: str | None = None,
description: str | None = None,
context_kwarg: str | None = None,
annotations: ToolAnnotations | None = None,
) -> Tool:
"""Create a Tool from a function."""
from mcp.server.fastmcp import Context
from mcp.server.fastmcp.server import Context
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what do we need to change this? I see the problem, but why do we even need it here?
on top of the file there is


if TYPE_CHECKING:
    from mcp.server.fastmcp.server import Context

Copy link
Contributor Author

@bhosmer-ant bhosmer-ant Apr 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's this a little further down in the function:

    if issubclass(param.annotation, Context):

(and having it up at the top introduces a circular dependency)


func_name = name or fn.__name__

Expand Down Expand Up @@ -73,6 +78,7 @@ def from_function(
fn_metadata=func_arg_metadata,
is_async=is_async,
context_kwarg=context_kwarg,
annotations=annotations,
)

async def run(
Expand Down
6 changes: 5 additions & 1 deletion src/mcp/server/fastmcp/tools/tool_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from mcp.server.fastmcp.tools.base import Tool
from mcp.server.fastmcp.utilities.logging import get_logger
from mcp.shared.context import LifespanContextT
from mcp.types import ToolAnnotations

if TYPE_CHECKING:
from mcp.server.fastmcp.server import Context
Expand Down Expand Up @@ -35,9 +36,12 @@ def add_tool(
fn: Callable[..., Any],
name: str | None = None,
description: str | None = None,
annotations: ToolAnnotations | 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:
Expand Down
50 changes: 50 additions & 0 deletions src/mcp/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand All @@ -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")


Expand Down
41 changes: 41 additions & 0 deletions tests/server/fastmcp/test_tool_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
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:
Expand Down Expand Up @@ -321,3 +322,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 = 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
111 changes: 111 additions & 0 deletions tests/server/test_lowlevel_tool_annotations.py
Original file line number Diff line number Diff line change
@@ -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
Loading