From 084e52badd2dfbb75bcf06ddfdcad0459f5e3ed6 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Tue, 17 Jun 2025 20:23:47 -0400
Subject: [PATCH 1/3] Consolidate prefix logic
---
src/fastmcp/server/server.py | 136 ++++++++++------------
test_revert_check.py | 50 ++++++++
tests/deprecated/test_deprecated.py | 80 -------------
tests/deprecated/test_mount_separators.py | 44 +++----
tests/server/test_server.py | 18 ++-
5 files changed, 144 insertions(+), 184 deletions(-)
create mode 100644 test_revert_check.py
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index 09b4815b6..ef50374a6 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -12,6 +12,7 @@ from contextlib import (
AsyncExitStack,
asynccontextmanager,
)
+from dataclasses import dataclass
from functools import partial
from pathlib import Path
from typing import TYPE_CHECKING, Any, Generic, Literal, overload
@@ -322,10 +323,14 @@ class FastMCP(Generic[LifespanResultT]):
"""Get all registered tools, indexed by registered key."""
if (tools := self._cache.get("tools")) is self._cache.NOT_FOUND:
tools: dict[str, Tool] = {}
- for prefix, server in self._mounted_servers.items():
+ for prefix, mounted_server in self._mounted_servers.items():
try:
- server_tools = await server.get_tools()
- tools.update(server_tools)
+ server_tools = await mounted_server.server.get_tools()
+ # Apply prefix to each tool key
+ prefixed_tools = {
+ f"{prefix}_{key}": tool for key, tool in server_tools.items()
+ }
+ tools.update(prefixed_tools)
except Exception as e:
logger.warning(
f"Failed to get tools from mounted server '{prefix}': {e}"
@@ -345,10 +350,17 @@ class FastMCP(Generic[LifespanResultT]):
"""Get all registered resources, indexed by registered key."""
if (resources := self._cache.get("resources")) is self._cache.NOT_FOUND:
resources: dict[str, Resource] = {}
- for prefix, server in self._mounted_servers.items():
+ for prefix, mounted_server in self._mounted_servers.items():
try:
- server_resources = await server.get_resources()
- resources.update(server_resources)
+ server_resources = await mounted_server.server.get_resources()
+ # Apply prefix to each resource key
+ prefixed_resources = {
+ add_resource_prefix(
+ key, prefix, mounted_server.server.resource_prefix_format
+ ): resource
+ for key, resource in server_resources.items()
+ }
+ resources.update(prefixed_resources)
except Exception as e:
logger.warning(
f"Failed to get resources from mounted server '{prefix}': {e}"
@@ -370,10 +382,19 @@ class FastMCP(Generic[LifespanResultT]):
templates := self._cache.get("resource_templates")
) is self._cache.NOT_FOUND:
templates: dict[str, ResourceTemplate] = {}
- for prefix, server in self._mounted_servers.items():
+ for prefix, mounted_server in self._mounted_servers.items():
try:
- server_templates = await server.get_resource_templates()
- templates.update(server_templates)
+ server_templates = (
+ await mounted_server.server.get_resource_templates()
+ )
+ # Apply prefix to each template key
+ prefixed_templates = {
+ add_resource_prefix(
+ key, prefix, mounted_server.server.resource_prefix_format
+ ): template
+ for key, template in server_templates.items()
+ }
+ templates.update(prefixed_templates)
except Exception as e:
logger.warning(
"Failed to get resource templates from mounted server "
@@ -396,10 +417,15 @@ class FastMCP(Generic[LifespanResultT]):
"""
if (prompts := self._cache.get("prompts")) is self._cache.NOT_FOUND:
prompts: dict[str, Prompt] = {}
- for prefix, server in self._mounted_servers.items():
+ for prefix, mounted_server in self._mounted_servers.items():
try:
- server_prompts = await server.get_prompts()
- prompts.update(server_prompts)
+ server_prompts = await mounted_server.server.get_prompts()
+ # Apply prefix to each prompt key
+ prefixed_prompts = {
+ f"{prefix}_{key}": prompt
+ for key, prompt in server_prompts.items()
+ }
+ prompts.update(prefixed_prompts)
except Exception as e:
logger.warning(
f"Failed to get prompts from mounted server '{prefix}': {e}"
@@ -562,10 +588,10 @@ class FastMCP(Generic[LifespanResultT]):
return await self._tool_manager.call_tool(key, arguments)
# Check mounted servers to see if they have the tool
- for server in self._mounted_servers.values():
- if server.match_tool(key):
- tool_key = server.strip_tool_prefix(key)
- return await server.server._call_tool(tool_key, arguments)
+ for prefix, mounted_server in self._mounted_servers.items():
+ if key.startswith(f"{prefix}_"):
+ tool_key = key.removeprefix(f"{prefix}_")
+ return await mounted_server.server._call_tool(tool_key, arguments)
raise NotFoundError(f"Unknown tool: {key!r}")
@@ -604,10 +630,14 @@ class FastMCP(Generic[LifespanResultT]):
)
]
else:
- for server in self._mounted_servers.values():
- if server.match_resource(str(uri)):
- new_uri = server.strip_resource_prefix(str(uri))
- return await server.server._mcp_read_resource(new_uri)
+ for prefix, mounted_server in self._mounted_servers.items():
+ if has_resource_prefix(
+ str(uri), prefix, mounted_server.server.resource_prefix_format
+ ):
+ new_uri = remove_resource_prefix(
+ str(uri), prefix, mounted_server.server.resource_prefix_format
+ )
+ return await mounted_server.server._mcp_read_resource(new_uri)
else:
raise NotFoundError(f"Unknown resource: {uri}")
@@ -653,10 +683,12 @@ class FastMCP(Generic[LifespanResultT]):
return await self._prompt_manager.render_prompt(name, arguments)
# Check mounted servers to see if they have the prompt
- for server in self._mounted_servers.values():
- if server.match_prompt(name):
- prompt_name = server.strip_prompt_prefix(name)
- return await server.server._mcp_get_prompt(prompt_name, arguments)
+ for prefix, mounted_server in self._mounted_servers.items():
+ if name.startswith(f"{prefix}_"):
+ prompt_name = name.removeprefix(f"{prefix}_")
+ return await mounted_server.server._mcp_get_prompt(
+ prompt_name, arguments
+ )
raise NotFoundError(f"Unknown prompt: {name}")
@@ -1731,60 +1763,10 @@ class FastMCP(Generic[LifespanResultT]):
return True
+@dataclass
class MountedServer:
- def __init__(
- self,
- prefix: str,
- server: FastMCP[LifespanResultT],
- ):
- self.server = server
- self.prefix = prefix
-
- async def get_tools(self) -> dict[str, Tool]:
- tools = await self.server.get_tools()
- return {f"{self.prefix}_{key}": tool for key, tool in tools.items()}
-
- async def get_resources(self) -> dict[str, Resource]:
- resources = await self.server.get_resources()
- return {
- add_resource_prefix(
- key, self.prefix, self.server.resource_prefix_format
- ): resource
- for key, resource in resources.items()
- }
-
- async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
- templates = await self.server.get_resource_templates()
- return {
- add_resource_prefix(
- key, self.prefix, self.server.resource_prefix_format
- ): template
- for key, template in templates.items()
- }
-
- async def get_prompts(self) -> dict[str, Prompt]:
- prompts = await self.server.get_prompts()
- return {f"{self.prefix}_{key}": prompt for key, prompt in prompts.items()}
-
- def match_tool(self, key: str) -> bool:
- return key.startswith(f"{self.prefix}_")
-
- def strip_tool_prefix(self, key: str) -> str:
- return key.removeprefix(f"{self.prefix}_")
-
- def match_resource(self, key: str) -> bool:
- return has_resource_prefix(key, self.prefix, self.server.resource_prefix_format)
-
- def strip_resource_prefix(self, key: str) -> str:
- return remove_resource_prefix(
- key, self.prefix, self.server.resource_prefix_format
- )
-
- def match_prompt(self, key: str) -> bool:
- return key.startswith(f"{self.prefix}_")
-
- def strip_prompt_prefix(self, key: str) -> str:
- return key.removeprefix(f"{self.prefix}_")
+ prefix: str
+ server: FastMCP[Any]
def add_resource_prefix(
diff --git a/test_revert_check.py b/test_revert_check.py
new file mode 100644
index 000000000..a9e360e1e
--- /dev/null
+++ b/test_revert_check.py
@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+"""Quick test to verify the revert worked correctly."""
+
+import asyncio
+
+from fastmcp import FastMCP
+from fastmcp.client import Client
+
+
+async def test_empty_prefix_behavior():
+ """Test that empty prefix correctly adds underscore."""
+
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ @sub_app.tool
+ def sub_tool() -> str:
+ return "This is from the sub app"
+
+ @sub_app.resource("data://test")
+ def sub_resource():
+ return "Resource data"
+
+ # Mount with empty prefix
+ main_app.mount("", sub_app)
+
+ # Check that tools have underscore prefix
+ tools = await main_app.get_tools()
+ print(f"Tools: {list(tools.keys())}")
+ assert "_sub_tool" in tools, f"Expected '_sub_tool' in {list(tools.keys())}"
+
+ # Check that resources work correctly
+ resources = await main_app.get_resources()
+ print(f"Resources: {list(resources.keys())}")
+ # Empty prefix for resources should result in no prefix change
+ assert "data://test" in resources, (
+ f"Expected 'data://test' in {list(resources.keys())}"
+ )
+
+ # Test calling the tool
+ async with Client(main_app) as client:
+ result = await client.call_tool("_sub_tool", {})
+ print(f"Tool result: {result[0].text}")
+ assert "This is from the sub app" in result[0].text
+
+ print("✅ Empty prefix correctly adds underscore for tools!")
+
+
+if __name__ == "__main__":
+ asyncio.run(test_empty_prefix_behavior())
diff --git a/tests/deprecated/test_deprecated.py b/tests/deprecated/test_deprecated.py
index a20b970bb..f71161a98 100644
--- a/tests/deprecated/test_deprecated.py
+++ b/tests/deprecated/test_deprecated.py
@@ -109,83 +109,3 @@ def test_from_client_deprecation_warning():
server = FastMCP("TestServer")
with pytest.warns(DeprecationWarning, match="from_client"):
FastMCP.from_client(Client(server))
-
-
-def test_mount_tool_separator_deprecation_warning():
- """Test that using tool_separator in mount() raises a deprecation warning."""
- main_app = FastMCP("MainApp")
- sub_app = FastMCP("SubApp")
-
- with pytest.warns(
- DeprecationWarning,
- match="The tool_separator parameter is deprecated and will be removed in a future version",
- ):
- main_app.mount("sub", sub_app, tool_separator="-")
-
- # Verify the separator is ignored and the default is used
- @sub_app.tool
- def test_tool():
- return "test"
-
- mounted_server = main_app._mounted_servers["sub"]
- assert mounted_server.match_tool("sub_test_tool")
- assert not mounted_server.match_tool("sub-test_tool")
-
-
-def test_mount_resource_separator_deprecation_warning():
- """Test that using resource_separator in mount() raises a deprecation warning."""
- main_app = FastMCP("MainApp")
- sub_app = FastMCP("SubApp")
-
- with pytest.warns(
- DeprecationWarning,
- match="The resource_separator parameter is deprecated and ignored",
- ):
- main_app.mount("sub", sub_app, resource_separator="+")
-
-
-def test_mount_prompt_separator_deprecation_warning():
- """Test that using prompt_separator in mount() raises a deprecation warning."""
- main_app = FastMCP("MainApp")
- sub_app = FastMCP("SubApp")
-
- with pytest.warns(
- DeprecationWarning,
- match="The prompt_separator parameter is deprecated and will be removed in a future version",
- ):
- main_app.mount("sub", sub_app, prompt_separator="-")
-
- # Verify the separator is ignored and the default is used
- @sub_app.prompt
- def test_prompt():
- return "test"
-
- mounted_server = main_app._mounted_servers["sub"]
- assert mounted_server.match_prompt("sub_test_prompt")
- assert not mounted_server.match_prompt("sub-test_prompt")
-
-
-async def test_import_server_separator_deprecation_warnings():
- """Test that using separators in import_server() raises deprecation warnings."""
- main_app = FastMCP("MainApp")
- sub_app = FastMCP("SubApp")
-
- with pytest.warns(
- DeprecationWarning,
- match="The tool_separator parameter is deprecated and will be removed in a future version",
- ):
- await main_app.import_server("sub", sub_app, tool_separator="-")
-
- main_app = FastMCP("MainApp")
- with pytest.warns(
- DeprecationWarning,
- match="The resource_separator parameter is deprecated and ignored",
- ):
- await main_app.import_server("sub", sub_app, resource_separator="+")
-
- main_app = FastMCP("MainApp")
- with pytest.warns(
- DeprecationWarning,
- match="The prompt_separator parameter is deprecated and will be removed in a future version",
- ):
- await main_app.import_server("sub", sub_app, prompt_separator="-")
diff --git a/tests/deprecated/test_mount_separators.py b/tests/deprecated/test_mount_separators.py
index 85ec63436..cabefc963 100644
--- a/tests/deprecated/test_mount_separators.py
+++ b/tests/deprecated/test_mount_separators.py
@@ -1,14 +1,27 @@
"""Tests for the deprecated separator parameters in mount() and import_server() methods."""
import pytest
+from mcp import McpError
-from fastmcp import FastMCP
+from fastmcp import Client, FastMCP
# reset deprecation warnings for this module
pytestmark = pytest.mark.filterwarnings("default::DeprecationWarning")
-def test_mount_tool_separator_deprecation_warning():
+def test_mount_resource_separator_deprecation_warning():
+ """Test that using resource_separator in mount() raises a deprecation warning."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ with pytest.warns(
+ DeprecationWarning,
+ match="The resource_separator parameter is deprecated and ignored",
+ ):
+ main_app.mount("sub", sub_app, resource_separator="+")
+
+
+async def test_mount_tool_separator_deprecation_warning():
"""Test that using tool_separator in mount() raises a deprecation warning."""
main_app = FastMCP("MainApp")
sub_app = FastMCP("SubApp")
@@ -24,24 +37,12 @@ def test_mount_tool_separator_deprecation_warning():
def test_tool():
return "test"
- mounted_server = main_app._mounted_servers["sub"]
- assert mounted_server.match_tool("sub_test_tool")
- assert not mounted_server.match_tool("sub-test_tool")
+ async with Client(main_app) as client:
+ assert "sub_test_tool" in {t.name for t in await client.list_tools()}
+ assert "sub-test_tool" not in {t.name for t in await client.list_tools()}
-def test_mount_resource_separator_deprecation_warning():
- """Test that using resource_separator in mount() raises a deprecation warning."""
- main_app = FastMCP("MainApp")
- sub_app = FastMCP("SubApp")
-
- with pytest.warns(
- DeprecationWarning,
- match="The resource_separator parameter is deprecated and ignored",
- ):
- main_app.mount("sub", sub_app, resource_separator="+")
-
-
-def test_mount_prompt_separator_deprecation_warning():
+async def test_mount_prompt_separator_deprecation_warning():
"""Test that using prompt_separator in mount() raises a deprecation warning."""
main_app = FastMCP("MainApp")
sub_app = FastMCP("SubApp")
@@ -57,9 +58,10 @@ def test_mount_prompt_separator_deprecation_warning():
def test_prompt():
return "test"
- mounted_server = main_app._mounted_servers["sub"]
- assert mounted_server.match_prompt("sub_test_prompt")
- assert not mounted_server.match_prompt("sub-test_prompt")
+ async with Client(main_app) as client:
+ assert await client.get_prompt("sub_test_prompt")
+ with pytest.raises(McpError, match="Unknown prompt"):
+ await client.get_prompt("sub-test_prompt")
async def test_import_server_separator_deprecation_warnings():
diff --git a/tests/server/test_server.py b/tests/server/test_server.py
index dd4719a6e..70a2c174b 100644
--- a/tests/server/test_server.py
+++ b/tests/server/test_server.py
@@ -9,7 +9,6 @@ from fastmcp.exceptions import NotFoundError
from fastmcp.prompts.prompt import FunctionPrompt, Prompt
from fastmcp.resources import Resource, ResourceTemplate
from fastmcp.server.server import (
- MountedServer,
add_resource_prefix,
has_resource_prefix,
remove_resource_prefix,
@@ -1183,16 +1182,23 @@ class TestResourcePrefixMounting:
async def test_mounted_server_matching_and_stripping(
self, uri, prefix, expected_match, expected_strip
):
- """Test that MountedServer correctly matches and strips resource prefixes."""
- # Create a basic server to mount
+ """Test that resource prefix utility functions correctly match and strip resource prefixes."""
+ from fastmcp.server.server import has_resource_prefix, remove_resource_prefix
+
+ # Create a basic server to get the default resource prefix format
server = FastMCP()
- mounted = MountedServer(prefix=prefix, server=server)
# Test matching
- assert mounted.match_resource(uri) == expected_match
+ assert (
+ has_resource_prefix(uri, prefix, server.resource_prefix_format)
+ == expected_match
+ )
# Test stripping
- assert mounted.strip_resource_prefix(uri) == expected_strip
+ assert (
+ remove_resource_prefix(uri, prefix, server.resource_prefix_format)
+ == expected_strip
+ )
async def test_import_server_with_new_prefix_format(self):
"""Test that import_server correctly uses the new prefix format."""
From 8eaa6096b467b3e3d25c3ba535ab1f87f8d0c917 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Wed, 18 Jun 2025 11:28:24 -0400
Subject: [PATCH 2/3] Make prefixes optional
---
src/fastmcp/server/server.py | 265 +++++++---
.../deprecated/test_mount_import_arg_order.py | 275 +++++++++++
tests/deprecated/test_mount_separators.py | 83 ++--
tests/deprecated/test_resource_prefixes.py | 10 +-
tests/server/openapi/test_openapi.py | 22 -
tests/server/test_import_server.py | 267 ++++++++--
tests/server/test_mount.py | 465 +++++++++++++++---
tests/server/test_resource_prefix_formats.py | 8 +-
tests/server/test_server.py | 4 +-
9 files changed, 1149 insertions(+), 250 deletions(-)
create mode 100644 tests/deprecated/test_mount_import_arg_order.py
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index ef50374a6..47bd001eb 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -15,7 +15,7 @@ from contextlib import (
from dataclasses import dataclass
from functools import partial
from pathlib import Path
-from typing import TYPE_CHECKING, Any, Generic, Literal, overload
+from typing import TYPE_CHECKING, Any, Generic, Literal, cast, overload
import anyio
import httpx
@@ -155,7 +155,7 @@ class FastMCP(Generic[LifespanResultT]):
self._cache = TimedCache(
expiration=datetime.timedelta(seconds=cache_expiration_seconds or 0)
)
- self._mounted_servers: dict[str, MountedServer] = {}
+ self._mounted_servers: list[MountedServer] = []
self._additional_http_routes: list[BaseRoute] = []
self._tool_manager = ToolManager(
duplicate_behavior=on_duplicate_tools,
@@ -323,17 +323,21 @@ class FastMCP(Generic[LifespanResultT]):
"""Get all registered tools, indexed by registered key."""
if (tools := self._cache.get("tools")) is self._cache.NOT_FOUND:
tools: dict[str, Tool] = {}
- for prefix, mounted_server in self._mounted_servers.items():
+
+ # iterate such that new mounts overwrite older ones
+ for mounted_server in self._mounted_servers:
try:
server_tools = await mounted_server.server.get_tools()
- # Apply prefix to each tool key
- prefixed_tools = {
- f"{prefix}_{key}": tool for key, tool in server_tools.items()
- }
- tools.update(prefixed_tools)
+ # Apply prefix to each tool key if prefix exists and is not empty
+ if mounted_server.prefix:
+ server_tools = {
+ f"{mounted_server.prefix}_{key}": tool
+ for key, tool in server_tools.items()
+ }
+ tools.update(server_tools)
except Exception as e:
logger.warning(
- f"Failed to get tools from mounted server '{prefix}': {e}"
+ f"Failed to get tools from mounted server '{mounted_server.prefix}': {e}"
)
continue
tools.update(self._tool_manager.get_tools())
@@ -350,20 +354,25 @@ class FastMCP(Generic[LifespanResultT]):
"""Get all registered resources, indexed by registered key."""
if (resources := self._cache.get("resources")) is self._cache.NOT_FOUND:
resources: dict[str, Resource] = {}
- for prefix, mounted_server in self._mounted_servers.items():
+
+ # iterate such that new mounts overwrite older ones
+ for mounted_server in self._mounted_servers:
try:
server_resources = await mounted_server.server.get_resources()
- # Apply prefix to each resource key
- prefixed_resources = {
- add_resource_prefix(
- key, prefix, mounted_server.server.resource_prefix_format
- ): resource
- for key, resource in server_resources.items()
- }
- resources.update(prefixed_resources)
+ # Apply prefix to each resource key if prefix exists
+ if mounted_server.prefix:
+ server_resources = {
+ add_resource_prefix(
+ key,
+ mounted_server.prefix,
+ mounted_server.server.resource_prefix_format,
+ ): resource
+ for key, resource in server_resources.items()
+ }
+ resources.update(server_resources)
except Exception as e:
logger.warning(
- f"Failed to get resources from mounted server '{prefix}': {e}"
+ f"Failed to get resources from mounted server '{mounted_server.prefix}': {e}"
)
continue
resources.update(self._resource_manager.get_resources())
@@ -382,23 +391,28 @@ class FastMCP(Generic[LifespanResultT]):
templates := self._cache.get("resource_templates")
) is self._cache.NOT_FOUND:
templates: dict[str, ResourceTemplate] = {}
- for prefix, mounted_server in self._mounted_servers.items():
+
+ # iterate such that new mounts overwrite older ones
+ for mounted_server in self._mounted_servers:
try:
server_templates = (
await mounted_server.server.get_resource_templates()
)
- # Apply prefix to each template key
- prefixed_templates = {
- add_resource_prefix(
- key, prefix, mounted_server.server.resource_prefix_format
- ): template
- for key, template in server_templates.items()
- }
- templates.update(prefixed_templates)
+ # Apply prefix to each template key if prefix exists
+ if mounted_server.prefix:
+ server_templates = {
+ add_resource_prefix(
+ key,
+ mounted_server.prefix,
+ mounted_server.server.resource_prefix_format,
+ ): template
+ for key, template in server_templates.items()
+ }
+ templates.update(server_templates)
except Exception as e:
logger.warning(
"Failed to get resource templates from mounted server "
- f"'{prefix}': {e}"
+ f"'{mounted_server.prefix}': {e}"
)
continue
templates.update(self._resource_manager.get_templates())
@@ -417,18 +431,21 @@ class FastMCP(Generic[LifespanResultT]):
"""
if (prompts := self._cache.get("prompts")) is self._cache.NOT_FOUND:
prompts: dict[str, Prompt] = {}
- for prefix, mounted_server in self._mounted_servers.items():
+
+ # iterate such that new mounts overwrite older ones
+ for mounted_server in self._mounted_servers:
try:
server_prompts = await mounted_server.server.get_prompts()
- # Apply prefix to each prompt key
- prefixed_prompts = {
- f"{prefix}_{key}": prompt
- for key, prompt in server_prompts.items()
- }
- prompts.update(prefixed_prompts)
+ # Apply prefix to each prompt key if prefix exists
+ if mounted_server.prefix:
+ server_prompts = {
+ f"{mounted_server.prefix}_{key}": prompt
+ for key, prompt in server_prompts.items()
+ }
+ prompts.update(server_prompts)
except Exception as e:
logger.warning(
- f"Failed to get prompts from mounted server '{prefix}': {e}"
+ f"Failed to get prompts from mounted server '{mounted_server.prefix}': {e}"
)
continue
prompts.update(self._prompt_manager.get_prompts())
@@ -588,10 +605,20 @@ class FastMCP(Generic[LifespanResultT]):
return await self._tool_manager.call_tool(key, arguments)
# Check mounted servers to see if they have the tool
- for prefix, mounted_server in self._mounted_servers.items():
- if key.startswith(f"{prefix}_"):
- tool_key = key.removeprefix(f"{prefix}_")
+ # iterate such that new mounts take precedence over older ones
+ for mounted_server in reversed(self._mounted_servers):
+ tool_key = key
+ try:
+ # If server has a prefix, check if key matches and strip prefix
+ if mounted_server.prefix:
+ if tool_key.startswith(f"{mounted_server.prefix}_"):
+ tool_key = tool_key.removeprefix(f"{mounted_server.prefix}_")
+ else:
+ continue
return await mounted_server.server._call_tool(tool_key, arguments)
+ except NotFoundError:
+ # Tool not found on this server, try the next one
+ continue
raise NotFoundError(f"Unknown tool: {key!r}")
@@ -630,14 +657,28 @@ class FastMCP(Generic[LifespanResultT]):
)
]
else:
- for prefix, mounted_server in self._mounted_servers.items():
- if has_resource_prefix(
- str(uri), prefix, mounted_server.server.resource_prefix_format
- ):
- new_uri = remove_resource_prefix(
- str(uri), prefix, mounted_server.server.resource_prefix_format
- )
- return await mounted_server.server._mcp_read_resource(new_uri)
+ # iterate such that new mounts take precedence over older ones
+ for mounted_server in reversed(self._mounted_servers):
+ resource_uri = uri
+ try:
+ if mounted_server.prefix:
+ # If server has a prefix, check if URI matches and strip prefix
+ if has_resource_prefix(
+ str(resource_uri),
+ mounted_server.prefix,
+ mounted_server.server.resource_prefix_format,
+ ):
+ resource_uri = remove_resource_prefix(
+ str(resource_uri),
+ mounted_server.prefix,
+ mounted_server.server.resource_prefix_format,
+ )
+ else:
+ continue
+ return await mounted_server.server._mcp_read_resource(resource_uri)
+ except NotFoundError:
+ # Resource not found on this server, try the next one
+ continue
else:
raise NotFoundError(f"Unknown resource: {uri}")
@@ -683,12 +724,24 @@ class FastMCP(Generic[LifespanResultT]):
return await self._prompt_manager.render_prompt(name, arguments)
# Check mounted servers to see if they have the prompt
- for prefix, mounted_server in self._mounted_servers.items():
- if name.startswith(f"{prefix}_"):
- prompt_name = name.removeprefix(f"{prefix}_")
+ # iterate such that new mounts take precedence over older ones
+ for mounted_server in reversed(self._mounted_servers):
+ prompt_name = name
+ try:
+ if mounted_server.prefix:
+ # If server has a prefix, check if name matches and strip prefix
+ if prompt_name.startswith(f"{mounted_server.prefix}_"):
+ prompt_name = prompt_name.removeprefix(
+ f"{mounted_server.prefix}_"
+ )
+ else:
+ continue
return await mounted_server.server._mcp_get_prompt(
prompt_name, arguments
)
+ except NotFoundError:
+ # Prompt not found on this server, try the next one
+ continue
raise NotFoundError(f"Unknown prompt: {name}")
@@ -1412,15 +1465,15 @@ class FastMCP(Generic[LifespanResultT]):
def mount(
self,
- prefix: str,
server: FastMCP[LifespanResultT],
+ prefix: str | None = None,
as_proxy: bool | None = None,
*,
tool_separator: str | None = None,
resource_separator: str | None = None,
prompt_separator: str | None = None,
) -> None:
- """Mount another FastMCP server on this server with the given prefix.
+ """Mount another FastMCP server on this server with an optional prefix.
Unlike importing (with import_server), mounting establishes a dynamic connection
between servers. When a client interacts with a mounted server's objects through
@@ -1428,7 +1481,7 @@ class FastMCP(Generic[LifespanResultT]):
This means changes to the mounted server are immediately reflected when accessed
through the parent.
- When a server is mounted:
+ When a server is mounted with a prefix:
- Tools from the mounted server are accessible with prefixed names.
Example: If server has a tool named "get_weather", it will be available as "prefix_get_weather".
- Resources are accessible with prefixed URIs.
@@ -1441,6 +1494,10 @@ class FastMCP(Generic[LifespanResultT]):
Example: If server has a prompt named "weather_prompt", it will be available as
"prefix_weather_prompt".
+ When a server is mounted without a prefix (prefix=None), its tools, resources, templates,
+ and prompts are accessible with their original names. Multiple servers can be mounted
+ without prefixes, and they will be tried in order until a match is found.
+
There are two modes for mounting servers:
1. Direct mounting (default when server has no custom lifespan): The parent server
directly accesses the mounted server's objects in-memory for better performance.
@@ -1453,8 +1510,9 @@ class FastMCP(Generic[LifespanResultT]):
execution, but with slightly higher overhead.
Args:
- prefix: Prefix to use for the mounted server's objects.
server: The FastMCP server to mount.
+ prefix: Optional prefix to use for the mounted server's objects. If None,
+ the server's objects are accessible with their original names.
as_proxy: Whether to treat the mounted server as a proxy. If None (default),
automatically determined based on whether the server has a custom lifespan
(True if it has a custom lifespan, False otherwise).
@@ -1466,6 +1524,20 @@ class FastMCP(Generic[LifespanResultT]):
from fastmcp.client.transports import FastMCPTransport
from fastmcp.server.proxy import FastMCPProxy
+ # Deprecated since 2.9.0
+ # Prior to 2.9.0, the first positional argument was the prefix and the
+ # second was the server. Here we swap them if needed now that the prefix
+ # is optional.
+ if isinstance(server, str):
+ if fastmcp.settings.deprecation_warnings:
+ warnings.warn(
+ "Mount prefixes are now optional and the first positional argument "
+ "should be the server you want to mount.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+ server, prefix = cast(FastMCP[Any], prefix), server
+
if tool_separator is not None:
# Deprecated since 2.4.0
if fastmcp.settings.deprecation_warnings:
@@ -1508,17 +1580,13 @@ class FastMCP(Generic[LifespanResultT]):
server=server,
prefix=prefix,
)
- self._mounted_servers[prefix] = mounted_server
- self._cache.clear()
-
- def unmount(self, prefix: str) -> None:
- self._mounted_servers.pop(prefix)
+ self._mounted_servers.append(mounted_server)
self._cache.clear()
async def import_server(
self,
- prefix: str,
server: FastMCP[LifespanResultT],
+ prefix: str | None = None,
tool_separator: str | None = None,
resource_separator: str | None = None,
prompt_separator: str | None = None,
@@ -1532,7 +1600,7 @@ class FastMCP(Generic[LifespanResultT]):
future changes to the imported server will not be reflected in the
importing server. Server-level configurations and lifespans are not imported.
- When a server is imported:
+ When a server is imported with a prefix:
- The tools are imported with prefixed names
Example: If server has a tool named "get_weather", it will be
available as "prefix_get_weather"
@@ -1546,14 +1614,33 @@ class FastMCP(Generic[LifespanResultT]):
Example: If server has a prompt named "weather_prompt", it will be available as
"prefix_weather_prompt"
+ When a server is imported without a prefix (prefix=None), its tools, resources,
+ templates, and prompts are imported with their original names.
+
Args:
- prefix: The prefix to use for the imported server
server: The FastMCP server to import
+ prefix: Optional prefix to use for the imported server's objects. If None,
+ objects are imported with their original names.
tool_separator: Deprecated. Separator for tool names.
resource_separator: Deprecated and ignored. Prefix is now
applied using the protocol://prefix/path format
prompt_separator: Deprecated. Separator for prompt names.
"""
+
+ # Deprecated since 2.9.0
+ # Prior to 2.9.0, the first positional argument was the prefix and the
+ # second was the server. Here we swap them if needed now that the prefix
+ # is optional.
+ if isinstance(server, str):
+ if fastmcp.settings.deprecation_warnings:
+ warnings.warn(
+ "Import prefixes are now optional and the first positional argument "
+ "should be the server you want to import.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+ server, prefix = cast(FastMCP[Any], prefix), server
+
if tool_separator is not None:
# Deprecated since 2.4.0
if fastmcp.settings.deprecation_warnings:
@@ -1584,29 +1671,49 @@ class FastMCP(Generic[LifespanResultT]):
stacklevel=2,
)
- # Import tools from the mounted server
- tool_prefix = f"{prefix}_"
+ # Import tools from the server
for key, tool in (await server.get_tools()).items():
- self._tool_manager.add_tool(tool, key=f"{tool_prefix}{key}")
+ if prefix:
+ tool_key = f"{prefix}_{key}"
+ else:
+ tool_key = key
+ self._tool_manager.add_tool(tool, key=tool_key)
- # Import resources and templates from the mounted server
+ # Import resources and templates from the server
for key, resource in (await server.get_resources()).items():
- prefixed_key = add_resource_prefix(key, prefix, self.resource_prefix_format)
- self._resource_manager.add_resource(resource, key=prefixed_key)
+ if prefix:
+ resource_key = add_resource_prefix(
+ key, prefix, self.resource_prefix_format
+ )
+ else:
+ resource_key = key
+ self._resource_manager.add_resource(resource, key=resource_key)
for key, template in (await server.get_resource_templates()).items():
- prefixed_key = add_resource_prefix(key, prefix, self.resource_prefix_format)
- self._resource_manager.add_template(template, key=prefixed_key)
+ if prefix:
+ template_key = add_resource_prefix(
+ key, prefix, self.resource_prefix_format
+ )
+ else:
+ template_key = key
+ self._resource_manager.add_template(template, key=template_key)
- # Import prompts from the mounted server
- prompt_prefix = f"{prefix}_"
+ # Import prompts from the server
for key, prompt in (await server.get_prompts()).items():
- self._prompt_manager.add_prompt(prompt, key=f"{prompt_prefix}{key}")
+ if prefix:
+ prompt_key = f"{prefix}_{key}"
+ else:
+ prompt_key = key
+ self._prompt_manager.add_prompt(prompt, key=prompt_key)
- logger.info(f"Imported server {server.name} with prefix '{prefix}'")
- logger.debug(f"Imported tools with prefix '{tool_prefix}'")
- logger.debug(f"Imported resources and templates with prefix '{prefix}/'")
- logger.debug(f"Imported prompts with prefix '{prompt_prefix}'")
+ if prefix:
+ logger.info(f"Imported server {server.name} with prefix '{prefix}'")
+ logger.debug(f"Imported tools with prefix '{prefix}_'")
+ logger.debug(f"Imported resources and templates with prefix '{prefix}/'")
+ logger.debug(f"Imported prompts with prefix '{prefix}_'")
+ else:
+ logger.info(f"Imported server {server.name}")
+ logger.debug("Imported tools, resources, templates, and prompts")
self._cache.clear()
@@ -1765,7 +1872,7 @@ class FastMCP(Generic[LifespanResultT]):
@dataclass
class MountedServer:
- prefix: str
+ prefix: str | None
server: FastMCP[Any]
diff --git a/tests/deprecated/test_mount_import_arg_order.py b/tests/deprecated/test_mount_import_arg_order.py
new file mode 100644
index 000000000..7fc273b36
--- /dev/null
+++ b/tests/deprecated/test_mount_import_arg_order.py
@@ -0,0 +1,275 @@
+import warnings
+
+from fastmcp import FastMCP
+from fastmcp.client import Client
+
+
+class TestDeprecatedMountArgOrder:
+ """Test deprecated positional argument order for mount() method."""
+
+ async def test_mount_deprecated_arg_order_with_warning(self):
+ """Test that mount(prefix, server) still works but raises deprecation warning."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ @sub_app.tool
+ def sub_tool() -> str:
+ return "Sub tool result"
+
+ # Test the deprecated argument order: mount(prefix, server)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ main_app.mount("sub", sub_app) # type: ignore[arg-type] # Old order: prefix first, server second
+
+ # Check that a deprecation warning was raised
+ assert len(w) == 1
+ assert issubclass(w[0].category, DeprecationWarning)
+ assert (
+ "Mount prefixes are now optional and the first positional argument should be the server"
+ in str(w[0].message)
+ )
+
+ # Verify the mount worked correctly despite deprecated order
+ tools = await main_app.get_tools()
+ assert "sub_sub_tool" in tools
+
+ # Test functionality
+ async with Client(main_app) as client:
+ result = await client.call_tool("sub_sub_tool", {})
+ assert result[0].text == "Sub tool result" # type: ignore[attr-defined]
+
+ async def test_mount_new_arg_order_no_warning(self):
+ """Test that mount(server, prefix) works without deprecation warning."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ @sub_app.tool
+ def sub_tool() -> str:
+ return "Sub tool result"
+
+ # Test the new argument order: mount(server, prefix)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ main_app.mount(sub_app, "sub") # New order: server first, prefix second
+
+ # Check that no deprecation warning was raised for argument order
+ mount_warnings = [
+ warning
+ for warning in w
+ if "Mount prefixes are now optional" in str(warning.message)
+ ]
+ assert len(mount_warnings) == 0
+
+ # Verify the mount worked correctly
+ tools = await main_app.get_tools()
+ assert "sub_sub_tool" in tools
+
+ async def test_mount_deprecated_order_no_prefix(self):
+ """Test deprecated order detection when first arg is empty string."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ @sub_app.tool
+ def sub_tool() -> str:
+ return "Sub tool result"
+
+ # Test with empty string as first argument (old style for no prefix)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ main_app.mount("", sub_app) # type: ignore[arg-type] # Old order: empty prefix first, server second
+
+ # Check that a deprecation warning was raised
+ assert len(w) == 1
+ assert issubclass(w[0].category, DeprecationWarning)
+ assert (
+ "Mount prefixes are now optional and the first positional argument should be the server"
+ in str(w[0].message)
+ )
+
+ # Verify the mount worked correctly (no prefix)
+ tools = await main_app.get_tools()
+ assert "sub_tool" in tools # No prefix applied
+
+
+class TestDeprecatedImportArgOrder:
+ """Test deprecated positional argument order for import_server() method."""
+
+ async def test_import_deprecated_arg_order_with_warning(self):
+ """Test that import_server(prefix, server) still works but raises deprecation warning."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ @sub_app.tool
+ def sub_tool() -> str:
+ return "Sub tool result"
+
+ # Test the deprecated argument order: import_server(prefix, server)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ await main_app.import_server("sub", sub_app) # type: ignore[arg-type] # Old order: prefix first, server second
+
+ # Check that a deprecation warning was raised
+ assert len(w) == 1
+ assert issubclass(w[0].category, DeprecationWarning)
+ assert (
+ "Import prefixes are now optional and the first positional argument should be the server"
+ in str(w[0].message)
+ )
+
+ # Verify the import worked correctly despite deprecated order
+ assert "sub_sub_tool" in main_app._tool_manager._tools
+
+ # Test functionality
+ async with Client(main_app) as client:
+ result = await client.call_tool("sub_sub_tool", {})
+ assert result[0].text == "Sub tool result" # type: ignore[attr-defined]
+
+ async def test_import_new_arg_order_no_warning(self):
+ """Test that import_server(server, prefix) works without deprecation warning."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ @sub_app.tool
+ def sub_tool() -> str:
+ return "Sub tool result"
+
+ # Test the new argument order: import_server(server, prefix)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ await main_app.import_server(
+ sub_app, "sub"
+ ) # New order: server first, prefix second
+
+ # Check that no deprecation warning was raised for argument order
+ import_warnings = [
+ warning
+ for warning in w
+ if "Import prefixes are now optional" in str(warning.message)
+ ]
+ assert len(import_warnings) == 0
+
+ # Verify the import worked correctly
+ assert "sub_sub_tool" in main_app._tool_manager._tools
+
+ async def test_import_deprecated_order_no_prefix(self):
+ """Test deprecated order detection when first arg is empty string."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ @sub_app.tool
+ def sub_tool() -> str:
+ return "Sub tool result"
+
+ # Test with empty string as first argument (old style for no prefix)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ await main_app.import_server("", sub_app) # type: ignore[arg-type] # Old order: empty prefix first, server second
+
+ # Check that a deprecation warning was raised
+ assert len(w) == 1
+ assert issubclass(w[0].category, DeprecationWarning)
+ assert (
+ "Import prefixes are now optional and the first positional argument should be the server"
+ in str(w[0].message)
+ )
+
+ # Verify the import worked correctly (no prefix)
+ assert "sub_tool" in main_app._tool_manager._tools # No prefix applied
+
+ async def test_import_deprecated_order_with_resources_and_prompts(self):
+ """Test deprecated order works with all component types."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ @sub_app.tool
+ def sub_tool() -> str:
+ return "Sub tool result"
+
+ @sub_app.resource(uri="data://config")
+ def sub_resource():
+ return "Sub resource data"
+
+ @sub_app.resource(uri="users://{user_id}/info")
+ def sub_template(user_id: str):
+ return f"Sub template for user {user_id}"
+
+ @sub_app.prompt
+ def sub_prompt() -> str:
+ return "Sub prompt content"
+
+ # Test the deprecated argument order with all component types
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ await main_app.import_server("api", sub_app) # type: ignore[arg-type] # Old order: prefix first, server second
+
+ # Check that a deprecation warning was raised
+ assert len(w) == 1
+ assert issubclass(w[0].category, DeprecationWarning)
+
+ # Verify all component types were imported correctly with prefix
+ assert "api_sub_tool" in main_app._tool_manager._tools
+ assert "data://api/config" in main_app._resource_manager._resources
+ assert "users://api/{user_id}/info" in main_app._resource_manager._templates
+ assert "api_sub_prompt" in main_app._prompt_manager._prompts
+
+
+class TestArgOrderDetection:
+ """Test that argument order detection works correctly."""
+
+ async def test_mount_correctly_identifies_server_vs_string(self):
+ """Test that mount correctly identifies FastMCP instances vs strings."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ # This should NOT trigger deprecation warning (server first, prefix second)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ main_app.mount(sub_app, "prefix")
+
+ mount_warnings = [
+ warning
+ for warning in w
+ if "Mount prefixes are now optional" in str(warning.message)
+ ]
+ assert len(mount_warnings) == 0
+
+ # This SHOULD trigger deprecation warning (string first, server second)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ main_app.mount("prefix2", sub_app) # type: ignore[arg-type]
+
+ mount_warnings = [
+ warning
+ for warning in w
+ if "Mount prefixes are now optional" in str(warning.message)
+ ]
+ assert len(mount_warnings) == 1
+
+ async def test_import_correctly_identifies_server_vs_string(self):
+ """Test that import_server correctly identifies FastMCP instances vs strings."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ # This should NOT trigger deprecation warning (server first, prefix second)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ await main_app.import_server(sub_app, "prefix")
+
+ import_warnings = [
+ warning
+ for warning in w
+ if "Import prefixes are now optional" in str(warning.message)
+ ]
+ assert len(import_warnings) == 0
+
+ # This SHOULD trigger deprecation warning (string first, server second)
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ await main_app.import_server("prefix2", sub_app) # type: ignore[arg-type]
+
+ import_warnings = [
+ warning
+ for warning in w
+ if "Import prefixes are now optional" in str(warning.message)
+ ]
+ assert len(import_warnings) == 1
diff --git a/tests/deprecated/test_mount_separators.py b/tests/deprecated/test_mount_separators.py
index cabefc963..a6b476a77 100644
--- a/tests/deprecated/test_mount_separators.py
+++ b/tests/deprecated/test_mount_separators.py
@@ -14,11 +14,16 @@ def test_mount_resource_separator_deprecation_warning():
main_app = FastMCP("MainApp")
sub_app = FastMCP("SubApp")
- with pytest.warns(
- DeprecationWarning,
- match="The resource_separator parameter is deprecated and ignored",
- ):
- main_app.mount("sub", sub_app, resource_separator="+")
+ with pytest.warns(DeprecationWarning) as warnings:
+ main_app.mount("sub", sub_app, resource_separator="+") # type: ignore[arg-type]
+
+ # Check that we get both the argument order warning and the resource_separator warning
+ warning_messages = [str(w.message) for w in warnings]
+ assert any(
+ "resource_separator parameter is deprecated and ignored" in msg
+ for msg in warning_messages
+ )
+ assert any("Mount prefixes are now optional" in msg for msg in warning_messages)
async def test_mount_tool_separator_deprecation_warning():
@@ -26,11 +31,15 @@ async def test_mount_tool_separator_deprecation_warning():
main_app = FastMCP("MainApp")
sub_app = FastMCP("SubApp")
- with pytest.warns(
- DeprecationWarning,
- match="The tool_separator parameter is deprecated and will be removed in a future version",
- ):
- main_app.mount("sub", sub_app, tool_separator="-")
+ with pytest.warns(DeprecationWarning) as warnings:
+ main_app.mount("sub", sub_app, tool_separator="-") # type: ignore[arg-type]
+
+ # Check that we get both the argument order warning and the tool_separator warning
+ warning_messages = [str(w.message) for w in warnings]
+ assert any(
+ "tool_separator parameter is deprecated" in msg for msg in warning_messages
+ )
+ assert any("Mount prefixes are now optional" in msg for msg in warning_messages)
# Verify the separator is ignored and the default is used
@sub_app.tool
@@ -47,11 +56,15 @@ async def test_mount_prompt_separator_deprecation_warning():
main_app = FastMCP("MainApp")
sub_app = FastMCP("SubApp")
- with pytest.warns(
- DeprecationWarning,
- match="The prompt_separator parameter is deprecated and will be removed in a future version",
- ):
- main_app.mount("sub", sub_app, prompt_separator="-")
+ with pytest.warns(DeprecationWarning) as warnings:
+ main_app.mount("sub", sub_app, prompt_separator="-") # type: ignore[arg-type]
+
+ # Check that we get both the argument order warning and the prompt_separator warning
+ warning_messages = [str(w.message) for w in warnings]
+ assert any(
+ "prompt_separator parameter is deprecated" in msg for msg in warning_messages
+ )
+ assert any("Mount prefixes are now optional" in msg for msg in warning_messages)
# Verify the separator is ignored and the default is used
@sub_app.prompt
@@ -69,22 +82,32 @@ async def test_import_server_separator_deprecation_warnings():
main_app = FastMCP("MainApp")
sub_app = FastMCP("SubApp")
- with pytest.warns(
- DeprecationWarning,
- match="The tool_separator parameter is deprecated and will be removed in a future version",
- ):
- await main_app.import_server("sub", sub_app, tool_separator="-")
+ with pytest.warns(DeprecationWarning) as warnings:
+ await main_app.import_server("sub", sub_app, tool_separator="-") # type: ignore[arg-type]
+
+ # Check that we get both warnings
+ warning_messages = [str(w.message) for w in warnings]
+ assert any(
+ "tool_separator parameter is deprecated" in msg for msg in warning_messages
+ )
+ assert any("Import prefixes are now optional" in msg for msg in warning_messages)
main_app = FastMCP("MainApp")
- with pytest.warns(
- DeprecationWarning,
- match="The resource_separator parameter is deprecated and ignored",
- ):
- await main_app.import_server("sub", sub_app, resource_separator="+")
+ with pytest.warns(DeprecationWarning) as warnings:
+ await main_app.import_server("sub", sub_app, resource_separator="+") # type: ignore[arg-type]
+
+ warning_messages = [str(w.message) for w in warnings]
+ assert any(
+ "resource_separator parameter is deprecated" in msg for msg in warning_messages
+ )
+ assert any("Import prefixes are now optional" in msg for msg in warning_messages)
main_app = FastMCP("MainApp")
- with pytest.warns(
- DeprecationWarning,
- match="The prompt_separator parameter is deprecated and will be removed in a future version",
- ):
- await main_app.import_server("sub", sub_app, prompt_separator="-")
+ with pytest.warns(DeprecationWarning) as warnings:
+ await main_app.import_server("sub", sub_app, prompt_separator="-") # type: ignore[arg-type]
+
+ warning_messages = [str(w.message) for w in warnings]
+ assert any(
+ "prompt_separator parameter is deprecated" in msg for msg in warning_messages
+ )
+ assert any("Import prefixes are now optional" in msg for msg in warning_messages)
diff --git a/tests/deprecated/test_resource_prefixes.py b/tests/deprecated/test_resource_prefixes.py
index 8775f400e..85ffdcf7d 100644
--- a/tests/deprecated/test_resource_prefixes.py
+++ b/tests/deprecated/test_resource_prefixes.py
@@ -67,8 +67,9 @@ async def test_mount_with_legacy_prefixes():
def get_test():
return "test content"
- # Mount the server with a prefix
- main_server.mount("sub", sub_server)
+ # Mount the server with a prefix (using old argument order for this legacy test)
+ with pytest.warns(DeprecationWarning, match="Mount prefixes are now optional"):
+ main_server.mount("sub", sub_server) # type: ignore[arg-type]
# Check that the resource is prefixed using the legacy format
resources = await main_server.get_resources()
@@ -93,8 +94,9 @@ async def test_import_server_with_legacy_prefixes():
def get_test():
return "test content"
- # Import the server with a prefix
- await main_server.import_server("sub", sub_server)
+ # Import the server with a prefix (using old argument order for this legacy test)
+ with pytest.warns(DeprecationWarning, match="Import prefixes are now optional"):
+ await main_server.import_server("sub", sub_server) # type: ignore[arg-type]
# Check that the resource is prefixed using the legacy format
resources = main_server._resource_manager.get_resources()
diff --git a/tests/server/openapi/test_openapi.py b/tests/server/openapi/test_openapi.py
index 366416975..c7a0b74f6 100644
--- a/tests/server/openapi/test_openapi.py
+++ b/tests/server/openapi/test_openapi.py
@@ -911,28 +911,6 @@ class TestOpenAPI31Compatibility:
assert order["items"] == ["item4", "item5"]
-class TestMountFastMCP:
- """Tests for mounting FastMCP servers."""
-
- async def test_mount_fastmcp(
- self, fastmcp_openapi_server_with_all_types: FastMCPOpenAPI
- ):
- """Test mounting an OpenAPI server."""
- mcp = FastMCP("MainApp")
-
- await mcp.import_server("fastapi", fastmcp_openapi_server_with_all_types)
-
- # Check that resources are available with prefixed URIs
- async with Client(mcp) as client:
- resources = await client.list_resources()
- assert len(resources) == 4 # Updated to account for new search endpoint
- # We're checking the key used by mcp to store the resource
- # The prefixed URI is used as the key, but the resource's original uri is preserved
- prefixed_uri = "resource://fastapi/get_users_users_get"
- resource = mcp._resource_manager.get_resources().get(prefixed_uri)
- assert resource is not None
-
-
async def test_empty_query_parameters_not_sent(
fastapi_app: FastAPI, api_client: httpx.AsyncClient
):
diff --git a/tests/server/test_import_server.py b/tests/server/test_import_server.py
index 581940561..3db05da93 100644
--- a/tests/server/test_import_server.py
+++ b/tests/server/test_import_server.py
@@ -18,7 +18,7 @@ async def test_import_basic_functionality():
return "This is from the sub app"
# Import the sub-app to the main app
- await main_app.import_server("sub", sub_app)
+ await main_app.import_server(sub_app, "sub")
# Verify the tool was imported with the prefix
assert "sub_sub_tool" in main_app._tool_manager._tools
@@ -49,8 +49,8 @@ async def test_import_multiple_apps():
return "News headlines"
# Import both sub-apps to the main app
- await main_app.import_server("weather", weather_app)
- await main_app.import_server("news", news_app)
+ await main_app.import_server(weather_app, "weather")
+ await main_app.import_server(news_app, "news")
# Verify tools were imported with the correct prefixes
assert "weather_get_forecast" in main_app._tool_manager._tools
@@ -74,11 +74,11 @@ async def test_import_combines_tools():
return "Second app tool"
# Import first app
- await main_app.import_server("api", first_app)
+ await main_app.import_server(first_app, "api")
assert "api_first_tool" in main_app._tool_manager._tools
# Import second app to same prefix
- await main_app.import_server("api", second_app)
+ await main_app.import_server(second_app, "api")
# Verify second tool is there
assert "api_second_tool" in main_app._tool_manager._tools
@@ -99,7 +99,7 @@ async def test_import_with_resources():
return ["user1", "user2"]
# Import the data app
- await main_app.import_server("data", data_app)
+ await main_app.import_server(data_app, "data")
# Verify the resource was imported with the prefix
assert "data://data/users" in main_app._resource_manager._resources
@@ -117,7 +117,7 @@ async def test_import_with_resource_templates():
return {"id": user_id, "name": f"User {user_id}"}
# Import the user app
- await main_app.import_server("api", user_app)
+ await main_app.import_server(user_app, "api")
# Verify the template was imported with the prefix
assert "users://api/{user_id}/profile" in main_app._resource_manager._templates
@@ -135,7 +135,7 @@ async def test_import_with_prompts():
return f"Hello, {name}!"
# Import the assistant app
- await main_app.import_server("assistant", assistant_app)
+ await main_app.import_server(assistant_app, "assistant")
# Verify the prompt was imported with the prefix
assert "assistant_greeting" in main_app._prompt_manager._prompts
@@ -158,8 +158,8 @@ async def test_import_multiple_resource_templates():
return f"News for {category}"
# Import both apps
- await main_app.import_server("data", weather_app)
- await main_app.import_server("content", news_app)
+ await main_app.import_server(weather_app, "data")
+ await main_app.import_server(news_app, "content")
# Verify templates were imported with correct prefixes
assert "weather://data/{city}" in main_app._resource_manager._templates
@@ -183,8 +183,8 @@ async def test_import_multiple_prompts():
return f"Explaining SQL query:\n{query}"
# Import both apps
- await main_app.import_server("python", python_app)
- await main_app.import_server("sql", sql_app)
+ await main_app.import_server(python_app, "python")
+ await main_app.import_server(sql_app, "sql")
# Verify prompts were imported with correct prefixes
assert "python_review_python" in main_app._prompt_manager._prompts
@@ -200,7 +200,7 @@ async def test_tool_custom_name_preserved_when_imported():
return f"Data for query: {query}"
api_app.add_tool(Tool.from_function(fetch_data, name="get_data"))
- await main_app.import_server("api", api_app)
+ await main_app.import_server(api_app, "api")
# Check that the tool is accessible by its prefixed name
tool = main_app._tool_manager.get_tool("api_get_data")
@@ -220,7 +220,7 @@ async def test_call_imported_custom_named_tool():
return f"Data for query: {query}"
api_app.add_tool(Tool.from_function(fetch_data, name="get_data"))
- await main_app.import_server("api", api_app)
+ await main_app.import_server(api_app, "api")
async with Client(main_app) as client:
result = await client.call_tool("api_get_data", {"query": "test"})
@@ -236,7 +236,7 @@ async def test_first_level_importing_with_custom_name():
return input * 2
provider_app.add_tool(Tool.from_function(calculate_value, name="compute"))
- await service_app.import_server("provider", provider_app)
+ await service_app.import_server(provider_app, "provider")
# Tool is accessible in the service app with the first prefix
tool = service_app._tool_manager.get_tool("provider_compute")
@@ -255,8 +255,8 @@ async def test_nested_importing_preserves_prefixes():
return input * 2
provider_app.add_tool(Tool.from_function(calculate_value, name="compute"))
- await service_app.import_server("provider", provider_app)
- await main_app.import_server("service", service_app)
+ await service_app.import_server(provider_app, "provider")
+ await main_app.import_server(service_app, "service")
# Tool is accessible in the main app with both prefixes
tool = main_app._tool_manager.get_tool("service_provider_compute")
@@ -273,13 +273,12 @@ async def test_call_nested_imported_tool():
return input * 2
provider_app.add_tool(Tool.from_function(calculate_value, name="compute"))
- await service_app.import_server("provider", provider_app)
- await main_app.import_server("service", service_app)
+ await service_app.import_server(provider_app, "provider")
+ await main_app.import_server(service_app, "service")
- result = await main_app._tool_manager.call_tool(
- "service_provider_compute", {"input": 21}
- )
- assert result[0].text == "42" # type: ignore[attr-defined]
+ async with Client(main_app) as client:
+ result = await client.call_tool("service_provider_compute", {"input": 21})
+ assert result[0].text == "42" # type: ignore[attr-defined]
async def test_import_with_proxy_tools():
@@ -299,10 +298,11 @@ async def test_import_with_proxy_tools():
return f"Data for query: {query}"
proxy_app = FastMCP.as_proxy(Client(api_app))
- await main_app.import_server("api", proxy_app)
+ await main_app.import_server(proxy_app, "api")
- result = await main_app._mcp_call_tool("api_get_data", {"query": "test"})
- assert result[0].text == "Data for query: test" # type: ignore[attr-defined]
+ async with Client(main_app) as client:
+ result = await client.call_tool("api_get_data", {"query": "test"})
+ assert result[0].text == "Data for query: test" # type: ignore[attr-defined]
async def test_import_with_proxy_prompts():
@@ -322,11 +322,12 @@ async def test_import_with_proxy_prompts():
return f"Hello, {name} from API!"
proxy_app = FastMCP.as_proxy(Client(api_app))
- await main_app.import_server("api", proxy_app)
+ await main_app.import_server(proxy_app, "api")
- result = await main_app._mcp_get_prompt("api_greeting", {"name": "World"})
- assert result.messages[0].content.text == "Hello, World from API!" # type: ignore[attr-defined]
- assert result.description == "Example greeting prompt."
+ async with Client(main_app) as client:
+ result = await client.get_prompt("api_greeting", {"name": "World"})
+ assert result.messages[0].content.text == "Hello, World from API!" # type: ignore[attr-defined]
+ assert result.description == "Example greeting prompt."
async def test_import_with_proxy_resources():
@@ -349,7 +350,7 @@ async def test_import_with_proxy_resources():
}
proxy_app = FastMCP.as_proxy(Client(api_app))
- await main_app.import_server("api", proxy_app)
+ await main_app.import_server(proxy_app, "api")
# Access the resource through the main app with the prefixed key
async with Client(main_app) as client:
@@ -376,7 +377,7 @@ async def test_import_with_proxy_resource_templates():
return {"name": name, "email": email}
proxy_app = FastMCP.as_proxy(Client(api_app))
- await main_app.import_server("api", proxy_app)
+ await main_app.import_server(proxy_app, "api")
# Instantiate the template through the main app with the prefixed key
@@ -396,7 +397,7 @@ async def test_import_invalid_resource_prefix():
# This test doesn't apply anymore with the new prefix format since we're not validating
# the protocol://prefix/path format
# Just import the server to maintain test coverage without deprecated parameters
- await main_app.import_server("api_sub", api_app)
+ await main_app.import_server(api_app, "api")
async def test_import_invalid_resource_separator():
@@ -405,4 +406,202 @@ async def test_import_invalid_resource_separator():
# This test is for maintaining coverage for importing with prefixes
# We no longer pass the deprecated resource_separator parameter
- await main_app.import_server("api", api_app)
+ await main_app.import_server(api_app, "api")
+
+
+async def test_import_with_no_prefix():
+ """Test importing a server without providing a prefix."""
+ main_app = FastMCP("MainApp")
+ sub_app = FastMCP("SubApp")
+
+ @sub_app.tool
+ def sub_tool() -> str:
+ return "Sub tool result"
+
+ @sub_app.resource(uri="data://config")
+ def sub_resource():
+ return "Sub resource data"
+
+ @sub_app.resource(uri="users://{user_id}/info")
+ def sub_template(user_id: str):
+ return f"Sub template for user {user_id}"
+
+ @sub_app.prompt
+ def sub_prompt() -> str:
+ return "Sub prompt content"
+
+ # Import without prefix
+ await main_app.import_server(sub_app)
+
+ # Verify all component types are accessible with original names
+ assert "sub_tool" in main_app._tool_manager._tools
+ assert "data://config" in main_app._resource_manager._resources
+ assert "users://{user_id}/info" in main_app._resource_manager._templates
+ assert "sub_prompt" in main_app._prompt_manager._prompts
+
+ # Test actual functionality through Client
+ async with Client(main_app) as client:
+ # Test tool
+ tool_result = await client.call_tool("sub_tool", {})
+ assert tool_result[0].text == "Sub tool result" # type: ignore[attr-defined]
+
+ # Test resource
+ resource_result = await client.read_resource("data://config")
+ assert resource_result[0].text == "Sub resource data" # type: ignore[attr-defined]
+
+ # Test template
+ template_result = await client.read_resource("users://123/info")
+ assert template_result[0].text == "Sub template for user 123" # type: ignore[attr-defined]
+
+ # Test prompt
+ prompt_result = await client.get_prompt("sub_prompt", {})
+ assert prompt_result.messages is not None
+ assert prompt_result.messages[0].content.text == "Sub prompt content" # type: ignore[attr-defined]
+
+
+async def test_import_conflict_resolution_tools():
+ """Test that later imported tools overwrite earlier ones when names conflict."""
+ 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"
+
+ # Import both apps without prefix
+ await main_app.import_server(first_app)
+ await main_app.import_server(second_app)
+
+ async with Client(main_app) as client:
+ # The later imported server should win
+ 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
+
+ result = await client.call_tool("shared_tool", {})
+ assert result[0].text == "Second app tool" # type: ignore[attr-defined]
+
+
+async def test_import_conflict_resolution_resources():
+ """Test that later imported resources overwrite earlier ones when URIs conflict."""
+ 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"
+
+ # Import both apps without prefix
+ await main_app.import_server(first_app)
+ await main_app.import_server(second_app)
+
+ async with Client(main_app) as client:
+ # The later imported server should win
+ 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
+
+ result = await client.read_resource("shared://data")
+ assert result[0].text == "Second app data" # type: ignore[attr-defined]
+
+
+async def test_import_conflict_resolution_templates():
+ """Test that later imported templates overwrite earlier ones when URI templates conflict."""
+ 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}"
+
+ # Import both apps without prefix
+ await main_app.import_server(first_app)
+ await main_app.import_server(second_app)
+
+ async with Client(main_app) as client:
+ # The later imported server should win
+ 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
+
+ result = await client.read_resource("users://123/profile")
+ assert result[0].text == "Second app user 123" # type: ignore[attr-defined]
+
+
+async def test_import_conflict_resolution_prompts():
+ """Test that later imported prompts overwrite earlier ones when names conflict."""
+ 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"
+
+ # Import both apps without prefix
+ await main_app.import_server(first_app)
+ await main_app.import_server(second_app)
+
+ async with Client(main_app) as client:
+ # The later imported server should win
+ 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
+
+ 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_import_conflict_resolution_with_prefix():
+ """Test that later imported components overwrite earlier ones when prefixed names conflict."""
+ 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"
+
+ # Import both apps with same prefix
+ await main_app.import_server(first_app, "api")
+ await main_app.import_server(second_app, "api")
+
+ async with Client(main_app) as client:
+ # The later imported server should win
+ 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
+
+ result = await client.call_tool("api_shared_tool", {})
+ assert result[0].text == "Second app tool" # type: ignore[attr-defined]
diff --git a/tests/server/test_mount.py b/tests/server/test_mount.py
index 97c2efd3e..6dfb6af05 100644
--- a/tests/server/test_mount.py
+++ b/tests/server/test_mount.py
@@ -7,7 +7,6 @@ import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.client.transports import FastMCPTransport, SSETransport
-from fastmcp.exceptions import NotFoundError
from fastmcp.server.proxy import FastMCPProxy
@@ -26,7 +25,7 @@ class TestBasicMount:
return "This is from the sub app"
# Mount the sub-app to the main app
- main_app.mount("sub", sub_app)
+ main_app.mount(sub_app, "sub")
# Get tools from main app, should include sub_app's tools
tools = await main_app.get_tools()
@@ -46,7 +45,7 @@ class TestBasicMount:
return f"Hello, {name}!"
# Mount without custom separator - custom separators are deprecated
- main_app.mount("sub", sub_app)
+ main_app.mount(sub_app, "sub")
# Tool should be accessible with the default separator
tools = await main_app.get_tools()
@@ -62,7 +61,7 @@ class TestBasicMount:
# This test doesn't apply anymore with the new prefix format
# just mount the server to maintain test coverage
- main_app.mount("api:sub", api_app)
+ main_app.mount(api_app, "api:sub")
async def test_mount_invalid_resource_separator(self):
main_app = FastMCP("MainApp")
@@ -70,34 +69,7 @@ class TestBasicMount:
# This test doesn't apply anymore with the new prefix format
# Mount without deprecated parameters
- main_app.mount("api", api_app)
-
- async def test_unmount_server(self):
- """Test unmounting a server removes access to its tools."""
- main_app = FastMCP("MainApp")
- sub_app = FastMCP("SubApp")
-
- @sub_app.tool
- def sub_tool() -> str:
- return "This is from the sub app"
-
- # Mount the sub-app
- main_app.mount("sub", sub_app)
-
- # Verify it was mounted
- tools = await main_app.get_tools()
- assert "sub_sub_tool" in tools
-
- # Unmount the sub-app
- main_app.unmount("sub")
-
- # Verify it was unmounted
- tools = await main_app.get_tools()
- assert "sub_sub_tool" not in tools
-
- # Calling the tool should fail
- with pytest.raises(NotFoundError, match="Unknown tool: sub_sub_tool"):
- await main_app._mcp_call_tool("sub_sub_tool", {})
+ main_app.mount(api_app, "api")
async def test_mount_with_no_prefix(self):
main_app = FastMCP("MainApp")
@@ -108,11 +80,111 @@ class TestBasicMount:
return "This is from the sub app"
# Mount with empty prefix but without deprecated separators
- main_app.mount(prefix="", server=sub_app)
+ main_app.mount(sub_app, prefix="")
tools = await main_app.get_tools()
- # With empty prefix, the format is now "_sub_tool" instead of "sub_tool"
- assert "_sub_tool" in 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:
@@ -133,8 +205,8 @@ class TestMultipleServerMount:
return "News headlines"
# Mount both apps
- main_app.mount("weather", weather_app)
- main_app.mount("news", news_app)
+ main_app.mount(weather_app, "weather")
+ main_app.mount(news_app, "news")
# Check both are accessible
tools = await main_app.get_tools()
@@ -163,18 +235,16 @@ class TestMultipleServerMount:
return "Second app tool"
# Mount first app
- main_app.mount("api", 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("api", second_app)
+ main_app.mount(second_app, "api")
tools = await main_app.get_tools()
- # First app's tool should no longer be accessible
- assert "api_first_tool" not in tools
-
- # Second app's tool should be accessible
+ # Both apps' tools should be accessible (new behavior)
+ assert "api_first_tool" in tools
assert "api_second_tool" in tools
@pytest.mark.skipif(
@@ -199,7 +269,7 @@ class TestMultipleServerMount:
return "Working prompt"
# Mount the working server
- main_app.mount("working", working_app)
+ main_app.mount(working_app, "working")
# Use an unreachable port
unreachable_client = Client(
@@ -210,7 +280,7 @@ class TestMultipleServerMount:
unreachable_proxy = FastMCP.as_proxy(unreachable_client)
# Mount the unreachable proxy
- main_app.mount("unreachable", 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:
@@ -251,6 +321,252 @@ class TestMultipleServerMount:
)
+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."""
@@ -260,7 +576,7 @@ class TestDynamicChanges:
sub_app = FastMCP("SubApp")
# Mount the sub-app before adding any tools
- main_app.mount("sub", sub_app)
+ main_app.mount(sub_app, "sub")
# Initially, there should be no tools from sub_app
tools = await main_app.get_tools()
@@ -289,7 +605,7 @@ class TestDynamicChanges:
return "Temporary tool"
# Mount the sub-app
- main_app.mount("sub", sub_app)
+ main_app.mount(sub_app, "sub")
# Initially, the tool should be accessible
tools = await main_app.get_tools()
@@ -331,7 +647,7 @@ class TestResourcesAndTemplates:
return ["user1", "user2"]
# Mount the data app
- main_app.mount("data", data_app)
+ main_app.mount(data_app, "data")
# Resource should be accessible through main app
resources = await main_app.get_resources()
@@ -352,7 +668,7 @@ class TestResourcesAndTemplates:
return {"id": user_id, "name": f"User {user_id}"}
# Mount the user app
- main_app.mount("api", user_app)
+ main_app.mount(user_app, "api")
# Template should be accessible through main app
templates = await main_app.get_resource_templates()
@@ -371,7 +687,7 @@ class TestResourcesAndTemplates:
data_app = FastMCP("DataApp")
# Mount the data app before adding resources
- main_app.mount("data", data_app)
+ main_app.mount(data_app, "data")
# Add a resource after mounting
@data_app.resource(uri="data://config")
@@ -402,7 +718,7 @@ class TestPrompts:
return f"Hello, {name}!"
# Mount the assistant app
- main_app.mount("assistant", assistant_app)
+ main_app.mount(assistant_app, "assistant")
# Prompt should be accessible through main app
prompts = await main_app.get_prompts()
@@ -419,7 +735,7 @@ class TestPrompts:
assistant_app = FastMCP("AssistantApp")
# Mount the assistant app before adding prompts
- main_app.mount("assistant", assistant_app)
+ main_app.mount(assistant_app, "assistant")
# Add a prompt after mounting
@assistant_app.prompt
@@ -455,7 +771,7 @@ class TestProxyServer:
# Mount proxy server
main_app = FastMCP("MainApp")
- main_app.mount("proxy", proxy_server)
+ main_app.mount(proxy_server, "proxy")
# Tool should be accessible through main app
tools = await main_app.get_tools()
@@ -477,7 +793,7 @@ class TestProxyServer:
# Mount proxy server
main_app = FastMCP("MainApp")
- main_app.mount("proxy", proxy_server)
+ main_app.mount(proxy_server, "proxy")
# Add a tool to the original server
@original_server.tool
@@ -508,7 +824,7 @@ class TestProxyServer:
# Mount proxy server
main_app = FastMCP("MainApp")
- main_app.mount("proxy", proxy_server)
+ main_app.mount(proxy_server, "proxy")
# Resource should be accessible through main app
result = await main_app._mcp_read_resource("config://proxy/settings")
@@ -531,7 +847,7 @@ class TestProxyServer:
# Mount proxy server
main_app = FastMCP("MainApp")
- main_app.mount("proxy", proxy_server)
+ main_app.mount(proxy_server, "proxy")
# Prompt should be accessible through main app
result = await main_app._mcp_get_prompt("proxy_welcome", {"name": "World"})
@@ -546,26 +862,25 @@ class TestAsProxyKwarg:
mcp = FastMCP("Main")
sub = FastMCP("Sub")
- mcp.mount("sub", sub)
-
- assert mcp._mounted_servers["sub"].server is 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)
+ mcp.mount(sub, "sub", as_proxy=False)
- assert mcp._mounted_servers["sub"].server is sub
+ 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)
+ mcp.mount(sub, "sub", as_proxy=True)
- assert mcp._mounted_servers["sub"].server is not sub
- assert isinstance(mcp._mounted_servers["sub"].server, FastMCPProxy)
+ 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
@@ -575,43 +890,43 @@ class TestAsProxyKwarg:
mcp = FastMCP("Main")
sub = FastMCP("Sub", lifespan=lifespan)
- mcp.mount("sub", sub)
+ mcp.mount(sub, "sub")
- assert mcp._mounted_servers["sub"].server is not sub
- assert isinstance(mcp._mounted_servers["sub"].server, FastMCPProxy)
+ 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", sub_proxy)
+ mcp.mount(sub_proxy, "sub")
- assert mcp._mounted_servers["sub"].server is sub_proxy
+ 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", sub_proxy, as_proxy=False)
+ mcp.mount(sub_proxy, "sub", as_proxy=False)
- assert mcp._mounted_servers["sub"].server is sub_proxy
+ 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", sub_proxy, as_proxy=True)
+ mcp.mount(sub_proxy, "sub", as_proxy=True)
- assert mcp._mounted_servers["sub"].server is sub_proxy
+ 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)
+ mcp.mount(sub, "sub", as_proxy=True)
assert len(await mcp.get_tools()) == 0
@@ -636,7 +951,7 @@ class TestAsProxyKwarg:
def hello():
return "hi"
- mcp.mount("sub", sub, as_proxy=True)
+ mcp.mount(sub, "sub", as_proxy=True)
assert lifespan_check == []
diff --git a/tests/server/test_resource_prefix_formats.py b/tests/server/test_resource_prefix_formats.py
index b8273845d..cc13dd45e 100644
--- a/tests/server/test_resource_prefix_formats.py
+++ b/tests/server/test_resource_prefix_formats.py
@@ -26,8 +26,8 @@ async def test_resource_prefix_format_in_constructor():
main_server_protocol = FastMCP("MainProtocol", resource_prefix_format="protocol")
# Mount the servers
- main_server_path.mount("sub", server_path)
- main_server_protocol.mount("sub", server_protocol)
+ main_server_path.mount(server_path, "sub")
+ main_server_protocol.mount(server_protocol, "sub")
# Check that the resources are prefixed correctly
path_resources = await main_server_path.get_resources()
@@ -49,11 +49,11 @@ async def test_resource_prefix_format_in_import_server():
# Import with path format
main_server_path = FastMCP("MainPath", resource_prefix_format="path")
- await main_server_path.import_server("sub", server)
+ await main_server_path.import_server(server, "sub")
# Import with protocol format
main_server_protocol = FastMCP("MainProtocol", resource_prefix_format="protocol")
- await main_server_protocol.import_server("sub", server)
+ await main_server_protocol.import_server(server, "sub")
# Check that the resources are prefixed correctly
path_resources = main_server_path._resource_manager.get_resources()
diff --git a/tests/server/test_server.py b/tests/server/test_server.py
index 70a2c174b..403c94cae 100644
--- a/tests/server/test_server.py
+++ b/tests/server/test_server.py
@@ -1125,7 +1125,7 @@ class TestResourcePrefixMounting:
# Create a main server and mount the resource server
main_server = FastMCP(name="MainServer")
- main_server.mount("prefix", server)
+ main_server.mount(server, "prefix")
# Check that the resources are mounted with the correct prefixes
resources = await main_server.get_resources()
@@ -1219,7 +1219,7 @@ class TestResourcePrefixMounting:
# Create target server and import the source server
target_server = FastMCP(name="TargetServer")
- await target_server.import_server("imported", source_server)
+ await target_server.import_server(source_server, "imported")
# Check that the resources were imported with the correct prefixes
resources = await target_server.get_resources()
From 922e1d5b82559c8a8b604fc167b74f47836f2a9b Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Wed, 18 Jun 2025 11:48:44 -0400
Subject: [PATCH 3/3] Update docs
---
docs/servers/composition.mdx | 90 ++++++++++++++++++++++++++++++++----
1 file changed, 81 insertions(+), 9 deletions(-)
diff --git a/docs/servers/composition.mdx b/docs/servers/composition.mdx
index 1ed826a83..cdbe23186 100644
--- a/docs/servers/composition.mdx
+++ b/docs/servers/composition.mdx
@@ -26,9 +26,10 @@ The choice of importing or mounting depends on your use case and requirements.
| Feature | Importing | Mounting |
|---------|----------------|---------|
-| **Method** | `FastMCP.import_server()` | `FastMCP.mount()` |
+| **Method** | `FastMCP.import_server(server, prefix=None)` | `FastMCP.mount(server, prefix=None)` |
| **Composition Type** | One-time copy (static) | Live link (dynamic) |
| **Updates** | Changes to subserver NOT reflected | Changes to subserver immediately reflected |
+| **Prefix** | Optional - omit for original names | Optional - omit for original names |
| **Best For** | Bundling finalized components | Modular runtime composition |
### Proxy Servers
@@ -41,7 +42,7 @@ You can also create proxies from configuration dictionaries that follow the MCPC
## Importing (Static Composition)
-The `import_server()` method copies all components (tools, resources, templates, prompts) from one `FastMCP` instance (the *subserver*) into another (the *main server*). A `prefix` is added to avoid naming conflicts.
+The `import_server()` method copies all components (tools, resources, templates, prompts) from one `FastMCP` instance (the *subserver*) into another (the *main server*). An optional `prefix` can be provided to avoid naming conflicts. If no prefix is provided, components are imported without modification. When multiple servers are imported with the same prefix (or no prefix), the most recently imported server's components take precedence.
```python
from fastmcp import FastMCP
@@ -65,7 +66,7 @@ main_mcp = FastMCP(name="MainApp")
# Import subserver
async def setup():
- await main_mcp.import_server("weather", weather_mcp)
+ await main_mcp.import_server(weather_mcp, prefix="weather")
# Result: main_mcp now contains prefixed components:
# - Tool: "weather_get_forecast"
@@ -78,7 +79,7 @@ if __name__ == "__main__":
### How Importing Works
-When you call `await main_mcp.import_server(prefix, subserver)`:
+When you call `await main_mcp.import_server(subserver, prefix={whatever})`:
1. **Tools**: All tools from `subserver` are added to `main_mcp` with names prefixed using `{prefix}_`.
- `subserver.tool(name="my_tool")` becomes `main_mcp.tool(name="{prefix}_my_tool")`.
@@ -91,9 +92,63 @@ When you call `await main_mcp.import_server(prefix, subserver)`:
Note that `import_server` performs a **one-time copy** of components. Changes made to the `subserver` *after* importing **will not** be reflected in `main_mcp`. The `subserver`'s `lifespan` context is also **not** executed by the main server.
+
+The `prefix` parameter is optional. If omitted, components are imported without modification.
+
+
+#### Importing Without Prefixes
+
+
+
+You can also import servers without specifying a prefix, which copies components using their original names:
+
+```python
+
+from fastmcp import FastMCP
+import asyncio
+
+# Define subservers
+weather_mcp = FastMCP(name="WeatherService")
+
+@weather_mcp.tool
+def get_forecast(city: str) -> dict:
+ """Get weather forecast."""
+ return {"city": city, "forecast": "Sunny"}
+
+@weather_mcp.resource("data://cities/supported")
+def list_supported_cities() -> list[str]:
+ """List cities with weather support."""
+ return ["London", "Paris", "Tokyo"]
+
+# Define main server
+main_mcp = FastMCP(name="MainApp")
+
+# Import subserver
+async def setup():
+ # Import without prefix - components keep original names
+ await main_mcp.import_server(weather_mcp)
+
+# Result: main_mcp now contains:
+# - Tool: "get_forecast" (original name preserved)
+# - Resource: "data://cities/supported" (original URI preserved)
+
+if __name__ == "__main__":
+ asyncio.run(setup())
+ main_mcp.run()
+```
+
+#### Conflict Resolution
+
+
+
+When importing multiple servers with the same prefix, or no prefix, components from the **most recently imported** server take precedence.
+
+
+
+
## Mounting (Live Linking)
-The `mount()` method creates a **live link** between the `main_mcp` server and the `subserver`. Instead of copying components, requests for components matching the `prefix` are **delegated** to the `subserver` at runtime.
+The `mount()` method creates a **live link** between the `main_mcp` server and the `subserver`. Instead of copying components, requests for components matching the optional `prefix` are **delegated** to the `subserver` at runtime. If no prefix is provided, the subserver's components are accessible without prefixing. When multiple servers are mounted with the same prefix (or no prefix), the most recently mounted server takes precedence for conflicting component names.
```python
import asyncio
@@ -109,7 +164,7 @@ def initial_tool():
# Mount subserver (synchronous operation)
main_mcp = FastMCP(name="MainAppLive")
-main_mcp.mount("dynamic", dynamic_mcp)
+main_mcp.mount(dynamic_mcp, prefix="dynamic")
# Add a tool AFTER mounting - it will be accessible through main_mcp
@dynamic_mcp.tool
@@ -143,6 +198,20 @@ When mounting is configured:
The same prefixing rules apply as with `import_server` for naming tools, resources, templates, and prompts.
+
+ The `prefix` parameter is optional. If omitted, components are mounted without modification.
+
+
+
+#### Mounting Without Prefixes
+
+
+
+You can also mount servers without specifying a prefix, which makes components accessible without prefixing. This works identically to [importing without prefixes](#importing-without-prefixes), including [conflict resolution](#conflict-resolution).
+
+
+
+
### Direct vs. Proxy Mounting
@@ -161,10 +230,13 @@ FastMCP supports two mounting modes:
```python
# Direct mounting (default when no custom lifespan)
-main_mcp.mount("api", api_server)
+main_mcp.mount(api_server, prefix="api")
# Proxy mounting (preserves full client lifecycle)
-main_mcp.mount("api", api_server, as_proxy=True)
+main_mcp.mount(api_server, prefix="api", as_proxy=True)
+
+# Mounting without a prefix (components accessible without prefixing)
+main_mcp.mount(api_server)
```
FastMCP automatically uses proxy mounting when the mounted server has a custom lifespan, but you can override this behavior with the `as_proxy` parameter.
@@ -178,7 +250,7 @@ When using `FastMCP.as_proxy()` to create a proxy server, mounting that server w
remote_proxy = FastMCP.as_proxy(Client("http://example.com/mcp"))
# Mount the proxy (always uses proxy mounting)
-main_server.mount("remote", remote_proxy)
+main_server.mount(remote_proxy, prefix="remote")
```