tests/server/test_mount.py

import json
import sys
from contextlib import asynccontextmanager

import pytest

from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.client.transports import FastMCPTransport, SSETransport
from fastmcp.server.proxy import FastMCPProxy


class TestBasicMount:
    """Test basic mounting functionality."""

    async def test_mount_simple_server(self):
        """Test mounting a simple server and accessing its tool."""
        # Create main app and sub-app
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        # Add a tool to the sub-app
        @sub_app.tool
        def sub_tool() -> str:
            return "This is from the sub app"

        # Mount the sub-app to the main app
        main_app.mount(sub_app, "sub")

        # Get tools from main app, should include sub_app's tools
        tools = await main_app.get_tools()
        assert "sub_sub_tool" in tools

        async with Client(main_app) as client:
            result = await client.call_tool("sub_sub_tool", {})
            assert result[0].text == "This is from the sub app"  # type: ignore[attr-defined]

    async def test_mount_with_custom_separator(self):
        """Test mounting with a custom tool separator (deprecated but still supported)."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.tool
        def greet(name: str) -> str:
            return f"Hello, {name}!"

        # Mount without custom separator - custom separators are deprecated
        main_app.mount(sub_app, "sub")

        # Tool should be accessible with the default separator
        tools = await main_app.get_tools()
        assert "sub_greet" in tools

        # Call the tool
        result = await main_app._mcp_call_tool("sub_greet", {"name": "World"})
        assert result[0].text == "Hello, World!"  # type: ignore[attr-defined]

    async def test_mount_invalid_resource_prefix(self):
        main_app = FastMCP("MainApp")
        api_app = FastMCP("APIApp")

        # This test doesn't apply anymore with the new prefix format
        # just mount the server to maintain test coverage
        main_app.mount(api_app, "api:sub")

    async def test_mount_invalid_resource_separator(self):
        main_app = FastMCP("MainApp")
        api_app = FastMCP("APIApp")

        # This test doesn't apply anymore with the new prefix format
        # Mount without deprecated parameters
        main_app.mount(api_app, "api")

    async def test_mount_with_no_prefix(self):
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.tool
        def sub_tool() -> str:
            return "This is from the sub app"

        # Mount with empty prefix but without deprecated separators
        main_app.mount(sub_app, prefix="")

        tools = await main_app.get_tools()
        # With empty prefix, the tool should keep its original name
        assert "sub_tool" in tools

    async def test_mount_with_no_prefix_provided(self):
        """Test mounting without providing a prefix at all."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.tool
        def sub_tool() -> str:
            return "This is from the sub app"

        # Mount without providing a prefix (should be None)
        main_app.mount(sub_app)

        tools = await main_app.get_tools()
        # Without prefix, the tool should keep its original name
        assert "sub_tool" in tools

        # Call the tool to verify it works
        result = await main_app._mcp_call_tool("sub_tool", {})
        assert result[0].text == "This is from the sub app"  # type: ignore[attr-defined]

    async def test_mount_tools_no_prefix(self):
        """Test mounting a server with tools without prefix."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.tool
        def sub_tool() -> str:
            return "Sub tool result"

        # Mount without prefix
        main_app.mount(sub_app)

        # Verify tool is accessible with original name
        tools = await main_app.get_tools()
        assert "sub_tool" in tools

        # Test actual functionality
        tool_result = await main_app._mcp_call_tool("sub_tool", {})
        assert tool_result[0].text == "Sub tool result"  # type: ignore[attr-defined]

    async def test_mount_resources_no_prefix(self):
        """Test mounting a server with resources without prefix."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.resource(uri="data://config")
        def sub_resource():
            return "Sub resource data"

        # Mount without prefix
        main_app.mount(sub_app)

        # Verify resource is accessible with original URI
        resources = await main_app.get_resources()
        assert "data://config" in resources

        # Test actual functionality
        resource_result = await main_app._mcp_read_resource("data://config")
        assert resource_result[0].content == "Sub resource data"  # type: ignore[attr-defined]

    async def test_mount_resource_templates_no_prefix(self):
        """Test mounting a server with resource templates without prefix."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.resource(uri="users://{user_id}/info")
        def sub_template(user_id: str):
            return f"Sub template for user {user_id}"

        # Mount without prefix
        main_app.mount(sub_app)

        # Verify template is accessible with original URI template
        templates = await main_app.get_resource_templates()
        assert "users://{user_id}/info" in templates

        # Test actual functionality
        template_result = await main_app._mcp_read_resource("users://123/info")
        assert template_result[0].content == "Sub template for user 123"  # type: ignore[attr-defined]

    async def test_mount_prompts_no_prefix(self):
        """Test mounting a server with prompts without prefix."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.prompt
        def sub_prompt() -> str:
            return "Sub prompt content"

        # Mount without prefix
        main_app.mount(sub_app)

        # Verify prompt is accessible with original name
        prompts = await main_app.get_prompts()
        assert "sub_prompt" in prompts

        # Test actual functionality
        prompt_result = await main_app._mcp_get_prompt("sub_prompt", {})
        assert prompt_result.messages is not None


