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

Remove deprecated FastMCP server methods for 4.0

Browse source 
Removes deprecations that shipped replacements in 3.0:
- FastMCP.as_proxy() classmethod -> create_proxy()
- import_server() -> mount()
- mount(prefix=) -> namespace=
- mount(as_proxy=) -> now meaningless; mounts always invoke lifespan/middleware
- add_tool_transformation() -> add_transform(ToolTransform({...}))
- remove_tool_transformation() -> pure no-op that warned callers it had no
  effect; removed entirely
- FastMCP.remove_tool() -> mcp.local_provider.remove_tool()

Also drops the now-unused URI_PATTERN/re import and FastMCPDeprecationWarning
import. The _REMOVED_KWARGS TypeError shim is retained.
This commit is contained in:
Jeremiah Lowin 2026-07-06 21:24:51 -04:00
commit e5d9d1f17d
No known key found for this signature in database
3 changed files with 4 additions and 1035 deletions
Download patch file Download diff file Expand all files Collapse all files

246
fastmcp_slim/fastmcp/server/server.py
View file

@ -4,7 +4,6 @@ from __future__ import annotations
import asyncio
import logging
import re
import secrets
import warnings
from collections.abc import (
@ -47,7 +46,6 @@ from fastmcp.apps.config import (
)
from fastmcp.exceptions import (
AuthorizationError,
FastMCPDeprecationWarning,
FastMCPError,
NotFoundError,
PromptError,
@ -179,9 +177,6 @@ def _check_removed_kwargs(kwargs: dict[str, Any]) -> None:
Transport = Literal["stdio", "http", "sse", "streamable-http"]
# Compiled URI parsing regex to split a URI into protocol and path components
URI_PATTERN = re.compile(r"^([^:]+://)(.*?)$")
LifespanCallable = Callable[
["FastMCP[LifespanResultT]"], AbstractAsyncContextManager[LifespanResultT]
@ -624,38 +619,6 @@ class FastMCP(
"""
self._transforms.append(transform)
def add_tool_transformation(
self, tool_name: str, transformation: ToolTransformConfig
) -> None:
"""Add a tool transformation.
.. deprecated::
Use ``add_transform(ToolTransform({...}))`` instead.
"""
if fastmcp.settings.deprecation_warnings:
warnings.warn(
"add_tool_transformation is deprecated. Use "
"server.add_transform(ToolTransform({tool_name: config})) instead.",
FastMCPDeprecationWarning,
stacklevel=2,
)
self.add_transform(ToolTransform({tool_name: transformation}))
def remove_tool_transformation(self, _tool_name: str) -> None:
"""Remove a tool transformation.
.. deprecated::
Tool transformations are now immutable. Use enable/disable controls instead.
"""
if fastmcp.settings.deprecation_warnings:
warnings.warn(
"remove_tool_transformation is deprecated and has no effect. "
"Transforms are immutable once added. Use server.disable(keys=[...]) "
"to hide tools instead.",
FastMCPDeprecationWarning,
stacklevel=2,
)
async def list_tools(self, *, run_middleware: bool = True) -> Sequence[Tool]:
"""List all enabled tools from providers.
@ -1674,35 +1637,6 @@ class FastMCP(
"""
return self._local_provider.add_tool(tool)
def remove_tool(self, name: str, version: str | None = None) -> None:
"""Remove tool(s) from the server.
.. deprecated::
Use ``mcp.local_provider.remove_tool(name)`` instead.
Args:
name: The name of the tool to remove.
version: If None, removes ALL versions. If specified, removes only that version.
Raises:
NotFoundError: If no matching tool is found.
"""
if fastmcp.settings.deprecation_warnings:
warnings.warn(
"remove_tool() is deprecated. Use "
"mcp.local_provider.remove_tool(name) instead.",
FastMCPDeprecationWarning,
stacklevel=2,
)
try:
self._local_provider.remove_tool(name, version)
except KeyError:
if version is None:
raise NotFoundError(f"Tool {name!r} not found") from None
raise NotFoundError(
f"Tool {name!r} version {version!r} not found"
) from None
@overload
def tool(
self,
@ -2138,17 +2072,14 @@ class FastMCP(
self,
server: FastMCP[LifespanResultT],
namespace: str | None = None,
as_proxy: bool | None = None,
tool_names: dict[str, str] | None = None,
prefix: str | None = None, # deprecated, use namespace
) -> None:
"""Mount another FastMCP server on this server with an optional namespace.
Unlike importing (with import_server), mounting establishes a dynamic connection
between servers. When a client interacts with a mounted server's objects through
the parent server, requests are forwarded to the mounted server in real-time.
This means changes to the mounted server are immediately reflected when accessed
through the parent.
Mounting establishes a dynamic connection between servers. When a client
interacts with a mounted server's objects through the parent server, requests
are forwarded to the mounted server in real-time. This means changes to the
mounted server are immediately reflected when accessed through the parent.
When a server is mounted with a namespace:
- Tools from the mounted server are accessible with namespaced names.
@ -2174,48 +2105,15 @@ class FastMCP(
server: The FastMCP server to mount.
namespace: Optional namespace to use for the mounted server's objects. If None,
the server's objects are accessible with their original names.
as_proxy: Deprecated. Mounted servers now always have their lifespan and
middleware invoked. To create a proxy server, use create_proxy()
explicitly before mounting.
tool_names: Optional mapping of original tool names to custom names. Use this
to override namespaced names. Keys are the original tool names from the
mounted server.
prefix: Deprecated. Use namespace instead.
"""
import warnings
from fastmcp.server.providers.fastmcp_provider import FastMCPProvider
if server is self:
raise ValueError("Cannot mount a server onto itself")
# Handle deprecated prefix parameter
if prefix is not None:
warnings.warn(
"The 'prefix' parameter is deprecated, use 'namespace' instead",
FastMCPDeprecationWarning,
stacklevel=2,
)
if namespace is None:
namespace = prefix
else:
raise ValueError("Cannot specify both 'prefix' and 'namespace'")
if as_proxy is not None:
warnings.warn(
"as_proxy is deprecated and will be removed in a future version. "
"Mounted servers now always have their lifespan and middleware invoked. "
"To create a proxy server, use create_proxy() explicitly.",
FastMCPDeprecationWarning,
stacklevel=2,
)
# Still honor the flag for backward compatibility
if as_proxy:
from fastmcp.server.providers.proxy import FastMCPProxy
if not isinstance(server, FastMCPProxy):
server = FastMCP.as_proxy(server)
# Warn if parent masks errors but child doesn't (or vice versa)
if self._mask_error_details and not server._mask_error_details:
logger.warning(
@ -2240,105 +2138,6 @@ class FastMCP(
# Use add_provider with namespace (applies namespace in AggregateProvider)
self.add_provider(provider, namespace=namespace or "")
async def import_server(
self,
server: FastMCP[LifespanResultT],
prefix: str | None = None,
) -> None:
"""
Import the MCP objects from another FastMCP server into this one,
optionally with a given prefix.
.. deprecated::
Use :meth:`mount` instead. ``import_server`` will be removed in a
future version.
Note that when a server is *imported*, its objects are immediately
registered to the importing server. This is a one-time operation and
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 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"
- The resources are imported with prefixed URIs using the new format
Example: If server has a resource with URI "weather://forecast", it will
be available as "weather://prefix/forecast"
- The templates are imported with prefixed URI templates using the new format
Example: If server has a template with URI "weather://location/{id}", it will
be available as "weather://prefix/location/{id}"
- The prompts are imported with prefixed names
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:
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.
"""
import warnings
warnings.warn(
"import_server is deprecated, use mount() instead",
FastMCPDeprecationWarning,
stacklevel=2,
)
def add_resource_prefix(uri: str, prefix: str) -> str:
"""Add prefix to resource URI: protocol://path → protocol://prefix/path."""
match = URI_PATTERN.match(uri)
if match:
protocol, path = match.groups()
return f"{protocol}{prefix}/{path}"
return uri
# Import tools from the server
for tool in await server.list_tools():
if prefix:
tool = tool.model_copy(update={"name": f"{prefix}_{tool.name}"})
self.add_tool(tool)
# Import resources and templates from the server
for resource in await server.list_resources():
if prefix:
new_uri = add_resource_prefix(str(resource.uri), prefix)
resource = resource.model_copy(update={"uri": new_uri})
self.add_resource(resource)
for template in await server.list_resource_templates():
if prefix:
new_uri_template = add_resource_prefix(template.uri_template, prefix)
template = template.model_copy(
update={"uri_template": new_uri_template}
)
self.add_template(template)
# Import prompts from the server
for prompt in await server.list_prompts():
if prefix:
prompt = prompt.model_copy(update={"name": f"{prefix}_{prompt.name}"})
self.add_prompt(prompt)
if server._lifespan != default_lifespan:
from warnings import warn
warn(
message="When importing from a server with a lifespan, the lifespan from the imported server will not be used.",
category=RuntimeWarning,
stacklevel=2,
)
if prefix:
logger.debug(
f"[{self.name}] Imported server {server.name} with prefix '{prefix}'"
)
else:
logger.debug(f"[{self.name}] Imported server {server.name}")
@classmethod
def from_openapi(
cls,
@ -2445,43 +2244,6 @@ class FastMCP(
)
return cls(name=server_name, providers=[provider], **settings)
@classmethod
def as_proxy(
cls,
backend: (
Client[ClientTransportT]
| ClientTransport
| FastMCP[Any]
| SDKServer
| AnyUrl
| Path
| MCPConfig
| dict[str, Any]
| str
),
**settings: Any,
) -> FastMCPProxy:
"""Create a FastMCP proxy server for the given backend.
.. deprecated::
Use :func:`fastmcp.server.create_proxy` instead.
This method will be removed in a future version.
The `backend` argument can be either an existing `fastmcp.client.Client`
instance or any value accepted as the `transport` argument of
`fastmcp.client.Client`. This mirrors the convenience of the
`fastmcp.client.Client` constructor.
"""
if fastmcp.settings.deprecation_warnings:
warnings.warn(
"FastMCP.as_proxy() is deprecated. Use create_proxy() from "
"fastmcp.server instead: `from fastmcp.server import create_proxy`",
FastMCPDeprecationWarning,
stacklevel=2,
)
# Call the module-level create_proxy function directly
return create_proxy(backend, **settings)
@classmethod
def generate_name(cls, name: str | None = None) -> str:
class_name = cls.__name__

79
tests/deprecated/test_add_tool_transformation.py
View file

@ -1,79 +0,0 @@
"""Tests for deprecated add_tool_transformation API."""
import warnings
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.tools.tool_transform import ToolTransformConfig
class TestAddToolTransformationDeprecated:
"""Test that add_tool_transformation still works but emits deprecation warning."""
async def test_add_tool_transformation_emits_warning(self):
"""add_tool_transformation should emit deprecation warning."""
mcp = FastMCP("test")
@mcp.tool
def my_tool() -> str:
return "hello"
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
mcp.add_tool_transformation(
"my_tool", ToolTransformConfig(name="renamed_tool")
)
assert len(w) == 1
assert issubclass(w[0].category, DeprecationWarning)
assert "add_tool_transformation is deprecated" in str(w[0].message)
async def test_add_tool_transformation_still_works(self):
"""add_tool_transformation should still apply the transformation."""
mcp = FastMCP("test")
@mcp.tool
def verbose_tool_name() -> str:
return "result"
# Suppress warning for this test - we just want to verify it works
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
mcp.add_tool_transformation(
"verbose_tool_name", ToolTransformConfig(name="short")
)
async with Client(mcp) as client:
tools = await client.list_tools()
tool_names = [t.name for t in tools]
# Original name should be gone, renamed version should exist
assert "verbose_tool_name" not in tool_names
assert "short" in tool_names
# Should be callable by new name
result = await client.call_tool("short", {})
assert result.content[0].text == "result"
async def test_remove_tool_transformation_emits_warning(self):
"""remove_tool_transformation should emit deprecation warning."""
mcp = FastMCP("test")
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
mcp.remove_tool_transformation("any_tool")
assert len(w) == 1
assert issubclass(w[0].category, DeprecationWarning)
assert "remove_tool_transformation is deprecated" in str(w[0].message)
assert "no effect" in str(w[0].message)
async def test_tool_transformations_constructor_raises_type_error(self):
"""tool_transformations constructor param should raise TypeError."""
import pytest
with pytest.raises(TypeError, match="no longer accepts `tool_transformations`"):
FastMCP(
"test",
tool_transformations={"my_tool": ToolTransformConfig(name="renamed")},
)

714
tests/deprecated/test_import_server.py
View file

@ -1,714 +0,0 @@
import json
from urllib.parse import quote
from mcp_types import TextContent, TextResourceContents
from fastmcp.client.client import Client
from fastmcp.server.server import FastMCP
from fastmcp.tools.base import Tool
from fastmcp.tools.function_tool import FunctionTool
from tests.conftest import get_fn_name
async def test_import_basic_functionality():
"""Test that the import method properly imports tools and other resources."""
# Create main app and sub-app
main_app = FastMCP("MainApp")
sub_app = FastMCP("SubApp")
# Add a tool to the sub-app
@sub_app.tool
def sub_tool() -> str:
return "This is from the sub app"
# Import the sub-app to the main app
await main_app.import_server(sub_app, "sub")
# Verify the tool was imported with the prefix
main_tools = await main_app.list_tools()
sub_tools = await sub_app.list_tools()
assert any(t.name == "sub_sub_tool" for t in main_tools)
assert any(t.name == "sub_tool" for t in sub_tools)
# Verify the original tool still exists in the sub-app
tool = await main_app.get_tool("sub_sub_tool")
assert tool is not None
# import_server creates copies with prefixed names (unlike mount which proxies)
assert tool.name == "sub_sub_tool"
assert isinstance(tool, FunctionTool)
assert callable(tool.fn)
async def test_import_multiple_apps():
"""Test importing multiple apps to a main app."""
# Create main app and multiple sub-apps
main_app = FastMCP("MainApp")
weather_app = FastMCP("WeatherApp")
news_app = FastMCP("NewsApp")
# Add tools to each sub-app
@weather_app.tool
def get_forecast() -> str:
return "Weather forecast"
@news_app.tool
def get_headlines() -> str:
return "News headlines"
# Import both sub-apps to the main 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
tools = await main_app.list_tools()
assert any(t.name == "weather_get_forecast" for t in tools)
assert any(t.name == "news_get_headlines" for t in tools)
async def test_import_combines_tools():
"""Test that importing preserves existing tools with the same prefix."""
# Create apps
main_app = FastMCP("MainApp")
first_app = FastMCP("FirstApp")
second_app = FastMCP("SecondApp")
# Add tools to each sub-app
@first_app.tool
def first_tool() -> str:
return "First app tool"
@second_app.tool
def second_tool() -> str:
return "Second app tool"
# Import first app
await main_app.import_server(first_app, "api")
tools = await main_app.list_tools()
assert any(t.name == "api_first_tool" for t in tools)
# Import second app to same prefix
await main_app.import_server(second_app, "api")
# Verify second tool is there
tools = await main_app.list_tools()
assert any(t.name == "api_second_tool" for t in tools)
# Tools from both imports are combined
assert any(t.name == "api_first_tool" for t in tools)
async def test_import_with_resources():
"""Test importing with resources."""
# Create apps
main_app = FastMCP("MainApp")
data_app = FastMCP("DataApp")
# Add a resource to the data app
@data_app.resource(uri="data://users")
async def get_users() -> str:
return "user1, user2"
# Import the data app
await main_app.import_server(data_app, "data")
# Verify the resource was imported with the prefix
resources = await main_app.list_resources()
assert any(str(r.uri) == "data://data/users" for r in resources)
async def test_import_with_resource_templates():
"""Test importing with resource templates."""
# Create apps
main_app = FastMCP("MainApp")
user_app = FastMCP("UserApp")
# Add a resource template to the user app
@user_app.resource(uri="users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
import json
return json.dumps(
{"id": user_id, "name": f"User {user_id}"}, separators=(",", ":")
)
# Import the user app
await main_app.import_server(user_app, "api")
# Verify the template was imported with the prefix
templates = await main_app.list_resource_templates()
assert any(t.uri_template == "users://api/{user_id}/profile" for t in templates)
async def test_import_with_prompts():
"""Test importing with prompts."""
# Create apps
main_app = FastMCP("MainApp")
assistant_app = FastMCP("AssistantApp")
# Add a prompt to the assistant app
@assistant_app.prompt
def greeting(name: str) -> str:
return f"Hello, {name}!"
# Import the assistant app
await main_app.import_server(assistant_app, "assistant")
# Verify the prompt was imported with the prefix
prompts = await main_app.list_prompts()
assert any(p.name == "assistant_greeting" for p in prompts)
async def test_import_multiple_resource_templates():
"""Test importing multiple apps with resource templates."""
# Create apps
main_app = FastMCP("MainApp")
weather_app = FastMCP("WeatherApp")
news_app = FastMCP("NewsApp")
# Add templates to each app
@weather_app.resource(uri="weather://{city}")
def get_weather(city: str) -> str:
return f"Weather for {city}"
@news_app.resource(uri="news://{category}")
def get_news(category: str) -> str:
return f"News for {category}"
# Import both apps
await main_app.import_server(weather_app, "data")
await main_app.import_server(news_app, "content")
# Verify templates were imported with correct prefixes
templates = await main_app.list_resource_templates()
assert any(t.uri_template == "weather://data/{city}" for t in templates)
assert any(t.uri_template == "news://content/{category}" for t in templates)
async def test_import_multiple_prompts():
"""Test importing multiple apps with prompts."""
# Create apps
main_app = FastMCP("MainApp")
python_app = FastMCP("PythonApp")
sql_app = FastMCP("SQLApp")
# Add prompts to each app
@python_app.prompt
def review_python(code: str) -> str:
return f"Reviewing Python code:\n{code}"
@sql_app.prompt
def explain_sql(query: str) -> str:
return f"Explaining SQL query:\n{query}"
# Import both apps
await main_app.import_server(python_app, "python")
await main_app.import_server(sql_app, "sql")
# Verify prompts were imported with correct prefixes
prompts = await main_app.list_prompts()
assert any(p.name == "python_review_python" for p in prompts)
assert any(p.name == "sql_explain_sql" for p in prompts)
async def test_tool_custom_name_preserved_when_imported():
"""Test that a tool's custom name is preserved when imported."""
main_app = FastMCP("MainApp")
api_app = FastMCP("APIApp")
def fetch_data(query: str) -> str:
return f"Data for query: {query}"
api_app.add_tool(Tool.from_function(fetch_data, name="get_data"))
await main_app.import_server(api_app, "api")
# Check that the tool is accessible by its prefixed name
tool = await main_app.get_tool("api_get_data")
assert tool is not None
# Check that the function name is preserved
assert isinstance(tool, FunctionTool)
assert get_fn_name(tool.fn) == "fetch_data"
async def test_call_imported_custom_named_tool():
"""Test calling an imported tool with a custom name."""
main_app = FastMCP("MainApp")
api_app = FastMCP("APIApp")
def fetch_data(query: str) -> str:
return f"Data for query: {query}"
api_app.add_tool(Tool.from_function(fetch_data, name="get_data"))
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"})
assert result.data == "Data for query: test"
async def test_first_level_importing_with_custom_name():
"""Test that a tool with a custom name is correctly imported at the first level."""
service_app = FastMCP("ServiceApp")
provider_app = FastMCP("ProviderApp")
def calculate_value(input: int) -> int:
return input * 2
provider_app.add_tool(Tool.from_function(calculate_value, name="compute"))
await service_app.import_server(provider_app, "provider")
# Tool is accessible in the service app with the first prefix
tool = await service_app.get_tool("provider_compute")
assert tool is not None
assert isinstance(tool, FunctionTool)
assert get_fn_name(tool.fn) == "calculate_value"
async def test_nested_importing_preserves_prefixes():
"""Test that importing a previously imported app preserves prefixes."""
main_app = FastMCP("MainApp")
service_app = FastMCP("ServiceApp")
provider_app = FastMCP("ProviderApp")
def calculate_value(input: int) -> int:
return input * 2
provider_app.add_tool(Tool.from_function(calculate_value, name="compute"))
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 = await main_app.get_tool("service_provider_compute")
assert tool is not None
async def test_call_nested_imported_tool():
"""Test calling a tool through multiple levels of importing."""
main_app = FastMCP("MainApp")
service_app = FastMCP("ServiceApp")
provider_app = FastMCP("ProviderApp")
def calculate_value(input: int) -> int:
return input * 2
provider_app.add_tool(Tool.from_function(calculate_value, name="compute"))
await service_app.import_server(provider_app, "provider")
await main_app.import_server(service_app, "service")
async with Client(main_app) as client:
result = await client.call_tool("service_provider_compute", {"input": 21})
assert result.data == 42
async def test_import_with_proxy_tools():
"""
Test importing with tools that have custom names (proxy tools).
This tests that the tool's name doesn't change even though the registered
name does, which is important because we need to forward that name to the
proxy server correctly.
"""
# Create apps
main_app = FastMCP("MainApp")
api_app = FastMCP("APIApp")
@api_app.tool
def get_data(query: str) -> str:
return f"Data for query: {query}"
proxy_app = FastMCP.as_proxy(api_app)
await main_app.import_server(proxy_app, "api")
async with Client(main_app) as client:
result = await client.call_tool("api_get_data", {"query": "test"})
assert result.data == "Data for query: test"
async def test_import_with_proxy_prompts():
"""
Test importing with prompts that have custom keys.
This tests that the prompt's name doesn't change even though the registered
key does, which is important for correct rendering.
"""
# Create apps
main_app = FastMCP("MainApp")
api_app = FastMCP("APIApp")
@api_app.prompt
def greeting(name: str) -> str:
"""Example greeting prompt."""
return f"Hello, {name} from API!"
proxy_app = FastMCP.as_proxy(api_app)
await main_app.import_server(proxy_app, "api")
async with Client(main_app) as client:
result = await client.get_prompt("api_greeting", {"name": "World"})
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Hello, World from API!"
assert result.description == "Example greeting prompt."
async def test_import_with_proxy_resources():
"""
Test importing with resources that have custom keys.
This tests that the resource's name doesn't change even though the registered
key does, which is important for correct access.
"""
# Create apps
main_app = FastMCP("MainApp")
api_app = FastMCP("APIApp")
# Create a resource in the API app
@api_app.resource(uri="config://settings")
def get_config() -> str:
import json
return json.dumps(
{
"api_key": "12345",
"base_url": "https://api.example.com",
}
)
proxy_app = FastMCP.as_proxy(api_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:
result = await client.read_resource("config://api/settings")
assert isinstance(result[0], TextResourceContents)
content = json.loads(result[0].text)
assert content["api_key"] == "12345"
assert content["base_url"] == "https://api.example.com"
async def test_import_with_proxy_resource_templates():
"""
Test importing with resource templates that have custom keys.
This tests that the template's name doesn't change even though the registered
key does, which is important for correct instantiation.
"""
# Create apps
main_app = FastMCP("MainApp")
api_app = FastMCP("APIApp")
# Create a resource template in the API app
@api_app.resource(uri="user://{name}/{email}")
def create_user(name: str, email: str) -> str:
import json
return json.dumps({"name": name, "email": email})
proxy_app = FastMCP.as_proxy(api_app)
await main_app.import_server(proxy_app, "api")
# Instantiate the template through the main app with the prefixed key
quoted_name = quote("John Doe", safe="")
quoted_email = quote("john@example.com", safe="")
async with Client(main_app) as client:
result = await client.read_resource(f"user://api/{quoted_name}/{quoted_email}")
assert isinstance(result[0], TextResourceContents)
content = json.loads(result[0].text)
assert content["name"] == "John Doe"
assert content["email"] == "john@example.com"
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
tools = await main_app.list_tools()
resources = await main_app.list_resources()
templates = await main_app.list_resource_templates()
prompts = await main_app.list_prompts()
assert any(t.name == "sub_tool" for t in tools)
assert any(str(r.uri) == "data://config" for r in resources)
assert any(t.uri_template == "users://{user_id}/info" for t in templates)
assert any(p.name == "sub_prompt" for p in 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.data == "Sub tool result"
# Test resource
resource_result = await client.read_resource("data://config")
assert isinstance(resource_result[0], TextResourceContents)
assert resource_result[0].text == "Sub resource data"
# Test template
template_result = await client.read_resource("users://123/info")
assert isinstance(template_result[0], TextResourceContents)
assert template_result[0].text == "Sub template for user 123"
# Test prompt
prompt_result = await client.get_prompt("sub_prompt", {})
assert prompt_result.messages is not None
assert isinstance(prompt_result.messages[0].content, TextContent)
assert prompt_result.messages[0].content.text == "Sub prompt content"
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.data == "Second app tool"
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 isinstance(result[0], TextResourceContents)
assert result[0].text == "Second app data"
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.uri_template 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 isinstance(result[0], TextResourceContents)
assert result[0].text == "Second app user 123"
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 isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Second app prompt"
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.data == "Second app tool"
async def test_import_server_resource_uri_prefixing():
"""Test that resource URIs are prefixed when using import_server (names are NOT prefixed)."""
# Create a sub-server with a resource
sub_server = FastMCP("SubServer")
@sub_server.resource("resource://test_resource")
def test_resource() -> str:
return "Test content"
# Create main server and import sub-server with prefix
main_server = FastMCP("MainServer")
await main_server.import_server(sub_server, prefix="imported")
# Get resources and verify URI prefixing (name should NOT be prefixed)
resources = await main_server.list_resources()
resource = next(
r for r in resources if str(r.uri) == "resource://imported/test_resource"
)
assert resource.name == "test_resource"
async def test_import_server_resource_template_uri_prefixing():
"""Test that resource template URIs are prefixed when using import_server (names are NOT prefixed)."""
# Create a sub-server with a resource template
sub_server = FastMCP("SubServer")
@sub_server.resource("resource://data/{item_id}")
def data_template(item_id: str) -> str:
return f"Data for {item_id}"
# Create main server and import sub-server with prefix
main_server = FastMCP("MainServer")
await main_server.import_server(sub_server, prefix="imported")
# Get resource templates and verify URI prefixing (name should NOT be prefixed)
templates = await main_server.list_resource_templates()
template = next(
t for t in templates if t.uri_template == "resource://imported/data/{item_id}"
)
assert template.name == "data_template"
async def test_import_server_with_new_prefix_format():
"""Test that import_server correctly uses the new prefix format."""
# Create a server with resources
source_server = FastMCP(name="SourceServer")
@source_server.resource("resource://test-resource")
def get_resource():
return "Resource content"
@source_server.resource("resource:///absolute/path")
def get_absolute_resource():
return "Absolute resource content"
@source_server.resource("resource://{param}/template")
def get_template_resource(param: str):
return f"Template resource with {param}"
# Create target server and import the source server
target_server = FastMCP(name="TargetServer")
await target_server.import_server(source_server, "imported")
# Check that the resources were imported with the correct prefixes
resources = await target_server.list_resources()
templates = await target_server.list_resource_templates()
assert any(str(r.uri) == "resource://imported/test-resource" for r in resources)
assert any(str(r.uri) == "resource://imported//absolute/path" for r in resources)
assert any(
t.uri_template == "resource://imported/{param}/template" for t in templates
)
# Verify we can access the resources
async with Client(target_server) as client:
result = await client.read_resource("resource://imported/test-resource")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Resource content"
result = await client.read_resource("resource://imported//absolute/path")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Absolute resource content"
result = await client.read_resource("resource://imported/param-value/template")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource with param-value"