Make resource prefix format configurable

This commit is contained in:
Jeremiah Lowin 2025-05-21 12:37:21 -04:00
commit 85c2acf66d
5 changed files with 338 additions and 45 deletions

View file

@ -177,6 +177,49 @@ remote_proxy = FastMCP.as_proxy(Client("http://example.com/mcp"))
main_server.mount("remote", remote_proxy)
```
<Warning>
Some MCP clients (like Claude Desktop) might have restrictions on characters allowed in tool names. FastMCP uses standard naming conventions: tools and prompts are prefixed with `{prefix}_` (e.g., "weather_forecast"), and resources use the format `protocol://{prefix}/path` (e.g., "data://weather/forecast").
</Warning>
## Resource Prefix Formats
When mounting or importing servers, resource URIs are usually prefixed to avoid naming conflicts. FastMCP supports two different formats for resource prefixes:
### Path Format (Default)
In path format, prefixes are added to the path component of the URI:
```
resource://prefix/path/to/resource
```
This is the default format since FastMCP 2.4. This format is recommended because it avoids issues with URI protocol restrictions (like underscores not being allowed in protocol names).
### Protocol Format (Legacy)
In protocol format, prefixes are added as part of the protocol:
```
prefix+resource://path/to/resource
```
This was the default format in FastMCP before 2.4. While still supported, it's not recommended for new code as it can cause problems with prefix names that aren't valid in URI protocols.
### Configuring the Prefix Format
You can configure the prefix format globally:
```python
from fastmcp import settings
settings.settings.resource_prefix_format = "protocol" # Switch to legacy format
```
Or per-server:
```python
# Create a server that uses legacy protocol format
server = FastMCP("LegacyServer", resource_prefix_format="protocol")
# Create a server that uses new path format
server = FastMCP("NewServer", resource_prefix_format="path")
```
When mounting or importing servers, the prefix format of the parent server is used.

View file