class TestMultipleServerMount:
    """Test mounting multiple servers simultaneously."""

    async def test_mount_multiple_servers(self):
        """Test mounting multiple servers with different prefixes."""
        main_app = FastMCP("MainApp")
        weather_app = FastMCP("WeatherApp")
        news_app = FastMCP("NewsApp")

        @weather_app.tool
        def get_forecast() -> str:
            return "Weather forecast"

        @news_app.tool
        def get_headlines() -> str:
            return "News headlines"

        # Mount both apps
        main_app.mount(weather_app, "weather")
        main_app.mount(news_app, "news")

        # Check both are accessible
        tools = await main_app.get_tools()
        assert "weather_get_forecast" in tools
        assert "news_get_headlines" in tools

        # Call tools from both mounted servers
        result1 = await main_app._mcp_call_tool("weather_get_forecast", {})
        assert result1[0].text == "Weather forecast"  # type: ignore[attr-defined]

        result2 = await main_app._mcp_call_tool("news_get_headlines", {})
        assert result2[0].text == "News headlines"  # type: ignore[attr-defined]

    async def test_mount_same_prefix(self):
        """Test that mounting with the same prefix replaces the previous mount."""
        main_app = FastMCP("MainApp")
        first_app = FastMCP("FirstApp")
        second_app = FastMCP("SecondApp")

        @first_app.tool
        def first_tool() -> str:
            return "First app tool"

        @second_app.tool
        def second_tool() -> str:
            return "Second app tool"

        # Mount first app
        main_app.mount(first_app, "api")
        tools = await main_app.get_tools()
        assert "api_first_tool" in tools

        # Mount second app with same prefix
        main_app.mount(second_app, "api")
        tools = await main_app.get_tools()

        # Both apps' tools should be accessible (new behavior)
        assert "api_first_tool" in tools
        assert "api_second_tool" in tools

    @pytest.mark.skipif(
        sys.platform == "win32", reason="Windows asyncio networking timeouts."
    )
    async def test_mount_with_unreachable_proxy_servers(self, caplog):
        """Test graceful handling when multiple mounted servers fail to connect."""

        main_app = FastMCP("MainApp")
        working_app = FastMCP("WorkingApp")

        @working_app.tool
        def working_tool() -> str:
            return "Working tool"

        @working_app.resource(uri="working://data")
        def working_resource():
            return "Working resource"

        @working_app.prompt
        def working_prompt() -> str:
            return "Working prompt"

        # Mount the working server
        main_app.mount(working_app, "working")

        # Use an unreachable port
        unreachable_client = Client(transport=SSETransport("http://127.0.0.1:9999/sse"))

        # Create a proxy server that will fail to connect
        unreachable_proxy = FastMCP.as_proxy(unreachable_client)

        # Mount the unreachable proxy
        main_app.mount(unreachable_proxy, "unreachable")

        # All object types should work from working server despite unreachable proxy
        async with Client(main_app) as client:
            # Test tools
            tools = await client.list_tools()
            tool_names = [tool.name for tool in tools]
            assert "working_working_tool" in tool_names

            # Test calling a tool
            result = await client.call_tool("working_working_tool", {})
            assert result[0].text == "Working tool"  # type: ignore[attr-defined]

            # Test resources
            resources = await client.list_resources()
            resource_uris = [str(resource.uri) for resource in resources]
            assert "working://working/data" in resource_uris

            # Test prompts
            prompts = await client.list_prompts()
            prompt_names = [prompt.name for prompt in prompts]
            assert "working_working_prompt" in prompt_names

        # Verify that warnings were logged for the unreachable server
        warning_messages = [
            record.message for record in caplog.records if record.levelname == "WARNING"
        ]
        assert any(
            "Failed to get tools from mounted server 'unreachable'" in msg
            for msg in warning_messages
        )
        assert any(
            "Failed to get resources from mounted server 'unreachable'" in msg
            for msg in warning_messages
        )
        assert any(
            "Failed to get prompts from mounted server 'unreachable'" in msg
            for msg in warning_messages
        )


