mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-11 16:19:10 +02:00
756 lines
25 KiB
Python
756 lines
25 KiB
Python
from typing import Annotated
|
|
|
|
import pytest
|
|
from mcp import McpError
|
|
from mcp.types import (
|
|
TextContent,
|
|
TextResourceContents,
|
|
)
|
|
from pydantic import Field
|
|
|
|
from fastmcp import Client, FastMCP
|
|
from fastmcp.exceptions import NotFoundError
|
|
|
|
|
|
class TestCreateServer:
|
|
async def test_create_server(self):
|
|
mcp = FastMCP(instructions="Server instructions")
|
|
assert mcp.name == "FastMCP"
|
|
assert mcp.instructions == "Server instructions"
|
|
|
|
async def test_non_ascii_description(self):
|
|
"""Test that FastMCP handles non-ASCII characters in descriptions correctly"""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(
|
|
description=(
|
|
"🌟 This tool uses emojis and UTF-8 characters: á é í ó ú ñ 漢字 🎉"
|
|
)
|
|
)
|
|
def hello_world(name: str = "世界") -> str:
|
|
return f"¡Hola, {name}! 👋"
|
|
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
assert len(tools) == 1
|
|
tool = tools[0]
|
|
assert tool.description is not None
|
|
assert "🌟" in tool.description
|
|
assert "漢字" in tool.description
|
|
assert "🎉" in tool.description
|
|
|
|
result = await client.call_tool("hello_world", {})
|
|
assert len(result) == 1
|
|
content = result[0]
|
|
assert isinstance(content, TextContent)
|
|
assert "¡Hola, 世界! 👋" == content.text
|
|
|
|
|
|
class TestTools:
|
|
async def test_mcp_tool_name(self):
|
|
"""Test MCPTool name for add_tool (key != tool.name)."""
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool()
|
|
def fn(x: int) -> int:
|
|
return x + 1
|
|
|
|
mcp_tools = await mcp._mcp_list_tools()
|
|
assert len(mcp_tools) == 1
|
|
assert mcp_tools[0].name == "fn"
|
|
|
|
async def test_mcp_tool_custom_name(self):
|
|
"""Test MCPTool name for add_tool (key != tool.name)."""
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(name="custom_name")
|
|
def fn(x: int) -> int:
|
|
return x + 1
|
|
|
|
mcp_tools = await mcp._mcp_list_tools()
|
|
assert len(mcp_tools) == 1
|
|
assert mcp_tools[0].name == "custom_name"
|
|
|
|
async def test_remove_tool_successfully(self):
|
|
"""Test that FastMCP.remove_tool removes the tool from the registry."""
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(name="adder")
|
|
def add(a: int, b: int) -> int:
|
|
return a + b
|
|
|
|
mcp_tools = await mcp.get_tools()
|
|
assert "adder" in mcp_tools
|
|
|
|
mcp.remove_tool("adder")
|
|
mcp_tools = await mcp.get_tools()
|
|
assert "adder" not in mcp_tools
|
|
|
|
with pytest.raises(NotFoundError, match="Unknown tool: adder"):
|
|
await mcp._mcp_call_tool("adder", {"a": 1, "b": 2})
|
|
|
|
|
|
class TestToolDecorator:
|
|
async def test_no_tools_before_decorator(self):
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(NotFoundError, match="Unknown tool: add"):
|
|
await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
|
|
|
|
async def test_tool_decorator(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool()
|
|
def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "3"
|
|
|
|
async def test_tool_decorator_incorrect_usage(self):
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(TypeError, match="The @tool decorator was used incorrectly"):
|
|
|
|
@mcp.tool # Missing parentheses #type: ignore
|
|
def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
async def test_tool_decorator_with_name(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(name="custom-add")
|
|
def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
result = await mcp._mcp_call_tool("custom-add", {"x": 1, "y": 2})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "3"
|
|
|
|
async def test_tool_decorator_with_description(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(description="Add two numbers")
|
|
def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
tools = await mcp._mcp_list_tools()
|
|
assert len(tools) == 1
|
|
tool = tools[0]
|
|
assert tool.description == "Add two numbers"
|
|
|
|
async def test_tool_decorator_instance_method(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
def __init__(self, x: int):
|
|
self.x = x
|
|
|
|
@mcp.tool()
|
|
def add(self, y: int) -> int:
|
|
return self.x + y
|
|
|
|
obj = MyClass(10)
|
|
mcp.add_tool(obj.add)
|
|
result = await mcp._mcp_call_tool("add", {"y": 2})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "12"
|
|
|
|
async def test_tool_decorator_classmethod(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
x: int = 10
|
|
|
|
@classmethod
|
|
def add(cls, y: int) -> int:
|
|
return cls.x + y
|
|
|
|
mcp.add_tool(MyClass.add)
|
|
result = await mcp._mcp_call_tool("add", {"y": 2})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "12"
|
|
|
|
async def test_tool_decorator_staticmethod(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
@staticmethod
|
|
@mcp.tool()
|
|
def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "3"
|
|
|
|
async def test_tool_decorator_async_function(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool()
|
|
async def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "3"
|
|
|
|
async def test_tool_decorator_classmethod_async_function(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
x = 10
|
|
|
|
@classmethod
|
|
async def add(cls, y: int) -> int:
|
|
return cls.x + y
|
|
|
|
mcp.add_tool(MyClass.add)
|
|
result = await mcp._mcp_call_tool("add", {"y": 2})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "12"
|
|
|
|
async def test_tool_decorator_staticmethod_async_function(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
@staticmethod
|
|
async def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
mcp.add_tool(MyClass.add)
|
|
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "3"
|
|
|
|
async def test_tool_decorator_with_tags(self):
|
|
"""Test that the tool decorator properly sets tags."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(tags={"example", "test-tag"})
|
|
def sample_tool(x: int) -> int:
|
|
return x * 2
|
|
|
|
# Verify the tags were set correctly
|
|
tools = mcp._tool_manager.list_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].tags == {"example", "test-tag"}
|
|
|
|
async def test_add_tool_with_custom_name(self):
|
|
"""Test adding a tool with a custom name using server.add_tool()."""
|
|
mcp = FastMCP()
|
|
|
|
def multiply(a: int, b: int) -> int:
|
|
"""Multiply two numbers."""
|
|
return a * b
|
|
|
|
mcp.add_tool(multiply, name="custom_multiply")
|
|
|
|
# Check that the tool is registered with the custom name
|
|
tools = await mcp.get_tools()
|
|
assert "custom_multiply" in tools
|
|
|
|
# Call the tool by its custom name
|
|
result = await mcp._mcp_call_tool("custom_multiply", {"a": 5, "b": 3})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "15"
|
|
|
|
# Original name should not be registered
|
|
assert "multiply" not in tools
|
|
|
|
async def test_tool_with_annotated_arguments(self):
|
|
"""Test that tools with annotated arguments work correctly."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool()
|
|
def add(
|
|
x: Annotated[int, Field(description="x is an int")],
|
|
y: Annotated[str, Field(description="y is not an int")],
|
|
) -> None:
|
|
pass
|
|
|
|
tool = (await mcp.get_tools())["add"]
|
|
assert tool.parameters["properties"]["x"]["description"] == "x is an int"
|
|
assert tool.parameters["properties"]["y"]["description"] == "y is not an int"
|
|
|
|
async def test_tool_with_field_defaults(self):
|
|
"""Test that tools with annotated arguments work correctly."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool()
|
|
def add(
|
|
x: int = Field(description="x is an int"),
|
|
y: str = Field(description="y is not an int"),
|
|
) -> None:
|
|
pass
|
|
|
|
tool = (await mcp.get_tools())["add"]
|
|
assert tool.parameters["properties"]["x"]["description"] == "x is an int"
|
|
assert tool.parameters["properties"]["y"]["description"] == "y is not an int"
|
|
|
|
|
|
class TestResourceDecorator:
|
|
async def test_no_resources_before_decorator(self):
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(McpError, match="Unknown resource"):
|
|
async with Client(mcp) as client:
|
|
await client.read_resource("resource://data")
|
|
|
|
async def test_resource_decorator(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://data")
|
|
def get_data() -> str:
|
|
return "Hello, world!"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Hello, world!"
|
|
|
|
async def test_resource_decorator_incorrect_usage(self):
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(
|
|
TypeError, match="The @resource decorator was used incorrectly"
|
|
):
|
|
|
|
@mcp.resource # Missing parentheses #type: ignore
|
|
def get_data() -> str:
|
|
return "Hello, world!"
|
|
|
|
async def test_resource_decorator_with_name(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://data", name="custom-data")
|
|
def get_data() -> str:
|
|
return "Hello, world!"
|
|
|
|
resources_dict = await mcp.get_resources()
|
|
resources = list(resources_dict.values())
|
|
assert len(resources) == 1
|
|
assert resources[0].name == "custom-data"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Hello, world!"
|
|
|
|
async def test_resource_decorator_with_description(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://data", description="Data resource")
|
|
def get_data() -> str:
|
|
return "Hello, world!"
|
|
|
|
resources_dict = await mcp.get_resources()
|
|
resources = list(resources_dict.values())
|
|
assert len(resources) == 1
|
|
assert resources[0].description == "Data resource"
|
|
|
|
async def test_resource_decorator_with_tags(self):
|
|
"""Test that the resource decorator properly sets tags."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://data", tags={"example", "test-tag"})
|
|
def get_data() -> str:
|
|
return "Hello, world!"
|
|
|
|
resources_dict = await mcp.get_resources()
|
|
resources = list(resources_dict.values())
|
|
assert len(resources) == 1
|
|
assert resources[0].tags == {"example", "test-tag"}
|
|
|
|
async def test_resource_decorator_instance_method(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
def __init__(self, prefix: str):
|
|
self.prefix = prefix
|
|
|
|
def get_data(self) -> str:
|
|
return f"{self.prefix} Hello, world!"
|
|
|
|
obj = MyClass("My prefix:")
|
|
mcp.add_resource_fn(
|
|
obj.get_data, uri="resource://data", name="instance-resource"
|
|
)
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "My prefix: Hello, world!"
|
|
|
|
async def test_resource_decorator_classmethod(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
prefix = "Class prefix:"
|
|
|
|
@classmethod
|
|
def get_data(cls) -> str:
|
|
return f"{cls.prefix} Hello, world!"
|
|
|
|
mcp.add_resource_fn(
|
|
MyClass.get_data, uri="resource://data", name="class-resource"
|
|
)
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Class prefix: Hello, world!"
|
|
|
|
async def test_resource_decorator_staticmethod(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
@staticmethod
|
|
@mcp.resource("resource://data")
|
|
def get_data() -> str:
|
|
return "Static Hello, world!"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Static Hello, world!"
|
|
|
|
async def test_resource_decorator_async_function(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://data")
|
|
async def get_data() -> str:
|
|
return "Async Hello, world!"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Async Hello, world!"
|
|
|
|
|
|
class TestTemplateDecorator:
|
|
async def test_template_decorator(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{name}/data")
|
|
def get_data(name: str) -> str:
|
|
return f"Data for {name}"
|
|
|
|
templates_dict = await mcp.get_resource_templates()
|
|
templates = list(templates_dict.values())
|
|
assert len(templates) == 1
|
|
assert templates[0].name == "get_data"
|
|
assert templates[0].uri_template == "resource://{name}/data"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://test/data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Data for test"
|
|
|
|
async def test_template_decorator_incorrect_usage(self):
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(
|
|
TypeError, match="The @resource decorator was used incorrectly"
|
|
):
|
|
|
|
@mcp.resource # Missing parentheses #type: ignore
|
|
def get_data(name: str) -> str:
|
|
return f"Data for {name}"
|
|
|
|
async def test_template_decorator_with_name(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{name}/data", name="custom-template")
|
|
def get_data(name: str) -> str:
|
|
return f"Data for {name}"
|
|
|
|
templates_dict = await mcp.get_resource_templates()
|
|
templates = list(templates_dict.values())
|
|
assert len(templates) == 1
|
|
assert templates[0].name == "custom-template"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://test/data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Data for test"
|
|
|
|
async def test_template_decorator_with_description(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{name}/data", description="Template description")
|
|
def get_data(name: str) -> str:
|
|
return f"Data for {name}"
|
|
|
|
templates_dict = await mcp.get_resource_templates()
|
|
templates = list(templates_dict.values())
|
|
assert len(templates) == 1
|
|
assert templates[0].description == "Template description"
|
|
|
|
async def test_template_decorator_instance_method(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
def __init__(self, prefix: str):
|
|
self.prefix = prefix
|
|
|
|
def get_data(self, name: str) -> str:
|
|
return f"{self.prefix} Data for {name}"
|
|
|
|
obj = MyClass("My prefix:")
|
|
mcp.add_resource_fn(
|
|
obj.get_data, uri="resource://{name}/data", name="instance-template"
|
|
)
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://test/data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "My prefix: Data for test"
|
|
|
|
async def test_template_decorator_classmethod(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
prefix = "Class prefix:"
|
|
|
|
@classmethod
|
|
def get_data(cls, name: str) -> str:
|
|
return f"{cls.prefix} Data for {name}"
|
|
|
|
mcp.add_resource_fn(
|
|
MyClass.get_data,
|
|
uri="resource://{name}/data",
|
|
name="class-template",
|
|
)
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://test/data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Class prefix: Data for test"
|
|
|
|
async def test_template_decorator_staticmethod(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
@staticmethod
|
|
@mcp.resource("resource://{name}/data")
|
|
def get_data(name: str) -> str:
|
|
return f"Static Data for {name}"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://test/data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Static Data for test"
|
|
|
|
async def test_template_decorator_async_function(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{name}/data")
|
|
async def get_data(name: str) -> str:
|
|
return f"Async Data for {name}"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("resource://test/data")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Async Data for test"
|
|
|
|
async def test_template_decorator_with_tags(self):
|
|
"""Test that the template decorator properly sets tags."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{param}", tags={"template", "test-tag"})
|
|
def template_resource(param: str) -> str:
|
|
return f"Template resource: {param}"
|
|
|
|
templates_dict = await mcp.get_resource_templates()
|
|
template = templates_dict["resource://{param}"]
|
|
assert template.tags == {"template", "test-tag"}
|
|
|
|
async def test_template_decorator_wildcard_param(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{param*}")
|
|
def template_resource(param: str) -> str:
|
|
return f"Template resource: {param}"
|
|
|
|
templates_dict = await mcp.get_resource_templates()
|
|
template = templates_dict["resource://{param*}"]
|
|
assert template.uri_template == "resource://{param*}"
|
|
assert template.name == "template_resource"
|
|
|
|
|
|
class TestPromptDecorator:
|
|
async def test_prompt_decorator(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt()
|
|
def fn() -> str:
|
|
return "Hello, world!"
|
|
|
|
prompts_dict = await mcp.get_prompts()
|
|
assert len(prompts_dict) == 1
|
|
prompt = prompts_dict["fn"]
|
|
assert prompt.name == "fn"
|
|
# Don't compare functions directly since validate_call wraps them
|
|
content = await prompt.render()
|
|
assert isinstance(content[0].content, TextContent)
|
|
assert content[0].content.text == "Hello, world!"
|
|
|
|
async def test_prompt_decorator_incorrect_usage(self):
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(
|
|
TypeError, match="The @prompt decorator was used incorrectly"
|
|
):
|
|
|
|
@mcp.prompt # Missing parentheses #type: ignore
|
|
def fn() -> str:
|
|
return "Hello, world!"
|
|
|
|
async def test_prompt_decorator_with_name(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(name="custom_name")
|
|
def fn() -> str:
|
|
return "Hello, world!"
|
|
|
|
prompts_dict = await mcp.get_prompts()
|
|
assert len(prompts_dict) == 1
|
|
prompt = prompts_dict["custom_name"]
|
|
assert prompt.name == "custom_name"
|
|
content = await prompt.render()
|
|
assert isinstance(content[0].content, TextContent)
|
|
assert content[0].content.text == "Hello, world!"
|
|
|
|
async def test_prompt_decorator_with_description(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(description="A custom description")
|
|
def fn() -> str:
|
|
return "Hello, world!"
|
|
|
|
prompts_dict = await mcp.get_prompts()
|
|
assert len(prompts_dict) == 1
|
|
prompt = prompts_dict["fn"]
|
|
assert prompt.description == "A custom description"
|
|
content = await prompt.render()
|
|
assert isinstance(content[0].content, TextContent)
|
|
assert content[0].content.text == "Hello, world!"
|
|
|
|
async def test_prompt_decorator_with_parameters(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt()
|
|
def test_prompt(name: str, greeting: str = "Hello") -> str:
|
|
return f"{greeting}, {name}!"
|
|
|
|
prompts_dict = await mcp.get_prompts()
|
|
assert len(prompts_dict) == 1
|
|
prompt = prompts_dict["test_prompt"]
|
|
assert prompt.arguments is not None
|
|
assert len(prompt.arguments) == 2
|
|
assert prompt.arguments[0].name == "name"
|
|
assert prompt.arguments[0].required is True
|
|
assert prompt.arguments[1].name == "greeting"
|
|
assert prompt.arguments[1].required is False
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt("test_prompt", {"name": "World"})
|
|
assert len(result.messages) == 1
|
|
message = result.messages[0]
|
|
assert isinstance(message.content, TextContent)
|
|
assert message.content.text == "Hello, World!"
|
|
|
|
result = await client.get_prompt(
|
|
"test_prompt", {"name": "World", "greeting": "Hi"}
|
|
)
|
|
assert len(result.messages) == 1
|
|
message = result.messages[0]
|
|
assert isinstance(message.content, TextContent)
|
|
assert message.content.text == "Hi, World!"
|
|
|
|
async def test_prompt_decorator_instance_method(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
def __init__(self, prefix: str):
|
|
self.prefix = prefix
|
|
|
|
def test_prompt(self) -> str:
|
|
return f"{self.prefix} Hello, world!"
|
|
|
|
obj = MyClass("My prefix:")
|
|
mcp.add_prompt(obj.test_prompt, name="test_prompt")
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt("test_prompt")
|
|
assert len(result.messages) == 1
|
|
message = result.messages[0]
|
|
assert isinstance(message.content, TextContent)
|
|
assert message.content.text == "My prefix: Hello, world!"
|
|
|
|
async def test_prompt_decorator_classmethod(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
prefix = "Class prefix:"
|
|
|
|
@classmethod
|
|
def test_prompt(cls) -> str:
|
|
return f"{cls.prefix} Hello, world!"
|
|
|
|
mcp.add_prompt(MyClass.test_prompt, name="test_prompt")
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt("test_prompt")
|
|
assert len(result.messages) == 1
|
|
message = result.messages[0]
|
|
assert isinstance(message.content, TextContent)
|
|
assert message.content.text == "Class prefix: Hello, world!"
|
|
|
|
async def test_prompt_decorator_staticmethod(self):
|
|
mcp = FastMCP()
|
|
|
|
class MyClass:
|
|
@staticmethod
|
|
@mcp.prompt()
|
|
def test_prompt() -> str:
|
|
return "Static Hello, world!"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt("test_prompt")
|
|
assert len(result.messages) == 1
|
|
message = result.messages[0]
|
|
assert isinstance(message.content, TextContent)
|
|
assert message.content.text == "Static Hello, world!"
|
|
|
|
async def test_prompt_decorator_async_function(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt()
|
|
async def test_prompt() -> str:
|
|
return "Async Hello, world!"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt("test_prompt")
|
|
assert len(result.messages) == 1
|
|
message = result.messages[0]
|
|
assert isinstance(message.content, TextContent)
|
|
assert message.content.text == "Async Hello, world!"
|
|
|
|
async def test_prompt_decorator_with_tags(self):
|
|
"""Test that the prompt decorator properly sets tags."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(tags={"example", "test-tag"})
|
|
def sample_prompt() -> str:
|
|
return "Hello, world!"
|
|
|
|
prompts_dict = await mcp.get_prompts()
|
|
assert len(prompts_dict) == 1
|
|
prompt = prompts_dict["sample_prompt"]
|
|
assert prompt.tags == {"example", "test-tag"}
|