mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-24 14:34:17 +02:00
329 lines
11 KiB
Python
329 lines
11 KiB
Python
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from textwrap import dedent
|
|
|
|
import pytest
|
|
from mcp.types import TextContent, TextResourceContents
|
|
|
|
from fastmcp import Client, FastMCP
|
|
from fastmcp.exceptions import NotFoundError
|
|
from fastmcp.tools import FunctionTool
|
|
from fastmcp.tools.tool import Tool
|
|
|
|
|
|
class TestCreateServer:
|
|
async def test_create_server(self):
|
|
mcp = FastMCP(instructions="Server instructions")
|
|
assert mcp.name.startswith("FastMCP-")
|
|
assert mcp.instructions == "Server instructions"
|
|
|
|
async def test_change_instruction(self):
|
|
mcp = FastMCP(instructions="Server instructions")
|
|
assert mcp.instructions == "Server instructions"
|
|
mcp.instructions = "New instructions"
|
|
assert mcp.instructions == "New 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 result.data == "¡Hola, 世界! 👋"
|
|
|
|
|
|
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._list_tools_mcp()
|
|
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._list_tools_mcp()
|
|
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._call_tool_mcp("adder", {"a": 1, "b": 2})
|
|
|
|
async def test_add_tool_at_init(self):
|
|
def f(x: int) -> int:
|
|
return x + 1
|
|
|
|
def g(x: int) -> int:
|
|
"""add two to a number"""
|
|
return x + 2
|
|
|
|
g_tool = FunctionTool.from_function(g, name="g-tool")
|
|
|
|
mcp = FastMCP(tools=[f, g_tool])
|
|
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 2
|
|
assert tools["f"].name == "f"
|
|
assert tools["g-tool"].name == "g-tool"
|
|
assert tools["g-tool"].description == "add two to a number"
|
|
|
|
|
|
class TestServerDelegation:
|
|
"""Test that FastMCP properly delegates to LocalProvider."""
|
|
|
|
async def test_tool_decorator_delegates_to_local_provider(self):
|
|
"""Test that @mcp.tool registers with the local provider."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def my_tool() -> str:
|
|
return "result"
|
|
|
|
# Verify the tool is in the local provider
|
|
tool = await mcp._local_provider.get_tool("my_tool")
|
|
assert tool is not None
|
|
assert tool.name == "my_tool"
|
|
|
|
async def test_resource_decorator_delegates_to_local_provider(self):
|
|
"""Test that @mcp.resource registers with the local provider."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://test")
|
|
def my_resource() -> str:
|
|
return "content"
|
|
|
|
# Verify the resource is in the local provider
|
|
resource = await mcp._local_provider.get_resource("resource://test")
|
|
assert resource is not None
|
|
|
|
async def test_prompt_decorator_delegates_to_local_provider(self):
|
|
"""Test that @mcp.prompt registers with the local provider."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt
|
|
def my_prompt() -> str:
|
|
return "prompt content"
|
|
|
|
# Verify the prompt is in the local provider
|
|
prompt = await mcp._local_provider.get_prompt("my_prompt")
|
|
assert prompt is not None
|
|
assert prompt.name == "my_prompt"
|
|
|
|
async def test_add_tool_delegates_to_local_provider(self):
|
|
"""Test that mcp.add_tool() registers with the local provider."""
|
|
mcp = FastMCP()
|
|
|
|
def standalone_tool() -> str:
|
|
return "result"
|
|
|
|
mcp.add_tool(FunctionTool.from_function(standalone_tool))
|
|
|
|
# Verify the tool is in the local provider
|
|
tool = await mcp._local_provider.get_tool("standalone_tool")
|
|
assert tool is not None
|
|
assert tool.name == "standalone_tool"
|
|
|
|
async def test_get_tools_includes_local_provider_tools(self):
|
|
"""Test that get_tools() returns tools from local provider."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def local_tool() -> str:
|
|
return "local"
|
|
|
|
tools = await mcp.get_tools()
|
|
assert "local_tool" in tools
|
|
|
|
|
|
class TestResourcePrefixMounting:
|
|
"""Test resource prefixing in mounted servers."""
|
|
|
|
async def test_mounted_server_resource_prefixing(self):
|
|
"""Test that resources in mounted servers use the correct prefix format."""
|
|
# Create a server with resources
|
|
server = FastMCP(name="ResourceServer")
|
|
|
|
@server.resource("resource://test-resource")
|
|
def get_resource():
|
|
return "Resource content"
|
|
|
|
@server.resource("resource:///absolute/path")
|
|
def get_absolute_resource():
|
|
return "Absolute resource content"
|
|
|
|
@server.resource("resource://{param}/template")
|
|
def get_template_resource(param: str):
|
|
return f"Template resource with {param}"
|
|
|
|
# Create a main server and mount the resource server
|
|
main_server = FastMCP(name="MainServer")
|
|
main_server.mount(server, "prefix")
|
|
|
|
# Check that the resources are mounted with the correct prefixes
|
|
resources = await main_server.get_resources()
|
|
templates = await main_server.get_resource_templates()
|
|
|
|
assert "resource://prefix/test-resource" in resources
|
|
assert "resource://prefix//absolute/path" in resources
|
|
assert "resource://prefix/{param}/template" in templates
|
|
|
|
# Test that prefixed resources can be accessed
|
|
async with Client(main_server) as client:
|
|
# Regular resource
|
|
result = await client.read_resource("resource://prefix/test-resource")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Resource content"
|
|
|
|
# Absolute path resource
|
|
result = await client.read_resource("resource://prefix//absolute/path")
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Absolute resource content"
|
|
|
|
# Template resource
|
|
result = await client.read_resource(
|
|
"resource://prefix/param-value/template"
|
|
)
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Template resource with param-value"
|
|
|
|
|
|
class TestSettingsFromEnvironment:
|
|
async def test_settings_from_environment_issue_1749(self):
|
|
"""Test that when auth is enabled, the server starts."""
|
|
from fastmcp.client.transports import PythonStdioTransport
|
|
from fastmcp.server.auth.providers.azure import AzureProvider
|
|
from fastmcp.settings import Settings
|
|
|
|
script = dedent("""
|
|
import os
|
|
|
|
os.environ["FASTMCP_SERVER_AUTH"] = "fastmcp.server.auth.providers.azure.AzureProvider"
|
|
|
|
os.environ["FASTMCP_SERVER_AUTH_AZURE_TENANT_ID"] = "A_Valid_Value"
|
|
os.environ["FASTMCP_SERVER_AUTH_AZURE_CLIENT_ID"] = "A_Valid_Value"
|
|
os.environ["FASTMCP_SERVER_AUTH_AZURE_CLIENT_SECRET"] = "A_Valid_Value"
|
|
os.environ["FASTMCP_SERVER_AUTH_AZURE_REDIRECT_PATH"] = "/auth/callback"
|
|
os.environ["FASTMCP_SERVER_AUTH_AZURE_BASE_URL"] = "http://localhost:8000"
|
|
os.environ["FASTMCP_SERVER_AUTH_AZURE_REQUIRED_SCOPES"] = "User.Read,email,profile"
|
|
os.environ["FASTMCP_SERVER_AUTH_AZURE_JWT_SIGNING_KEY"] = "test-secret"
|
|
|
|
import fastmcp
|
|
|
|
mcp = fastmcp.FastMCP("TestServer")
|
|
|
|
mcp.run()
|
|
""")
|
|
|
|
with TemporaryDirectory() as temp_dir:
|
|
server_file = Path(temp_dir) / "server.py"
|
|
server_file.write_text(script)
|
|
|
|
transport: PythonStdioTransport = PythonStdioTransport(
|
|
script_path=server_file
|
|
)
|
|
|
|
async with Client[PythonStdioTransport](transport=transport) as client:
|
|
tools = await client.list_tools()
|
|
|
|
assert tools == []
|
|
|
|
settings = Settings(
|
|
server_auth="fastmcp.server.auth.providers.azure.AzureProvider"
|
|
)
|
|
|
|
auth_class = settings.server_auth_class
|
|
|
|
assert auth_class is AzureProvider
|
|
|
|
|
|
class TestAbstractCollectionTypes:
|
|
"""Test that FastMCP accepts abstract collection types from collections.abc."""
|
|
|
|
async def test_fastmcp_init_with_tuples(self):
|
|
"""Test FastMCP accepts tuples for sequence parameters."""
|
|
|
|
def dummy_tool() -> str:
|
|
return "test"
|
|
|
|
# Test with tuples and other abstract types
|
|
mcp = FastMCP(
|
|
"test",
|
|
middleware=(), # Empty tuple
|
|
tools=(Tool.from_function(dummy_tool),), # Tuple of tools
|
|
include_tags={"tag1", "tag2"}, # Set
|
|
exclude_tags=frozenset({"tag3"}), # Frozen set
|
|
)
|
|
assert mcp is not None
|
|
assert mcp.name == "test"
|
|
assert isinstance(mcp.middleware, list) # Should be converted to list
|
|
|
|
async def test_fastmcp_init_with_readonly_mapping(self):
|
|
"""Test FastMCP accepts read-only mappings."""
|
|
from types import MappingProxyType
|
|
|
|
# Test with read-only mapping
|
|
mcp = FastMCP(
|
|
"test2",
|
|
tool_transformations=MappingProxyType({}), # Read-only mapping
|
|
)
|
|
assert mcp is not None
|
|
|
|
async def test_fastmcp_works_with_abstract_types(self):
|
|
"""Test that abstract types work end-to-end with a client."""
|
|
|
|
def greet(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
|
|
# Create server with tuple of tools
|
|
mcp = FastMCP("test", tools=(Tool.from_function(greet),))
|
|
|
|
# Verify it works with a client
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("greet", {"name": "World"})
|
|
assert isinstance(result.content[0], TextContent)
|
|
assert result.content[0].text == "Hello, World!"
|