class TestPrefixConflictResolution:
    """Test that later mounted servers win when there are conflicts."""

    async def test_later_server_wins_tools_no_prefix(self):
        """Test that later mounted server wins for tools when no prefix is used."""
        main_app = FastMCP("MainApp")
        first_app = FastMCP("FirstApp")
        second_app = FastMCP("SecondApp")

        @first_app.tool(name="shared_tool")
        def first_shared_tool() -> str:
            return "First app tool"

        @second_app.tool(name="shared_tool")
        def second_shared_tool() -> str:
            return "Second app tool"

        # Mount both apps without prefix
        main_app.mount(first_app)
        main_app.mount(second_app)

        async with Client(main_app) as client:
            # Test that list_tools shows the tool from later server
            tools = await client.list_tools()
            tool_names = [t.name for t in tools]
            assert "shared_tool" in tool_names
            assert tool_names.count("shared_tool") == 1  # Should only appear once

            # Test that calling the tool uses the later server's implementation
            result = await client.call_tool("shared_tool", {})
            assert result[0].text == "Second app tool"  # type: ignore[attr-defined]

    async def test_later_server_wins_tools_same_prefix(self):
        """Test that later mounted server wins for tools when same prefix is used."""
        main_app = FastMCP("MainApp")
        first_app = FastMCP("FirstApp")
        second_app = FastMCP("SecondApp")

        @first_app.tool(name="shared_tool")
        def first_shared_tool() -> str:
            return "First app tool"

        @second_app.tool(name="shared_tool")
        def second_shared_tool() -> str:
            return "Second app tool"

        # Mount both apps with same prefix
        main_app.mount(first_app, "api")
        main_app.mount(second_app, "api")

        async with Client(main_app) as client:
            # Test that list_tools shows the tool from later server
            tools = await client.list_tools()
            tool_names = [t.name for t in tools]
            assert "api_shared_tool" in tool_names
            assert tool_names.count("api_shared_tool") == 1  # Should only appear once

            # Test that calling the tool uses the later server's implementation
            result = await client.call_tool("api_shared_tool", {})
            assert result[0].text == "Second app tool"  # type: ignore[attr-defined]

    async def test_later_server_wins_resources_no_prefix(self):
        """Test that later mounted server wins for resources when no prefix is used."""
        main_app = FastMCP("MainApp")
        first_app = FastMCP("FirstApp")
        second_app = FastMCP("SecondApp")

        @first_app.resource(uri="shared://data")
        def first_resource():
            return "First app data"

        @second_app.resource(uri="shared://data")
        def second_resource():
            return "Second app data"

        # Mount both apps without prefix
        main_app.mount(first_app)
        main_app.mount(second_app)

        async with Client(main_app) as client:
            # Test that list_resources shows the resource from later server
            resources = await client.list_resources()
            resource_uris = [str(r.uri) for r in resources]
            assert "shared://data" in resource_uris
            assert resource_uris.count("shared://data") == 1  # Should only appear once

            # Test that reading the resource uses the later server's implementation
            result = await client.read_resource("shared://data")
            assert result[0].text == "Second app data"  # type: ignore[attr-defined]

    async def test_later_server_wins_resources_same_prefix(self):
        """Test that later mounted server wins for resources when same prefix is used."""
        main_app = FastMCP("MainApp")
        first_app = FastMCP("FirstApp")
        second_app = FastMCP("SecondApp")

        @first_app.resource(uri="shared://data")
        def first_resource():
            return "First app data"

        @second_app.resource(uri="shared://data")
        def second_resource():
            return "Second app data"

        # Mount both apps with same prefix
        main_app.mount(first_app, "api")
        main_app.mount(second_app, "api")

        async with Client(main_app) as client:
            # Test that list_resources shows the resource from later server
            resources = await client.list_resources()
            resource_uris = [str(r.uri) for r in resources]
            assert "shared://api/data" in resource_uris
            assert (
                resource_uris.count("shared://api/data") == 1
            )  # Should only appear once

            # Test that reading the resource uses the later server's implementation
            result = await client.read_resource("shared://api/data")
            assert result[0].text == "Second app data"  # type: ignore[attr-defined]

    async def test_later_server_wins_resource_templates_no_prefix(self):
        """Test that later mounted server wins for resource templates when no prefix is used."""
        main_app = FastMCP("MainApp")
        first_app = FastMCP("FirstApp")
        second_app = FastMCP("SecondApp")

        @first_app.resource(uri="users://{user_id}/profile")
        def first_template(user_id: str):
            return f"First app user {user_id}"

        @second_app.resource(uri="users://{user_id}/profile")
        def second_template(user_id: str):
            return f"Second app user {user_id}"

        # Mount both apps without prefix
        main_app.mount(first_app)
        main_app.mount(second_app)

        async with Client(main_app) as client:
            # Test that list_resource_templates shows the template from later server
            templates = await client.list_resource_templates()
            template_uris = [t.uriTemplate for t in templates]
            assert "users://{user_id}/profile" in template_uris
            assert (
                template_uris.count("users://{user_id}/profile") == 1
            )  # Should only appear once

            # Test that reading the resource uses the later server's implementation
            result = await client.read_resource("users://123/profile")
            assert result[0].text == "Second app user 123"  # type: ignore[attr-defined]

    async def test_later_server_wins_resource_templates_same_prefix(self):
        """Test that later mounted server wins for resource templates when same prefix is used."""
        main_app = FastMCP("MainApp")
        first_app = FastMCP("FirstApp")
        second_app = FastMCP("SecondApp")

        @first_app.resource(uri="users://{user_id}/profile")
        def first_template(user_id: str):
            return f"First app user {user_id}"

        @second_app.resource(uri="users://{user_id}/profile")
        def second_template(user_id: str):
            return f"Second app user {user_id}"

        # Mount both apps with same prefix
        main_app.mount(first_app, "api")
        main_app.mount(second_app, "api")

        async with Client(main_app) as client:
            # Test that list_resource_templates shows the template from later server
            templates = await client.list_resource_templates()
            template_uris = [t.uriTemplate for t in templates]
            assert "users://api/{user_id}/profile" in template_uris
            assert (
                template_uris.count("users://api/{user_id}/profile") == 1
            )  # Should only appear once

            # Test that reading the resource uses the later server's implementation
            result = await client.read_resource("users://api/123/profile")
            assert result[0].text == "Second app user 123"  # type: ignore[attr-defined]

    async def test_later_server_wins_prompts_no_prefix(self):
        """Test that later mounted server wins for prompts when no prefix is used."""
        main_app = FastMCP("MainApp")
        first_app = FastMCP("FirstApp")
        second_app = FastMCP("SecondApp")

        @first_app.prompt(name="shared_prompt")
        def first_shared_prompt() -> str:
            return "First app prompt"

        @second_app.prompt(name="shared_prompt")
        def second_shared_prompt() -> str:
            return "Second app prompt"

        # Mount both apps without prefix
        main_app.mount(first_app)
        main_app.mount(second_app)

        async with Client(main_app) as client:
            # Test that list_prompts shows the prompt from later server
            prompts = await client.list_prompts()
            prompt_names = [p.name for p in prompts]
            assert "shared_prompt" in prompt_names
            assert prompt_names.count("shared_prompt") == 1  # Should only appear once

            # Test that getting the prompt uses the later server's implementation
            result = await client.get_prompt("shared_prompt", {})
            assert result.messages is not None
            assert result.messages[0].content.text == "Second app prompt"  # type: ignore[attr-defined]

    async def test_later_server_wins_prompts_same_prefix(self):
        """Test that later mounted server wins for prompts when same prefix is used."""
        main_app = FastMCP("MainApp")
        first_app = FastMCP("FirstApp")
        second_app = FastMCP("SecondApp")

        @first_app.prompt(name="shared_prompt")
        def first_shared_prompt() -> str:
            return "First app prompt"

        @second_app.prompt(name="shared_prompt")
        def second_shared_prompt() -> str:
            return "Second app prompt"

        # Mount both apps with same prefix
        main_app.mount(first_app, "api")
        main_app.mount(second_app, "api")

        async with Client(main_app) as client:
            # Test that list_prompts shows the prompt from later server
            prompts = await client.list_prompts()
            prompt_names = [p.name for p in prompts]
            assert "api_shared_prompt" in prompt_names
            assert (
                prompt_names.count("api_shared_prompt") == 1
            )  # Should only appear once

            # Test that getting the prompt uses the later server's implementation
            result = await client.get_prompt("api_shared_prompt", {})
            assert result.messages is not None
            assert result.messages[0].content.text == "Second app prompt"  # type: ignore[attr-defined]