@ -123,6 +123,7 @@ class FastMCP(Generic[LifespanResultT]):
on_duplicate_tools: DuplicateBehavior | None = None,
on_duplicate_resources: DuplicateBehavior | None = None,
on_duplicate_prompts: DuplicateBehavior | None = None,
resource_prefix_format: Literal["protocol", "path"] | None = None,
**settings: Any,
):
if settings:
@ -137,6 +138,14 @@ class FastMCP(Generic[LifespanResultT]):
)
self.settings = fastmcp.settings.ServerSettings(**settings)
self.resource_prefix_format: Literal["protocol", "path"]
if resource_prefix_format is None:
self.resource_prefix_format = (
fastmcp.settings.settings.resource_prefix_format
)
else:
self.resource_prefix_format = resource_prefix_format
self.tags: set[str] = tags or set()
self.dependencies = dependencies
self._cache = TimedCache(
@ -1109,11 +1118,11 @@ class FastMCP(Generic[LifespanResultT]):
# Import resources and templates from the mounted server
for key, resource in (await server.get_resources()).items():
prefixed_key = add_resource_prefix(key, prefix)
prefixed_key = add_resource_prefix(key, prefix, self.resource_prefix_format)
self._resource_manager.add_resource(resource, key=prefixed_key)
for key, template in (await server.get_resource_templates()).items():
prefixed_key = add_resource_prefix(key, prefix)
prefixed_key = add_resource_prefix(key, prefix, self.resource_prefix_format)
self._resource_manager.add_template(template, key=prefixed_key)
# Import prompts from the mounted server
@ -1258,14 +1267,18 @@ class MountedServer:
async def get_resources(self) -> dict[str, Resource]:
resources = await self.server.get_resources()
return {
add_resource_prefix(key, self.prefix): resource
add_resource_prefix(
key, self.prefix, self.server.resource_prefix_format
): resource
for key, resource in resources.items()
}
async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
templates = await self.server.get_resource_templates()
return {
add_resource_prefix(key, self.prefix): template
add_resource_prefix(
key, self.prefix, self.server.resource_prefix_format
): template
for key, template in templates.items()
}
@ -1280,10 +1293,12 @@ class MountedServer:
return key.removeprefix(f"{self.prefix}_")
def match_resource(self, key: str) -> bool:
return has_resource_prefix(key, self.prefix)
return has_resource_prefix(key, self.prefix, self.server.resource_prefix_format)
def strip_resource_prefix(self, key: str) -> str:
return remove_resource_prefix(key, self.prefix)
return remove_resource_prefix(
key, self.prefix, self.server.resource_prefix_format
)
def match_prompt(self, key: str) -> bool:
return key.startswith(f"{self.prefix}_")
@ -1292,7 +1307,9 @@ class MountedServer:
return key.removeprefix(f"{self.prefix}_")
def add_resource_prefix(uri: str, prefix: str) -> str:
def add_resource_prefix(
uri: str, prefix: str, prefix_format: Literal["protocol", "path"] | None = None
) -> str:
"""Add a prefix to a resource URI.
Args:
@ -1304,9 +1321,11 @@ def add_resource_prefix(uri: str, prefix: str) -> str:
Examples:
>>> add_resource_prefix("resource://path/to/resource", "prefix")
"resource://prefix/path/to/resource"
"resource://prefix/path/to/resource" # with new style
>>> add_resource_prefix("resource://path/to/resource", "prefix")
"prefix+resource://path/to/resource" # with legacy style
>>> add_resource_prefix("resource:///absolute/path", "prefix")
"resource://prefix//absolute/path"
"resource://prefix//absolute/path" # with new style
Raises:
ValueError: If the URI doesn't match the expected protocol://path format
@ -1314,32 +1333,50 @@ def add_resource_prefix(uri: str, prefix: str) -> str:
if not prefix:
return uri
# Split the URI into protocol and path
match = URI_PATTERN.match(uri)
if not match:
raise ValueError(f"Invalid URI format: {uri}. Expected protocol://path format.")
# Get the server settings to check for legacy format preference
protocol, path = match.groups()
if prefix_format is None:
prefix_format = fastmcp.settings.settings.resource_prefix_format
# Add the prefix to the path
return f"{protocol}{prefix}/{path}"
if prefix_format == "protocol":
# Legacy style: prefix+protocol://path
return f"{prefix}+{uri}"
elif prefix_format == "path":
# New style: protocol://prefix/path
# Split the URI into protocol and path
match = URI_PATTERN.match(uri)
if not match:
raise ValueError(
f"Invalid URI format: {uri}. Expected protocol://path format."
)
protocol, path = match.groups()
# Add the prefix to the path
return f"{protocol}{prefix}/{path}"
else:
raise ValueError(f"Invalid prefix format: {prefix_format}")
def remove_resource_prefix(uri: str, prefix: str) -> str:
def remove_resource_prefix(
uri: str, prefix: str, prefix_format: Literal["protocol", "path"] | None = None
) -> str:
"""Remove a prefix from a resource URI.
Args:
uri: The resource URI with a prefix
prefix: The prefix to remove
prefix_format: The format of the prefix to remove
Returns:
The resource URI with the prefix removed
Examples:
>>> remove_resource_prefix("resource://prefix/path/to/resource", "prefix")
"resource://path/to/resource"
"resource://path/to/resource" # with new style
>>> remove_resource_prefix("prefix+resource://path/to/resource", "prefix")
"resource://path/to/resource" # with legacy style
>>> remove_resource_prefix("resource://prefix//absolute/path", "prefix")
"resource:///absolute/path"
"resource:///absolute/path" # with new style
Raises:
ValueError: If the URI doesn't match the expected protocol://path format
@ -1347,24 +1384,41 @@ def remove_resource_prefix(uri: str, prefix: str) -> str:
if not prefix:
return uri
# Split the URI into protocol and path
match = URI_PATTERN.match(uri)
if not match:
raise ValueError(f"Invalid URI format: {uri}. Expected protocol://path format.")
if prefix_format is None:
prefix_format = fastmcp.settings.settings.resource_prefix_format
protocol, path = match.groups()
# Check if the path starts with the prefix followed by a /
prefix_pattern = f"^{re.escape(prefix)}/(.*?)$"
path_match = re.match(prefix_pattern, path)
if not path_match:
if prefix_format == "protocol":
# Legacy style: prefix+protocol://path
legacy_prefix = f"{prefix}+"
if uri.startswith(legacy_prefix):
return uri[len(legacy_prefix) :]
return uri
elif prefix_format == "path":
# New style: protocol://prefix/path
# Split the URI into protocol and path
match = URI_PATTERN.match(uri)
if not match:
raise ValueError(
f"Invalid URI format: {uri}. Expected protocol://path format."
)
# Return the URI without the prefix
return f"{protocol}{path_match.group(1)}"
protocol, path = match.groups()
# Check if the path starts with the prefix followed by a /
prefix_pattern = f"^{re.escape(prefix)}/(.*?)$"
path_match = re.match(prefix_pattern, path)
if not path_match:
return uri
# Return the URI without the prefix
return f"{protocol}{path_match.group(1)}"
else:
raise ValueError(f"Invalid prefix format: {prefix_format}")
def has_resource_prefix(uri: str, prefix: str) -> bool:
def has_resource_prefix(
uri: str, prefix: str, prefix_format: Literal["protocol", "path"] | None = None
) -> bool:
"""Check if a resource URI has a specific prefix.
Args:
@ -1376,7 +1430,9 @@ def has_resource_prefix(uri: str, prefix: str) -> bool:
Examples:
>>> has_resource_prefix("resource://prefix/path/to/resource", "prefix")
True
True # with new style
>>> has_resource_prefix("prefix+resource://path/to/resource", "prefix")
True # with legacy style
>>> has_resource_prefix("resource://other/path/to/resource", "prefix")
False
@ -1386,13 +1442,28 @@ def has_resource_prefix(uri: str, prefix: str) -> bool:
if not prefix:
return False
# Split the URI into protocol and path
match = URI_PATTERN.match(uri)
if not match:
raise ValueError(f"Invalid URI format: {uri}. Expected protocol://path format.")
# Get the server settings to check for legacy format preference
_, path = match.groups()
if prefix_format is None:
prefix_format = fastmcp.settings.settings.resource_prefix_format
# Check if the path starts with the prefix followed by a /
prefix_pattern = f"^{re.escape(prefix)}/"
return bool(re.match(prefix_pattern, path))
if prefix_format == "protocol":
# Legacy style: prefix+protocol://path
legacy_prefix = f"{prefix}+"
return uri.startswith(legacy_prefix)
elif prefix_format == "path":
# New style: protocol://prefix/path
# Split the URI into protocol and path
match = URI_PATTERN.match(uri)
if not match:
raise ValueError(
f"Invalid URI format: {uri}. Expected protocol://path format."
)
_, path = match.groups()
# Check if the path starts with the prefix followed by a /
prefix_pattern = f"^{re.escape(prefix)}/"
return bool(re.match(prefix_pattern, path))
else:
raise ValueError(f"Invalid prefix format: {prefix_format}")

View file

@ -29,6 +29,7 @@ class Settings(BaseSettings):
test_mode: bool = False
log_level: LOG_LEVEL = "INFO"
client_raise_first_exceptiongroup_error: Annotated[
bool,
Field(
@ -44,6 +45,21 @@ class Settings(BaseSettings):
),
),
] = True
resource_prefix_format: Annotated[
Literal["protocol", "path"],
Field(
default="path",
description=inspect.cleandoc(
"""
When perfixing a resource URI, either use path formatting (resource://prefix/path)
or protocol formatting (prefix+resource://path). Protocol formatting was the default in FastMCP < 2.4;
path formatting is current default.
"""
),
),
] = "path"
tool_attempt_parse_json_args: Annotated[
bool,
Field(

View file

@ -0,0 +1,98 @@
"""Tests for legacy resource prefix behavior."""
from fastmcp import Client, FastMCP
from fastmcp.server.server import (
add_resource_prefix,
has_resource_prefix,
remove_resource_prefix,
)
from fastmcp.utilities.tests import temporary_settings
class TestLegacyResourcePrefixes:
"""Test the legacy resource prefix behavior."""
def test_add_resource_prefix_legacy(self):
"""Test that add_resource_prefix uses the legacy format when resource_prefix_format is 'protocol'."""
with temporary_settings(resource_prefix_format="protocol"):
result = add_resource_prefix("resource://path/to/resource", "prefix")
assert result == "prefix+resource://path/to/resource"
# Empty prefix should return the original URI
result = add_resource_prefix("resource://path/to/resource", "")
assert result == "resource://path/to/resource"
def test_remove_resource_prefix_legacy(self):
"""Test that remove_resource_prefix uses the legacy format when resource_prefix_format is 'protocol'."""
with temporary_settings(resource_prefix_format="protocol"):
result = remove_resource_prefix(
"prefix+resource://path/to/resource", "prefix"
)
assert result == "resource://path/to/resource"
# URI without the prefix should be returned as is
result = remove_resource_prefix("resource://path/to/resource", "prefix")
assert result == "resource://path/to/resource"
# Empty prefix should return the original URI
result = remove_resource_prefix("resource://path/to/resource", "")
assert result == "resource://path/to/resource"
def test_has_resource_prefix_legacy(self):
"""Test that has_resource_prefix uses the legacy format when resource_prefix_format is 'protocol'."""
with temporary_settings(resource_prefix_format="protocol"):
result = has_resource_prefix("prefix+resource://path/to/resource", "prefix")
assert result is True
result = has_resource_prefix("resource://path/to/resource", "prefix")
assert result is False
# Empty prefix should always return False
result = has_resource_prefix("resource://path/to/resource", "")
assert result is False
async def test_mount_with_legacy_prefixes():
"""Test mounting a server with legacy resource prefixes."""
with temporary_settings(resource_prefix_format="protocol"):
main_server = FastMCP("MainServer")
sub_server = FastMCP("SubServer")
@sub_server.resource("resource://test")
def get_test():
return "test content"
# Mount the server with a prefix
main_server.mount("sub", sub_server)
# Check that the resource is prefixed using the legacy format
resources = await main_server.get_resources()
# In legacy format, the key would be "sub+resource://test"
assert "sub+resource://test" in resources
# Test accessing the resource through client
async with Client(main_server) as client:
result = await client.read_resource("sub+resource://test")
# Different content types might be returned, but we just want to verify we got something
assert len(result) > 0
async def test_import_server_with_legacy_prefixes():
"""Test importing a server with legacy resource prefixes."""
with temporary_settings(resource_prefix_format="protocol"):
main_server = FastMCP("MainServer")
sub_server = FastMCP("SubServer")
@sub_server.resource("resource://test")
def get_test():
return "test content"
# Import the server with a prefix
await main_server.import_server("sub", sub_server)
# Check that the resource is prefixed using the legacy format
resources = main_server._resource_manager.get_resources()
# In legacy format, the key would be "sub+resource://test"
assert "sub+resource://test" in resources

View file

@ -0,0 +1,65 @@
"""Tests for different resource prefix formats in server mounting and importing."""
from fastmcp import FastMCP
async def test_resource_prefix_format_in_constructor():
"""Test that the resource_prefix_format parameter is respected in the constructor."""
server_path = FastMCP("PathFormat", resource_prefix_format="path")
server_protocol = FastMCP("ProtocolFormat", resource_prefix_format="protocol")
# Check that the format is stored correctly
assert server_path.resource_prefix_format == "path"
assert server_protocol.resource_prefix_format == "protocol"
# Register resources
@server_path.resource("resource://test")
def get_test_path():
return "test content"
@server_protocol.resource("resource://test")
def get_test_protocol():
return "test content"
# Create mount servers
main_server_path = FastMCP("MainPath", resource_prefix_format="path")
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)
# Check that the resources are prefixed correctly
path_resources = await main_server_path.get_resources()
protocol_resources = await main_server_protocol.get_resources()
# Path format should be resource://sub/test
assert "resource://sub/test" in path_resources
# Protocol format should be sub+resource://test
assert "sub+resource://test" in protocol_resources
async def test_resource_prefix_format_in_import_server():
"""Test that the resource_prefix_format parameter is respected in import_server."""
server = FastMCP("TestServer")
@server.resource("resource://test")
def get_test():
return "test content"
# Import with path format
main_server_path = FastMCP("MainPath", resource_prefix_format="path")
await main_server_path.import_server("sub", server)
# Import with protocol format
main_server_protocol = FastMCP("MainProtocol", resource_prefix_format="protocol")
await main_server_protocol.import_server("sub", server)
# Check that the resources are prefixed correctly
path_resources = main_server_path._resource_manager.get_resources()
protocol_resources = main_server_protocol._resource_manager.get_resources()
# Path format should be resource://sub/test
assert "resource://sub/test" in path_resources
# Protocol format should be sub+resource://test
assert "sub+resource://test" in protocol_resources