diff --git a/fastmcp_slim/fastmcp/server/server.py b/fastmcp_slim/fastmcp/server/server.py index dc64c44c2..a7ccba6cc 100644 --- a/fastmcp_slim/fastmcp/server/server.py +++ b/fastmcp_slim/fastmcp/server/server.py @@ -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__ diff --git a/tests/deprecated/test_add_tool_transformation.py b/tests/deprecated/test_add_tool_transformation.py deleted file mode 100644 index 0228b8b60..000000000 --- a/tests/deprecated/test_add_tool_transformation.py +++ /dev/null @@ -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")}, - ) diff --git a/tests/deprecated/test_import_server.py b/tests/deprecated/test_import_server.py deleted file mode 100644 index d43382ad6..000000000 --- a/tests/deprecated/test_import_server.py +++ /dev/null @@ -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"