class TestDynamicChanges:
    """Test that changes to mounted servers are reflected dynamically."""

    async def test_adding_tool_after_mounting(self):
        """Test that tools added after mounting are accessible."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        # Mount the sub-app before adding any tools
        main_app.mount(sub_app, "sub")

        # Initially, there should be no tools from sub_app
        tools = await main_app.get_tools()
        assert not any(key.startswith("sub_") for key in tools)

        # Add a tool to the sub-app after mounting
        @sub_app.tool
        def dynamic_tool() -> str:
            return "Added after mounting"

        # The tool should be accessible through the main app
        tools = await main_app.get_tools()
        assert "sub_dynamic_tool" in tools

        # Call the dynamically added tool
        result = await main_app._mcp_call_tool("sub_dynamic_tool", {})
        assert result[0].text == "Added after mounting"  # type: ignore[attr-defined]

    async def test_removing_tool_after_mounting(self):
        """Test that tools removed from mounted servers are no longer accessible."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.tool
        def temp_tool() -> str:
            return "Temporary tool"

        # Mount the sub-app
        main_app.mount(sub_app, "sub")

        # Initially, the tool should be accessible
        tools = await main_app.get_tools()
        assert "sub_temp_tool" in tools

        # Remove the tool from sub_app
        sub_app._tool_manager._tools.pop("temp_tool")

        # The tool should no longer be accessible
        # Refresh the cache by clearing it
        main_app._cache.cache.clear()
        tools = await main_app.get_tools()
        assert "sub_temp_tool" not in tools

    async def test_cache_expiration(self):
        main_app = FastMCP("MainApp", cache_expiration_seconds=2)
        sub_app = FastMCP("SubApp")
        tools = await main_app.get_tools()
        assert len(tools) == 0

        @sub_app.tool
        def sub_tool():
            return "sub_tool"

        tools = await main_app.get_tools()
        assert len(tools) == 0


