mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
5fa2883670TasksExtension serves io.modelcontextprotocol/tasks on the extension API: a decide-and-task tools/call interceptor (era-gated to modern connections), tasks/get with inlined results and inputRequests, tasks/update delivering poll-based in-task elicitation, tasks/cancel, durable creation, and auth-scoped task isolation. Wire models validate against the vendored ext-tasks schema. Worker-side Context hooks are refcounted so sibling servers cannot strand each other's workers. Co-Authored-By: Claude <noreply@anthropic.com>
259 lines
8.6 KiB
Python
259 lines
8.6 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 fastmcp_tasks import TasksExtension
|
|
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.add_extension(TasksExtension())
|
|
|
|
@mcp.tool(task=True)
|
|
async def background_tool(data: str) -> str:
|
|
"""A tool that runs in background."""
|
|
return f"Processed: {data}"
|
|
|
|
# The rendered tool descriptor auto-populates `execution.task_support` from
|
|
# the tool's task config. (The modern wire drops the SEP-1686 `execution`
|
|
# field, so this is asserted on the server-side render.)
|
|
tool = await mcp.get_tool("background_tool")
|
|
assert tool is not None
|
|
mcp_tool = tool.to_mcp_tool()
|
|
assert isinstance(mcp_tool, MCPTool)
|
|
assert isinstance(mcp_tool.execution, ToolExecution)
|
|
assert mcp_tool.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
|