fastmcp/tests/server/test_server.py

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!"