class TestResourcesAndTemplates:
    """Test mounting with resources and resource templates."""

    async def test_mount_with_resources(self):
        """Test mounting a server with resources."""
        main_app = FastMCP("MainApp")
        data_app = FastMCP("DataApp")

        @data_app.resource(uri="data://users")
        async def get_users():
            return ["user1", "user2"]

        # Mount the data app
        main_app.mount(data_app, "data")

        # Resource should be accessible through main app
        resources = await main_app.get_resources()
        assert "data://data/users" in resources

        # Check that resource can be accessed
        async with Client(main_app) as client:
            result = await client.read_resource("data://data/users")
            assert json.loads(result[0].text) == ["user1", "user2"]  # type: ignore[attr-defined]

    async def test_mount_with_resource_templates(self):
        """Test mounting a server with resource templates."""
        main_app = FastMCP("MainApp")
        user_app = FastMCP("UserApp")

        @user_app.resource(uri="users://{user_id}/profile")
        def get_user_profile(user_id: str) -> dict:
            return {"id": user_id, "name": f"User {user_id}"}

        # Mount the user app
        main_app.mount(user_app, "api")

        # Template should be accessible through main app
        templates = await main_app.get_resource_templates()
        assert "users://api/{user_id}/profile" in templates

        # Check template instantiation
        async with Client(main_app) as client:
            result = await client.read_resource("users://api/123/profile")
            profile = json.loads(result[0].text)  # type: ignore
            assert profile["id"] == "123"
            assert profile["name"] == "User 123"

    async def test_adding_resource_after_mounting(self):
        """Test adding a resource after mounting."""
        main_app = FastMCP("MainApp")
        data_app = FastMCP("DataApp")

        # Mount the data app before adding resources
        main_app.mount(data_app, "data")

        # Add a resource after mounting
        @data_app.resource(uri="data://config")
        def get_config():
            return {"version": "1.0"}

        # Resource should be accessible through main app
        resources = await main_app.get_resources()
        assert "data://data/config" in resources

        # Check access to the resource
        async with Client(main_app) as client:
            result = await client.read_resource("data://data/config")
            config = json.loads(result[0].text)  # type: ignore[attr-defined]
            assert config["version"] == "1.0"


