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 any(t.name == "adder" for t in mcp_tools) mcp.remove_tool("adder") mcp_tools = await mcp.get_tools() assert not any(t.name == "adder" for t 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 f_tool = next(t for t in tools if t.name == "f") g_tool = next(t for t in tools if t.name == "g-tool") assert f_tool.name == "f" assert g_tool.name == "g-tool" assert 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 any(t.name == "local_tool" for t 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 any(str(r.uri) == "resource://prefix/test-resource" for r in resources) assert any(str(r.uri) == "resource://prefix//absolute/path" for r in resources) assert any( t.uri_template == "resource://prefix/{param}/template" for t 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!"