fastmcp/tests/server/test_mount.py

import json
import sys
from contextlib import asynccontextmanager

import pytest
from mcp.types import TextContent

from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.client.transports import FastMCPTransport, SSETransport
from fastmcp.exceptions import NotFoundError
from fastmcp.server.providers import FastMCPProvider
from fastmcp.server.providers.proxy import FastMCPProxy
from fastmcp.server.providers.wrapped_provider import _WrappedProvider
from fastmcp.server.transforms import Namespace
from fastmcp.tools.tool import Tool
from fastmcp.tools.tool_transform import TransformedTool
from fastmcp.utilities.tests import caplog_for_fastmcp


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

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

        sub_tool = Tool.from_function(tool)

        transformed_tool = TransformedTool.from_tool(
            name="transformed_tool", tool=sub_tool
        )

        sub_app = FastMCP("SubApp", tools=[transformed_tool, sub_tool])

        # 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.list_tools()
        assert any(t.name == "sub_tool" for t in tools)
        assert any(t.name == "sub_transformed_tool" for t in tools)

        result = await main_app.call_tool("sub_tool", {})
        assert result.structured_content == {"result": "This is from the sub app"}

    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.list_tools()
        assert any(t.name == "sub_greet" for t in tools)

        # Call the tool
        result = await main_app.call_tool("sub_greet", {"name": "World"})
        assert result.structured_content == {"result": "Hello, World!"}

    @pytest.mark.parametrize("prefix", ["", None])
    async def test_mount_with_no_prefix(self, prefix):
        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, namespace=prefix)

        tools = await main_app.list_tools()
        # With empty prefix, the tool should keep its original name
        assert any(t.name == "sub_tool" for t 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.list_tools()
        # Without prefix, the tool should keep its original name
        assert any(t.name == "sub_tool" for t in tools)

        # Call the tool to verify it works
        result = await main_app.call_tool("sub_tool", {})
        assert result.structured_content == {"result": "This is from the sub app"}

    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.list_tools()
        assert any(t.name == "sub_tool" for t in tools)

        # Test actual functionality
        tool_result = await main_app.call_tool("sub_tool", {})
        assert tool_result.structured_content == {"result": "Sub tool result"}

    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.list_resources()
        assert any(str(r.uri) == "data://config" for r in resources)

        # Test actual functionality
        resource_result = await main_app.read_resource("data://config")
        assert resource_result.contents[0].content == "Sub resource data"

    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.list_resource_templates()
        assert any(t.uri_template == "users://{user_id}/info" for t in templates)

        # Test actual functionality
        template_result = await main_app.read_resource("users://123/info")
        assert template_result.contents[0].content == "Sub template for user 123"

    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.list_prompts()
        assert any(p.name == "sub_prompt" for p in prompts)

        # Test actual functionality
        prompt_result = await main_app.render_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.list_tools()
        assert any(t.name == "weather_get_forecast" for t in tools)
        assert any(t.name == "news_get_headlines" for t in tools)

        # Call tools from both mounted servers
        result1 = await main_app.call_tool("weather_get_forecast", {})
        assert result1.structured_content == {"result": "Weather forecast"}
        result2 = await main_app.call_tool("news_get_headlines", {})
        assert result2.structured_content == {"result": "News headlines"}

    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.list_tools()
        assert any(t.name == "api_first_tool" for t in tools)

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

        # Both apps' tools should be accessible (new behavior)
        assert any(t.name == "api_first_tool" for t in tools)
        assert any(t.name == "api_second_tool" for t 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/"),
            name="unreachable_client",
        )

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

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

        # All object types should work from working server despite unreachable proxy
        with caplog_for_fastmcp(caplog):
            async with Client(main_app, name="main_app_client") 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.data == "Working tool"

                # 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 errors were logged for the unreachable provider (at DEBUG level)
        debug_messages = [
            record.message for record in caplog.records if record.levelname == "DEBUG"
        ]
        assert any(
            "Error during list_tools from provider" in msg for msg in debug_messages
        )
        assert any(
            "Error during list_resources from provider" in msg for msg in debug_messages
        )
        assert any(
            "Error during list_prompts from provider" in msg for msg in debug_messages
        )


class TestPrefixConflictResolution:
    """Test that first registered provider wins when there are conflicts.

    Provider semantics: 'Providers are queried in registration order; first non-None wins'
    """

    async def test_first_server_wins_tools_no_prefix(self):
        """Test that first 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)

        # list_tools returns all components; execution uses first match
        tools = await main_app.list_tools()
        tool_names = [t.name for t in tools]
        assert "shared_tool" in tool_names

        # Test that calling the tool uses the first server's implementation
        result = await main_app.call_tool("shared_tool", {})
        assert result.structured_content == {"result": "First app tool"}

    async def test_first_server_wins_tools_same_prefix(self):
        """Test that first 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")

        # list_tools returns all components; execution uses first match
        tools = await main_app.list_tools()
        tool_names = [t.name for t in tools]
        assert "api_shared_tool" in tool_names

        # Test that calling the tool uses the first server's implementation
        result = await main_app.call_tool("api_shared_tool", {})
        assert result.structured_content == {"result": "First app tool"}

    async def test_first_server_wins_resources_no_prefix(self):
        """Test that first 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)

        # list_resources returns all components; execution uses first match
        resources = await main_app.list_resources()
        resource_uris = [str(r.uri) for r in resources]
        assert "shared://data" in resource_uris

        # Test that reading the resource uses the first server's implementation
        result = await main_app.read_resource("shared://data")
        assert result.contents[0].content == "First app data"

    async def test_first_server_wins_resources_same_prefix(self):
        """Test that first 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")

        # list_resources returns all components; execution uses first match
        resources = await main_app.list_resources()
        resource_uris = [str(r.uri) for r in resources]
        assert "shared://api/data" in resource_uris

        # Test that reading the resource uses the first server's implementation
        result = await main_app.read_resource("shared://api/data")
        assert result.contents[0].content == "First app data"

    async def test_first_server_wins_resource_templates_no_prefix(self):
        """Test that first 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)

        # list_resource_templates returns all components; execution uses first match
        templates = await main_app.list_resource_templates()
        template_uris = [t.uri_template for t in templates]
        assert "users://{user_id}/profile" in template_uris

        # Test that reading the resource uses the first server's implementation
        result = await main_app.read_resource("users://123/profile")
        assert result.contents[0].content == "First app user 123"

    async def test_first_server_wins_resource_templates_same_prefix(self):
        """Test that first 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")

        # list_resource_templates returns all components; execution uses first match
        templates = await main_app.list_resource_templates()
        template_uris = [t.uri_template for t in templates]
        assert "users://api/{user_id}/profile" in template_uris

        # Test that reading the resource uses the first server's implementation
        result = await main_app.read_resource("users://api/123/profile")
        assert result.contents[0].content == "First app user 123"

    async def test_first_server_wins_prompts_no_prefix(self):
        """Test that first 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)

        # list_prompts returns all components; execution uses first match
        prompts = await main_app.list_prompts()
        prompt_names = [p.name for p in prompts]
        assert "shared_prompt" in prompt_names

        # Test that getting the prompt uses the first server's implementation
        result = await main_app.render_prompt("shared_prompt")
        assert result.messages is not None
        assert isinstance(result.messages[0].content, TextContent)
        assert result.messages[0].content.text == "First app prompt"

    async def test_first_server_wins_prompts_same_prefix(self):
        """Test that first 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")

        # list_prompts returns all components; execution uses first match
        prompts = await main_app.list_prompts()
        prompt_names = [p.name for p in prompts]
        assert "api_shared_prompt" in prompt_names

        # Test that getting the prompt uses the first server's implementation
        result = await main_app.render_prompt("api_shared_prompt")
        assert result.messages is not None
        assert isinstance(result.messages[0].content, TextContent)
        assert result.messages[0].content.text == "First app prompt"


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.list_tools()
        assert not any(t.name.startswith("sub_") for t 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.list_tools()
        assert any(t.name == "sub_dynamic_tool" for t in tools)

        # Call the dynamically added tool
        result = await main_app.call_tool("sub_dynamic_tool", {})
        assert result.structured_content == {"result": "Added after mounting"}

    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.list_tools()
        assert any(t.name == "sub_temp_tool" for t in tools)

        # Remove the tool from sub_app using public API
        sub_app.remove_tool("temp_tool")

        # The tool should no longer be accessible
        tools = await main_app.list_tools()
        assert not any(t.name == "sub_temp_tool" for t in tools)


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() -> str:
            return "user1, user2"

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

        # Resource should be accessible through main app
        resources = await main_app.list_resources()
        assert any(str(r.uri) == "data://data/users" for r in resources)

        # Check that resource can be accessed
        result = await main_app.read_resource("data://data/users")
        assert len(result.contents) == 1
        # Note: The function returns "user1, user2" which is not valid JSON
        # This test should be updated to return proper JSON or check the string directly
        assert result.contents[0].content == "user1, user2"

    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) -> str:
            return json.dumps({"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.list_resource_templates()
        assert any(t.uri_template == "users://api/{user_id}/profile" for t in templates)

        # Check template instantiation
        result = await main_app.read_resource("users://api/123/profile")
        assert len(result.contents) == 1
        profile = json.loads(result.contents[0].content)
        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() -> str:
            return json.dumps({"version": "1.0"})

        # Resource should be accessible through main app
        resources = await main_app.list_resources()
        assert any(str(r.uri) == "data://data/config" for r in resources)

        # Check access to the resource
        result = await main_app.read_resource("data://data/config")
        assert len(result.contents) == 1
        config = json.loads(result.contents[0].content)
        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.list_prompts()
        assert any(p.name == "assistant_greeting" for p in prompts)

        # Render the prompt
        result = await main_app.render_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.list_prompts()
        assert any(p.name == "assistant_farewell" for p in prompts)

        # Render the prompt
        result = await main_app.render_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(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.list_tools()
        assert any(t.name == "proxy_get_data" for t in tools)

        # Call the tool
        result = await main_app.call_tool("proxy_get_data", {"query": "test"})
        assert result.structured_content == {"result": "Data for test"}

    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(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.list_tools()
        assert any(t.name == "proxy_dynamic_data" for t in tools)

        # Call the tool
        result = await main_app.call_tool("proxy_dynamic_data", {})
        assert result.structured_content == {"result": "Dynamic data"}

    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() -> str:
            return json.dumps({"api_key": "12345"})

        # Create proxy server
        proxy_server = FastMCP.as_proxy(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.read_resource("config://proxy/settings")
        assert len(result.contents) == 1
        config = json.loads(result.contents[0].content)
        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(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.render_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")

        @sub.tool
        def sub_tool() -> str:
            return "test"

        mcp.mount(sub, "sub")
        # Index 1 because LocalProvider is at index 0
        provider = mcp.providers[1]
        # Provider is wrapped with Namespace transform
        assert isinstance(provider, _WrappedProvider)
        assert len(provider._transforms) == 1
        assert isinstance(provider._transforms[0], Namespace)
        # Inner provider is FastMCPProvider
        assert isinstance(provider._inner, FastMCPProvider)
        assert provider._inner.server is sub
        # Verify namespace is applied
        tools = await mcp.list_tools()
        assert {t.name for t in tools} == {"sub_sub_tool"}

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

        @sub.tool
        def sub_tool() -> str:
            return "test"

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

        # Index 1 because LocalProvider is at index 0
        provider = mcp.providers[1]
        # Provider is wrapped with Namespace transform
        assert isinstance(provider, _WrappedProvider)
        assert len(provider._transforms) == 1
        assert isinstance(provider._transforms[0], Namespace)
        # Inner provider is FastMCPProvider
        assert isinstance(provider._inner, FastMCPProvider)
        assert provider._inner.server is sub
        # Verify namespace is applied
        tools = await mcp.list_tools()
        assert {t.name for t in tools} == {"sub_sub_tool"}

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

        @sub.tool
        def sub_tool() -> str:
            return "test"

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

        # Index 1 because LocalProvider is at index 0
        provider = mcp.providers[1]
        # Provider is wrapped with Namespace transform
        assert isinstance(provider, _WrappedProvider)
        assert len(provider._transforms) == 1
        assert isinstance(provider._transforms[0], Namespace)
        # Inner provider is FastMCPProvider wrapping a proxy
        assert isinstance(provider._inner, FastMCPProvider)
        assert provider._inner.server is not sub
        assert isinstance(provider._inner.server, FastMCPProxy)
        # Verify namespace is applied
        tools = await mcp.list_tools()
        assert {t.name for t in tools} == {"sub_sub_tool"}

    async def test_lifespan_server_mounted_directly(self):
        """Test that servers with lifespan are mounted directly (not auto-proxied).

        Since FastMCPProvider now handles lifespan via the provider lifespan interface,
        there's no need to auto-convert to a proxy. The server is mounted directly.
        """

        @asynccontextmanager
        async def server_lifespan(mcp: FastMCP):
            yield

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

        @sub.tool
        def sub_tool() -> str:
            return "test"

        mcp.mount(sub, "sub")

        # Server should be mounted directly without auto-proxying
        # Index 1 because LocalProvider is at index 0
        provider = mcp.providers[1]
        # Provider is wrapped with Namespace transform
        assert isinstance(provider, _WrappedProvider)
        assert len(provider._transforms) == 1
        assert isinstance(provider._transforms[0], Namespace)
        # Inner provider is FastMCPProvider
        assert isinstance(provider._inner, FastMCPProvider)
        assert provider._inner.server is sub
        # Verify namespace is applied
        tools = await mcp.list_tools()
        assert {t.name for t in tools} == {"sub_sub_tool"}

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

        mcp.mount(sub_proxy, "sub")

        # Index 1 because LocalProvider is at index 0
        provider = mcp.providers[1]
        # Provider is wrapped with Namespace transform
        assert isinstance(provider, _WrappedProvider)
        assert len(provider._transforms) == 1
        assert isinstance(provider._transforms[0], Namespace)
        # Inner provider is FastMCPProvider
        assert isinstance(provider._inner, FastMCPProvider)
        assert provider._inner.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(FastMCPTransport(sub))

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

        # Index 1 because LocalProvider is at index 0
        provider = mcp.providers[1]
        # Provider is wrapped with Namespace transform
        assert isinstance(provider, _WrappedProvider)
        assert len(provider._transforms) == 1
        assert isinstance(provider._transforms[0], Namespace)
        # Inner provider is FastMCPProvider
        assert isinstance(provider._inner, FastMCPProvider)
        assert provider._inner.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(FastMCPTransport(sub))

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

        # Index 1 because LocalProvider is at index 0
        provider = mcp.providers[1]
        # Provider is wrapped with Namespace transform
        assert isinstance(provider, _WrappedProvider)
        assert len(provider._transforms) == 1
        assert isinstance(provider._transforms[0], Namespace)
        # Inner provider is FastMCPProvider
        assert isinstance(provider._inner, FastMCPProvider)
        assert provider._inner.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.list_tools()) == 0

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

        assert len(await mcp.list_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, as_proxy=True)

        assert lifespan_check == []

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

        # Lifespan is executed at least once (may be multiple times for proxy connections)
        assert len(lifespan_check) >= 1
        assert all(x == "start" for x in lifespan_check)


class TestResourceUriPrefixing:
    """Test that resource and resource template URIs get prefixed when mounted (names are NOT prefixed)."""

    async def test_resource_uri_prefixing(self):
        """Test that resource URIs are prefixed when mounted (names are NOT prefixed)."""

        # Create a sub-app with a resource
        sub_app = FastMCP("SubApp")

        @sub_app.resource("resource://my_resource")
        def my_resource() -> str:
            return "Resource content"

        # Create main app and mount sub-app with prefix
        main_app = FastMCP("MainApp")
        main_app.mount(sub_app, "prefix")

        # Get resources from main app
        resources = await main_app.list_resources()

        # Should have prefixed key (using path format: resource://prefix/resource_name)
        assert any(str(r.uri) == "resource://prefix/my_resource" for r in resources)

        # The resource name should NOT be prefixed (only URI is prefixed)
        resource = next(
            r for r in resources if str(r.uri) == "resource://prefix/my_resource"
        )
        assert resource.name == "my_resource"

    async def test_resource_template_uri_prefixing(self):
        """Test that resource template URIs are prefixed when mounted (names are NOT prefixed)."""

        # Create a sub-app with a resource template
        sub_app = FastMCP("SubApp")

        @sub_app.resource("resource://user/{user_id}")
        def user_template(user_id: str) -> str:
            return f"User {user_id} data"

        # Create main app and mount sub-app with prefix
        main_app = FastMCP("MainApp")
        main_app.mount(sub_app, "prefix")

        # Get resource templates from main app
        templates = await main_app.list_resource_templates()

        # Should have prefixed key (using path format: resource://prefix/template_uri)
        assert any(
            t.uri_template == "resource://prefix/user/{user_id}" for t in templates
        )

        # The template name should NOT be prefixed (only URI template is prefixed)
        template = next(
            t for t in templates if t.uri_template == "resource://prefix/user/{user_id}"
        )
        assert template.name == "user_template"


class TestParentTagFiltering:
    """Test that parent server tag filters apply recursively to mounted servers."""

    async def test_parent_include_tags_filters_mounted_tools(self):
        """Test that parent include_tags filters out non-matching mounted tools."""
        parent = FastMCP("Parent", include_tags={"allowed"})
        mounted = FastMCP("Mounted")

        @mounted.tool(tags={"allowed"})
        def allowed_tool() -> str:
            return "allowed"

        @mounted.tool(tags={"blocked"})
        def blocked_tool() -> str:
            return "blocked"

        parent.mount(mounted)

        tools = await parent.list_tools()
        tool_names = {t.name for t in tools}
        assert "allowed_tool" in tool_names
        assert "blocked_tool" not in tool_names

        # Verify execution also respects filters
        result = await parent.call_tool("allowed_tool", {})
        assert result.structured_content == {"result": "allowed"}

        with pytest.raises(NotFoundError, match="Unknown tool"):
            await parent.call_tool("blocked_tool", {})

    async def test_parent_exclude_tags_filters_mounted_tools(self):
        """Test that parent exclude_tags filters out matching mounted tools."""
        parent = FastMCP("Parent", exclude_tags={"blocked"})
        mounted = FastMCP("Mounted")

        @mounted.tool(tags={"production"})
        def production_tool() -> str:
            return "production"

        @mounted.tool(tags={"blocked"})
        def blocked_tool() -> str:
            return "blocked"

        parent.mount(mounted)

        tools = await parent.list_tools()
        tool_names = {t.name for t in tools}
        assert "production_tool" in tool_names
        assert "blocked_tool" not in tool_names

    async def test_parent_filters_apply_to_mounted_resources(self):
        """Test that parent tag filters apply to mounted resources."""
        parent = FastMCP("Parent", include_tags={"allowed"})
        mounted = FastMCP("Mounted")

        @mounted.resource("resource://allowed", tags={"allowed"})
        def allowed_resource() -> str:
            return "allowed"

        @mounted.resource("resource://blocked", tags={"blocked"})
        def blocked_resource() -> str:
            return "blocked"

        parent.mount(mounted)

        resources = await parent.list_resources()
        resource_uris = {str(r.uri) for r in resources}
        assert "resource://allowed" in resource_uris
        assert "resource://blocked" not in resource_uris

    async def test_parent_filters_apply_to_mounted_prompts(self):
        """Test that parent tag filters apply to mounted prompts."""
        parent = FastMCP("Parent", exclude_tags={"blocked"})
        mounted = FastMCP("Mounted")

        @mounted.prompt(tags={"allowed"})
        def allowed_prompt() -> str:
            return "allowed"

        @mounted.prompt(tags={"blocked"})
        def blocked_prompt() -> str:
            return "blocked"

        parent.mount(mounted)

        prompts = await parent.list_prompts()
        prompt_names = {p.name for p in prompts}
        assert "allowed_prompt" in prompt_names
        assert "blocked_prompt" not in prompt_names


class TestCustomRouteForwarding:
    """Test that custom HTTP routes from mounted servers are forwarded."""

    async def test_get_additional_http_routes_empty(self):
        """Test _get_additional_http_routes returns empty list for server with no routes."""
        server = FastMCP("TestServer")
        routes = server._get_additional_http_routes()
        assert routes == []

    async def test_get_additional_http_routes_with_custom_route(self):
        """Test _get_additional_http_routes returns server's own routes."""
        server = FastMCP("TestServer")

        @server.custom_route("/test", methods=["GET"])
        async def test_route(request):
            from starlette.responses import JSONResponse

            return JSONResponse({"message": "test"})

        routes = server._get_additional_http_routes()
        assert len(routes) == 1
        assert hasattr(routes[0], "path")
        assert routes[0].path == "/test"

    async def test_mounted_servers_tracking(self):
        """Test that providers list tracks mounted servers correctly."""
        from fastmcp.server.providers.local_provider import LocalProvider

        main_server = FastMCP("MainServer")
        sub_server1 = FastMCP("SubServer1")
        sub_server2 = FastMCP("SubServer2")

        @sub_server1.tool
        def tool1() -> str:
            return "1"

        @sub_server2.tool
        def tool2() -> str:
            return "2"

        # Initially only LocalProvider
        assert len(main_server.providers) == 1
        assert isinstance(main_server.providers[0], LocalProvider)

        # Mount first server
        main_server.mount(sub_server1, "sub1")
        assert len(main_server.providers) == 2
        # LocalProvider is at index 0, mounted provider (wrapped) at index 1
        provider1 = main_server.providers[1]
        assert isinstance(provider1, _WrappedProvider)
        assert isinstance(provider1._inner, FastMCPProvider)
        assert provider1._inner.server == sub_server1

        # Mount second server
        main_server.mount(sub_server2, "sub2")
        assert len(main_server.providers) == 3
        provider2 = main_server.providers[2]
        assert isinstance(provider2, _WrappedProvider)
        assert isinstance(provider2._inner, FastMCPProvider)
        assert provider2._inner.server == sub_server2

        # Verify namespacing is applied by checking tool names
        tools = await main_server.list_tools()
        tool_names = {t.name for t in tools}
        assert tool_names == {"sub1_tool1", "sub2_tool2"}

    async def test_multiple_routes_same_server(self):
        """Test that multiple custom routes from same server are all included."""
        server = FastMCP("TestServer")

        @server.custom_route("/route1", methods=["GET"])
        async def route1(request):
            from starlette.responses import JSONResponse

            return JSONResponse({"message": "route1"})

        @server.custom_route("/route2", methods=["POST"])
        async def route2(request):
            from starlette.responses import JSONResponse

            return JSONResponse({"message": "route2"})

        routes = server._get_additional_http_routes()
        assert len(routes) == 2
        route_paths = [route.path for route in routes if hasattr(route, "path")]
        assert "/route1" in route_paths
        assert "/route2" in route_paths


class TestDeeplyNestedMount:
    """Test deeply nested mount scenarios (3+ levels deep).

    This tests the fix for https://github.com/jlowin/fastmcp/issues/2583
    where tools/resources/prompts mounted more than 2 levels deep would fail
    to invoke even though they were correctly listed.
    """

    async def test_three_level_nested_tool_invocation(self):
        """Test invoking tools from servers mounted 3 levels deep."""
        root = FastMCP("root")
        middle = FastMCP("middle")
        leaf = FastMCP("leaf")

        @leaf.tool
        def add(a: int, b: int) -> int:
            return a + b

        @middle.tool
        def multiply(a: int, b: int) -> int:
            return a * b

        middle.mount(leaf, namespace="leaf")
        root.mount(middle, namespace="middle")

        # Tool at level 2 should work
        result = await root.call_tool("middle_multiply", {"a": 3, "b": 4})
        assert result.structured_content == {"result": 12}

        # Tool at level 3 should also work (this was the bug)
        result = await root.call_tool("middle_leaf_add", {"a": 5, "b": 7})
        assert result.structured_content == {"result": 12}

    async def test_three_level_nested_resource_invocation(self):
        """Test reading resources from servers mounted 3 levels deep."""
        root = FastMCP("root")
        middle = FastMCP("middle")
        leaf = FastMCP("leaf")

        @leaf.resource("leaf://data")
        def leaf_data() -> str:
            return "leaf data"

        @middle.resource("middle://data")
        def middle_data() -> str:
            return "middle data"

        middle.mount(leaf, namespace="leaf")
        root.mount(middle, namespace="middle")

        # Resource at level 2 should work
        result = await root.read_resource("middle://middle/data")
        assert result.contents[0].content == "middle data"

        # Resource at level 3 should also work
        result = await root.read_resource("leaf://middle/leaf/data")
        assert result.contents[0].content == "leaf data"

    async def test_three_level_nested_resource_template_invocation(self):
        """Test reading resource templates from servers mounted 3 levels deep."""
        root = FastMCP("root")
        middle = FastMCP("middle")
        leaf = FastMCP("leaf")

        @leaf.resource("leaf://item/{id}")
        def leaf_item(id: str) -> str:
            return f"leaf item {id}"

        @middle.resource("middle://item/{id}")
        def middle_item(id: str) -> str:
            return f"middle item {id}"

        middle.mount(leaf, namespace="leaf")
        root.mount(middle, namespace="middle")

        # Resource template at level 2 should work
        result = await root.read_resource("middle://middle/item/42")
        assert result.contents[0].content == "middle item 42"

        # Resource template at level 3 should also work
        result = await root.read_resource("leaf://middle/leaf/item/99")
        assert result.contents[0].content == "leaf item 99"

    async def test_three_level_nested_prompt_invocation(self):
        """Test getting prompts from servers mounted 3 levels deep."""
        root = FastMCP("root")
        middle = FastMCP("middle")
        leaf = FastMCP("leaf")

        @leaf.prompt
        def leaf_prompt(name: str) -> str:
            return f"Hello from leaf: {name}"

        @middle.prompt
        def middle_prompt(name: str) -> str:
            return f"Hello from middle: {name}"

        middle.mount(leaf, namespace="leaf")
        root.mount(middle, namespace="middle")

        # Prompt at level 2 should work
        result = await root.render_prompt("middle_middle_prompt", {"name": "World"})
        assert isinstance(result.messages[0].content, TextContent)
        assert "Hello from middle: World" in result.messages[0].content.text

        # Prompt at level 3 should also work
        result = await root.render_prompt("middle_leaf_leaf_prompt", {"name": "Test"})
        assert isinstance(result.messages[0].content, TextContent)
        assert "Hello from leaf: Test" in result.messages[0].content.text

    async def test_four_level_nested_tool_invocation(self):
        """Test invoking tools from servers mounted 4 levels deep."""
        root = FastMCP("root")
        level1 = FastMCP("level1")
        level2 = FastMCP("level2")
        level3 = FastMCP("level3")

        @level3.tool
        def deep_tool() -> str:
            return "very deep"

        level2.mount(level3, namespace="l3")
        level1.mount(level2, namespace="l2")
        root.mount(level1, namespace="l1")

        # Verify tool is listed
        tools = await root.list_tools()
        tool_names = [t.name for t in tools]
        assert "l1_l2_l3_deep_tool" in tool_names

        # Tool at level 4 should work
        result = await root.call_tool("l1_l2_l3_deep_tool", {})
        assert result.structured_content == {"result": "very deep"}


class TestToolNameOverrides:
    """Test tool and prompt name overrides in mount() (issue #2596)."""

    async def test_tool_names_override_via_transforms(self):
        """Test that tool_names renames tools via ToolTransform layer.

        Tool renames are applied first, then namespace prefixing.
        So original_tool → custom_name → prefix_custom_name.
        """
        sub = FastMCP("Sub")

        @sub.tool
        def original_tool() -> str:
            return "test"

        main = FastMCP("Main")
        # tool_names renames first, then namespace is applied
        main.mount(
            sub,
            namespace="prefix",
            tool_names={"original_tool": "custom_name"},
        )

        # Server introspection shows renamed + namespaced names
        tools = await main.list_tools()
        tool_names = [t.name for t in tools]
        assert "prefix_custom_name" in tool_names
        assert "original_tool" not in tool_names
        assert "prefix_original_tool" not in tool_names
        assert "custom_name" not in tool_names

    async def test_tool_names_override_applied_in_list_tools(self):
        """Test that tool_names override is reflected in list_tools()."""
        sub = FastMCP("Sub")

        @sub.tool
        def original_tool() -> str:
            return "test"

        main = FastMCP("Main")
        main.mount(
            sub,
            namespace="prefix",
            tool_names={"original_tool": "custom_name"},
        )

        tools = await main.list_tools()
        tool_names = [t.name for t in tools]
        assert "prefix_custom_name" in tool_names
        assert "prefix_original_tool" not in tool_names

    async def test_tool_call_with_overridden_name(self):
        """Test that overridden tool can be called by its new name."""
        sub = FastMCP("Sub")

        @sub.tool
        def original_tool() -> str:
            return "success"

        main = FastMCP("Main")
        main.mount(
            sub,
            namespace="prefix",
            tool_names={"original_tool": "renamed"},
        )

        # Tool is renamed then namespaced: original_tool → renamed → prefix_renamed
        result = await main.call_tool("prefix_renamed", {})
        assert result.structured_content == {"result": "success"}

    def test_duplicate_tool_rename_targets_raises_error(self):
        """Test that duplicate target names in tool_renames raises ValueError."""
        sub = FastMCP("Sub")
        main = FastMCP("Main")

        with pytest.raises(ValueError, match="duplicate target name"):
            main.mount(
                sub,
                tool_names={"tool_a": "same_name", "tool_b": "same_name"},
            )


class TestMountedServerDocketBehavior:
    """Regression tests for mounted server lifecycle behavior.

    These tests guard against architectural changes that could accidentally
    start Docket instances for mounted servers. Mounted servers should only
    run their user-defined lifespan, not the full _lifespan_manager which
    includes Docket creation.
    """

    async def test_mounted_server_does_not_have_docket(self):
        """Test that a mounted server doesn't create its own Docket.

        MountedProvider.lifespan() should call only the server's _lifespan
        (user-defined lifespan), not _lifespan_manager (which includes Docket).
        """
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        # Need a task-enabled component to trigger Docket initialization
        @main_app.tool(task=True)
        async def _trigger_docket() -> str:
            return "trigger"

        @sub_app.tool
        def my_tool() -> str:
            return "test"

        main_app.mount(sub_app, "sub")

        # After running the main app's lifespan, the sub app should not have
        # its own Docket instance
        async with Client(main_app) as client:
            # The main app should have a docket (created by _lifespan_manager)
            # because it has a task-enabled component
            assert main_app.docket is not None

            # The mounted sub app should NOT have its own docket
            # It uses the parent's docket for background tasks
            assert sub_app.docket is None

            # But the tool should still work (prefixed as sub_my_tool)
            result = await client.call_tool("sub_my_tool", {})
            assert result.data == "test"


class TestComponentServicePrefixLess:
    """Test that enable/disable works with prefix-less mounted servers."""

    async def test_enable_tool_prefixless_mount(self):
        """Test enabling a tool on a prefix-less mounted server."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.tool
        def my_tool() -> str:
            return "test"

        # Mount without prefix
        main_app.mount(sub_app)

        # Initially the tool is enabled
        tools = await main_app.list_tools()
        assert any(t.name == "my_tool" for t in tools)

        # Disable and re-enable
        main_app.disable(names={"my_tool"}, components={"tool"})
        # Verify tool is now disabled
        tools = await main_app.list_tools()
        assert not any(t.name == "my_tool" for t in tools)

        main_app.enable(names={"my_tool"}, components={"tool"})
        # Verify tool is now enabled
        tools = await main_app.list_tools()
        assert any(t.name == "my_tool" for t in tools)

    async def test_enable_resource_prefixless_mount(self):
        """Test enabling a resource on a prefix-less mounted server."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.resource(uri="data://test")
        def my_resource() -> str:
            return "test data"

        # Mount without prefix
        main_app.mount(sub_app)

        # Disable and re-enable
        main_app.disable(names={"data://test"}, components={"resource"})
        # Verify resource is now disabled
        resources = await main_app.list_resources()
        assert not any(str(r.uri) == "data://test" for r in resources)

        main_app.enable(names={"data://test"}, components={"resource"})
        # Verify resource is now enabled
        resources = await main_app.list_resources()
        assert any(str(r.uri) == "data://test" for r in resources)

    async def test_enable_prompt_prefixless_mount(self):
        """Test enabling a prompt on a prefix-less mounted server."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.prompt
        def my_prompt() -> str:
            return "test prompt"

        # Mount without prefix
        main_app.mount(sub_app)

        # Disable and re-enable
        main_app.disable(names={"my_prompt"}, components={"prompt"})
        # Verify prompt is now disabled
        prompts = await main_app.list_prompts()
        assert not any(p.name == "my_prompt" for p in prompts)

        main_app.enable(names={"my_prompt"}, components={"prompt"})
        # Verify prompt is now enabled
        prompts = await main_app.list_prompts()
        assert any(p.name == "my_prompt" for p in prompts)