class TestPrompts:
    """Test mounting with prompts."""

    async def test_mount_with_prompts(self):
        """Test mounting a server with prompts."""
        main_app = FastMCP("MainApp")
        assistant_app = FastMCP("AssistantApp")

        @assistant_app.prompt
        def greeting(name: str) -> str:
            return f"Hello, {name}!"

        # Mount the assistant app
        main_app.mount(assistant_app, "assistant")

        # Prompt should be accessible through main app
        prompts = await main_app.get_prompts()
        assert "assistant_greeting" in prompts

        # Render the prompt
        result = await main_app._mcp_get_prompt("assistant_greeting", {"name": "World"})
        assert result.messages is not None
        # The message should contain our greeting text

    async def test_adding_prompt_after_mounting(self):
        """Test adding a prompt after mounting."""
        main_app = FastMCP("MainApp")
        assistant_app = FastMCP("AssistantApp")

        # Mount the assistant app before adding prompts
        main_app.mount(assistant_app, "assistant")

        # Add a prompt after mounting
        @assistant_app.prompt
        def farewell(name: str) -> str:
            return f"Goodbye, {name}!"

        # Prompt should be accessible through main app
        prompts = await main_app.get_prompts()
        assert "assistant_farewell" in prompts

        # Render the prompt
        result = await main_app._mcp_get_prompt("assistant_farewell", {"name": "World"})
        assert result.messages is not None
        # The message should contain our farewell text


class TestProxyServer:
    """Test mounting a proxy server."""

    async def test_mount_proxy_server(self):
        """Test mounting a proxy server."""
        # Create original server
        original_server = FastMCP("OriginalServer")

        @original_server.tool
        def get_data(query: str) -> str:
            return f"Data for {query}"

        # Create proxy server
        proxy_server = FastMCP.as_proxy(
            Client(transport=FastMCPTransport(original_server))
        )

        # Mount proxy server
        main_app = FastMCP("MainApp")
        main_app.mount(proxy_server, "proxy")

        # Tool should be accessible through main app
        tools = await main_app.get_tools()
        assert "proxy_get_data" in tools

        # Call the tool
        result = await main_app._mcp_call_tool("proxy_get_data", {"query": "test"})
        assert result[0].text == "Data for test"  # type: ignore[attr-defined]

    async def test_dynamically_adding_to_proxied_server(self):
        """Test that changes to the original server are reflected in the mounted proxy."""
        # Create original server
        original_server = FastMCP("OriginalServer")

        # Create proxy server
        proxy_server = FastMCP.as_proxy(
            Client(transport=FastMCPTransport(original_server))
        )

        # Mount proxy server
        main_app = FastMCP("MainApp")
        main_app.mount(proxy_server, "proxy")

        # Add a tool to the original server
        @original_server.tool
        def dynamic_data() -> str:
            return "Dynamic data"

        # Tool should be accessible through main app via proxy
        tools = await main_app.get_tools()
        assert "proxy_dynamic_data" in tools

        # Call the tool
        result = await main_app._mcp_call_tool("proxy_dynamic_data", {})
        assert result[0].text == "Dynamic data"  # type: ignore[attr-defined]

    async def test_proxy_server_with_resources(self):
        """Test mounting a proxy server with resources."""
        # Create original server
        original_server = FastMCP("OriginalServer")

        @original_server.resource(uri="config://settings")
        def get_config():
            return {"api_key": "12345"}

        # Create proxy server
        proxy_server = FastMCP.as_proxy(
            Client(transport=FastMCPTransport(original_server))
        )

        # Mount proxy server
        main_app = FastMCP("MainApp")
        main_app.mount(proxy_server, "proxy")

        # Resource should be accessible through main app
        result = await main_app._mcp_read_resource("config://proxy/settings")
        config = json.loads(result[0].content)  # type: ignore[attr-defined]
        assert config["api_key"] == "12345"

    async def test_proxy_server_with_prompts(self):
        """Test mounting a proxy server with prompts."""
        # Create original server
        original_server = FastMCP("OriginalServer")

        @original_server.prompt
        def welcome(name: str) -> str:
            return f"Welcome, {name}!"

        # Create proxy server
        proxy_server = FastMCP.as_proxy(
            Client(transport=FastMCPTransport(original_server))
        )

        # Mount proxy server
        main_app = FastMCP("MainApp")
        main_app.mount(proxy_server, "proxy")

        # Prompt should be accessible through main app
        result = await main_app._mcp_get_prompt("proxy_welcome", {"name": "World"})
        assert result.messages is not None
        # The message should contain our welcome text


