wehub-resource-sync
60e0ffc959
Schema Crash Test / Real-world schema crash test (232K schemas) (push) Waiting to run
Run static analysis / static_analysis (push) Waiting to run
Tests / Tests: Python 3.10 on ubuntu-latest (push) Waiting to run
Tests / Tests: Python 3.13 on ubuntu-latest (push) Waiting to run
Tests / Tests: Python 3.10 on windows-latest (push) Waiting to run
Tests / Tests with lowest-direct dependencies (push) Waiting to run
Tests / Package install smoke (push) Waiting to run
Upgrade checks / Static analysis (push) Waiting to run
Upgrade checks / Tests: Python 3.10 on ubuntu-latest (push) Waiting to run
Upgrade checks / Tests: Python 3.13 on ubuntu-latest (push) Waiting to run
Upgrade checks / Tests: Python 3.10 on windows-latest (push) Waiting to run
Upgrade checks / Integration tests (push) Waiting to run
Upgrade checks / Notify on failure (push) Blocked by required conditions
Upgrade checks / Close issue on success (push) Blocked by required conditions
Update MCPServerConfig Schema / update-config-schema (push) Waiting to run
Update SDK Documentation / update-sdk-docs (push) Waiting to run
Tests / MCP conformance tests (push) Waiting to run
Tests / Integration tests (push) Waiting to run
256 lines
8.4 KiB
Python
256 lines
8.4 KiB
Python
from typing import Any
|
|
|
|
from mcp_types import Tool as MCPTool
|
|
from mcp_types import ToolAnnotations, ToolExecution
|
|
|
|
from fastmcp import Client, FastMCP
|
|
from fastmcp.tools.base import Tool
|
|
from tests.conftest import make_server_request_context
|
|
|
|
|
|
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",
|
|
read_only_hint=True,
|
|
open_world_hint=False,
|
|
)
|
|
)
|
|
def echo(message: str) -> str:
|
|
"""Echo back the message provided."""
|
|
return message
|
|
|
|
# Check internal tool objects directly
|
|
tools = await mcp.list_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].annotations is not None
|
|
assert tools[0].annotations.title == "Echo Tool"
|
|
assert tools[0].annotations.read_only_hint is True
|
|
assert tools[0].annotations.open_world_hint 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",
|
|
read_only_hint=True,
|
|
open_world_hint=False,
|
|
)
|
|
)
|
|
def echo(message: str) -> str:
|
|
"""Echo back the message provided."""
|
|
return message
|
|
|
|
# Check via MCP protocol
|
|
result = await mcp._on_list_tools(make_server_request_context(), None)
|
|
assert len(result.tools) == 1
|
|
assert result.tools[0].annotations is not None
|
|
assert result.tools[0].annotations.title == "Echo Tool"
|
|
assert result.tools[0].annotations.read_only_hint is True
|
|
assert result.tools[0].annotations.open_world_hint 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",
|
|
read_only_hint=True,
|
|
open_world_hint=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.read_only_hint is True
|
|
assert tools_result[0].annotations.open_world_hint 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.read_only_hint is True
|
|
assert tools_result[0].annotations.open_world_hint 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",
|
|
read_only_hint=False,
|
|
destructive_hint=True,
|
|
idempotent_hint=False,
|
|
open_world_hint=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 = await mcp.list_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].annotations is not None
|
|
assert tools[0].annotations.title == "Direct Tool"
|
|
assert tools[0].annotations.read_only_hint is False
|
|
assert tools[0].annotations.destructive_hint is True
|
|
assert tools[0].annotations.idempotent_hint is False
|
|
assert tools[0].annotations.open_world_hint 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",
|
|
read_only_hint=False,
|
|
destructive_hint=True,
|
|
idempotent_hint=False,
|
|
open_world_hint=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.read_only_hint is False
|
|
assert tools_result[0].annotations.destructive_hint 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",
|
|
read_only_hint=False,
|
|
destructive_hint=False,
|
|
),
|
|
)
|
|
|
|
mcp.add_tool(tool)
|
|
|
|
# Check internal tool objects directly
|
|
tools = await mcp.list_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].annotations is not None
|
|
assert tools[0].annotations.title == "Create Item"
|
|
assert tools[0].annotations.read_only_hint is False
|
|
assert tools[0].annotations.destructive_hint 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",
|
|
read_only_hint=False,
|
|
destructive_hint=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}
|
|
|
|
|
|
async def test_task_execution_auto_populated_for_task_enabled_tool():
|
|
"""Test that execution.task_support is automatically set when tool has task=True."""
|
|
mcp = FastMCP("Test Server")
|
|
|
|
@mcp.tool(task=True)
|
|
async def background_tool(data: str) -> str:
|
|
"""A tool that runs in background."""
|
|
return f"Processed: {data}"
|
|
|
|
async with Client(mcp) as client:
|
|
tools_result = await client.list_tools()
|
|
assert len(tools_result) == 1
|
|
assert tools_result[0].name == "background_tool"
|
|
assert isinstance(tools_result[0], MCPTool)
|
|
assert isinstance(tools_result[0].execution, ToolExecution)
|
|
assert tools_result[0].execution.task_support == "optional"
|
|
|
|
|
|
async def test_task_execution_omitted_for_task_disabled_tool():
|
|
"""Test that execution is not set when tool has task=False or default."""
|
|
mcp = FastMCP("Test Server")
|
|
|
|
@mcp.tool(task=False)
|
|
def sync_tool(data: str) -> str:
|
|
"""A synchronous tool."""
|
|
return f"Processed: {data}"
|
|
|
|
async with Client(mcp) as client:
|
|
tools_result = await client.list_tools()
|
|
assert len(tools_result) == 1
|
|
assert tools_result[0].name == "sync_tool"
|
|
# execution should be None for non-task tools (default is False, omitted)
|
|
assert tools_result[0].execution is None
|