ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-24 06:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Make prefixes optional

Browse source
This commit is contained in:
Jeremiah Lowin 2025-06-18 11:28:24 -04:00
commit 8eaa6096b4
9 changed files with 1139 additions and 240 deletions
Download patch file Download diff file Expand all files Collapse all files

265
src/fastmcp/server/server.py
View file

@ -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]

275
tests/deprecated/test_mount_import_arg_order.py Normal file
View file

@ -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

83
tests/deprecated/test_mount_separators.py
View file

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

10
tests/deprecated/test_resource_prefixes.py
View file

@ -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()

22
tests/server/openapi/test_openapi.py
View file

@ -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
):

267
tests/server/test_import_server.py
View file

@ -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]

465
tests/server/test_mount.py
View file

@ -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 == []

8
tests/server/test_resource_prefix_formats.py
View file

@ -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()

4
tests/server/test_server.py
View file

@ -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()