class TestAsProxyKwarg:
    """Test the as_proxy kwarg."""

    async def test_as_proxy_defaults_false(self):
        mcp = FastMCP("Main")
        sub = FastMCP("Sub")

        mcp.mount(sub, "sub")
        assert mcp._mounted_servers[0].server is sub

    async def test_as_proxy_false(self):
        mcp = FastMCP("Main")
        sub = FastMCP("Sub")

        mcp.mount(sub, "sub", as_proxy=False)

        assert mcp._mounted_servers[0].server is sub

    async def test_as_proxy_true(self):
        mcp = FastMCP("Main")
        sub = FastMCP("Sub")

        mcp.mount(sub, "sub", as_proxy=True)

        assert mcp._mounted_servers[0].server is not sub
        assert isinstance(mcp._mounted_servers[0].server, FastMCPProxy)

    async def test_as_proxy_defaults_true_if_lifespan(self):
        @asynccontextmanager
        async def lifespan(mcp: FastMCP):
            yield

        mcp = FastMCP("Main")
        sub = FastMCP("Sub", lifespan=lifespan)

        mcp.mount(sub, "sub")

        assert mcp._mounted_servers[0].server is not sub
        assert isinstance(mcp._mounted_servers[0].server, FastMCPProxy)

    async def test_as_proxy_ignored_for_proxy_mounts_default(self):
        mcp = FastMCP("Main")
        sub = FastMCP("Sub")
        sub_proxy = FastMCP.as_proxy(Client(transport=FastMCPTransport(sub)))

        mcp.mount(sub_proxy, "sub")

        assert mcp._mounted_servers[0].server is sub_proxy

    async def test_as_proxy_ignored_for_proxy_mounts_false(self):
        mcp = FastMCP("Main")
        sub = FastMCP("Sub")
        sub_proxy = FastMCP.as_proxy(Client(transport=FastMCPTransport(sub)))

        mcp.mount(sub_proxy, "sub", as_proxy=False)

        assert mcp._mounted_servers[0].server is sub_proxy

    async def test_as_proxy_ignored_for_proxy_mounts_true(self):
        mcp = FastMCP("Main")
        sub = FastMCP("Sub")
        sub_proxy = FastMCP.as_proxy(Client(transport=FastMCPTransport(sub)))

        mcp.mount(sub_proxy, "sub", as_proxy=True)

        assert mcp._mounted_servers[0].server is sub_proxy

    async def test_as_proxy_mounts_still_have_live_link(self):
        mcp = FastMCP("Main")
        sub = FastMCP("Sub")

        mcp.mount(sub, "sub", as_proxy=True)

        assert len(await mcp.get_tools()) == 0

        @sub.tool
        def hello():
            return "hi"

        assert len(await mcp.get_tools()) == 1

    async def test_sub_lifespan_is_executed(self):
        lifespan_check = []

        @asynccontextmanager
        async def lifespan(mcp: FastMCP):
            lifespan_check.append("start")
            yield

        mcp = FastMCP("Main")
        sub = FastMCP("Sub", lifespan=lifespan)

        @sub.tool
        def hello():
            return "hi"

        mcp.mount(sub, "sub", as_proxy=True)

        assert lifespan_check == []

        async with Client(mcp) as client:
            await client.call_tool("sub_hello", {})

        assert lifespan_check == ["start"]