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

Simplify .key as computed property (#2648)

Browse source 
* Simplify .key as computed property

Keep .key as the standard lookup interface for all components but
implement it as a computed property instead of a stored field.

- Remove _key private attribute and custom model_copy(key=...)
- .key returns .name for tools/prompts, str(.uri) for resources,
  .uri_template for templates
- Use .key universally for component lookups in managers
- MountedProvider: prefix URIs only for resources/templates, not names
- Docket registration: tools/prompts use .key, resources use .name
  (matches fn.__name__ for function lookup)

* Simplify .key as computed property

Keep .key as the standard lookup interface for all components but
implement it as a computed property instead of a stored field.

- Remove _key private attribute and custom model_copy(key=...)
- .key returns .name for tools/prompts, str(.uri) for resources,
  .uri_template for templates
- Use .key universally for component lookups and Docket registration
- MountedProvider: prefix URIs only for resources/templates, not names
- Add _backend_* fields to proxy classes to preserve original identifiers
  for backend calls when prefixed via import_server

* Standardize .key as computed property

- .key is now a read-only computed property:
  - Tools/Prompts: returns .name
  - Resources: returns str(.uri)
  - Templates: returns .uri_template
- Prefixing uses model_copy(update={...}) to change underlying field
- Resource/template names are NOT prefixed, only URIs
- Move import_server tests to tests/deprecated/
This commit is contained in:
Jeremiah Lowin 2025-12-18 14:48:47 -05:00 • committed by GitHub
commit caeaa86394
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 212 additions and 355 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/fastmcp/resources/resource.py
View file

@ -274,13 +274,8 @@ class Resource(FastMCPComponent):
@property
def key(self) -> str:
"""
The key of the component. This is used for internal bookkeeping
and may reflect e.g. prefixes or other identifiers. You should not depend on
keys having a certain value, as the same tool loaded from different
hierarchies of servers may have different keys.
"""
return self._key or str(self.uri)
"""The lookup key for this resource. Returns str(uri)."""
return str(self.uri)
class FunctionResource(Resource):

8
src/fastmcp/resources/resource_manager.py
View file

@ -154,8 +154,8 @@ class ResourceManager:
Args:
resource: A Resource instance to add. The resource's .key attribute
will be used as the storage key. To overwrite it, call
Resource.model_copy(key=new_key) before calling this method.
(which is str(uri)) will be used as the storage key. To use a
different key, change the uri via model_copy(update={"uri": new_uri}).
"""
existing = self._resources.get(resource.key)
if existing:
@ -203,8 +203,8 @@ class ResourceManager:
Args:
template: A ResourceTemplate instance to add. The template's .key attribute
will be used as the storage key. To overwrite it, call
ResourceTemplate.model_copy(key=new_key) before calling this method.
(which is uri_template) will be used as the storage key. To use a
different key, change uri_template via model_copy(update={"uri_template": new_uri}).
Returns:
The added template. If a template with the same URI already exists,

9
src/fastmcp/resources/template.py
View file

@ -219,13 +219,8 @@ class ResourceTemplate(FastMCPComponent):
@property
def key(self) -> str:
"""
The key of the component. This is used for internal bookkeeping
and may reflect e.g. prefixes or other identifiers. You should not depend on
keys having a certain value, as the same tool loaded from different
hierarchies of servers may have different keys.
"""
return self._key or self.uri_template
"""The lookup key for this template. Returns uri_template."""
return self.uri_template
class FunctionResourceTemplate(ResourceTemplate):

42
src/fastmcp/server/providers/mounted.py
View file

@ -179,39 +179,37 @@ class MountedProvider(Provider):
# -------------------------------------------------------------------------
def _prefix_tool(self, tool: Tool) -> Tool:
"""Apply prefix to a tool."""
"""Apply prefix to a tool. Changes name which updates .key."""
if self.tool_names and tool.name in self.tool_names:
new_key = self.tool_names[tool.name]
new_name = self.tool_names[tool.name]
else:
new_key = self._add_tool_prefix(tool.key)
return tool.model_copy(key=new_key) if new_key != tool.key else tool
new_name = self._add_tool_prefix(tool.name)
if new_name != tool.name:
return tool.model_copy(update={"name": new_name})
return tool
def _prefix_resource(self, resource: Resource) -> Resource:
"""Apply prefix to a resource."""
new_key = self._add_resource_prefix(resource.key)
update: dict[str, Any] = {}
if self.prefix and resource.name:
update["name"] = f"{self.prefix}_{resource.name}"
if new_key != resource.key or update:
return resource.model_copy(key=new_key, update=update)
"""Apply prefix to a resource URI (name is NOT prefixed)."""
if self.prefix:
new_uri = self._add_resource_prefix(str(resource.uri))
if new_uri != str(resource.uri):
return resource.model_copy(update={"uri": new_uri})
return resource
def _prefix_template(self, template: ResourceTemplate) -> ResourceTemplate:
"""Apply prefix to a resource template."""
new_key = self._add_resource_prefix(template.key)
update: dict[str, Any] = {}
if self.prefix and template.name:
update["name"] = f"{self.prefix}_{template.name}"
"""Apply prefix to a resource template URI (name is NOT prefixed)."""
if self.prefix and template.uri_template:
update["uri_template"] = self._add_resource_prefix(template.uri_template)
if new_key != template.key or update:
return template.model_copy(key=new_key, update=update)
new_template = self._add_resource_prefix(template.uri_template)
if new_template != template.uri_template:
return template.model_copy(update={"uri_template": new_template})
return template
def _prefix_prompt(self, prompt: Prompt) -> Prompt:
"""Apply prefix to a prompt."""
new_key = self._add_tool_prefix(prompt.key)
return prompt.model_copy(key=new_key) if new_key != prompt.key else prompt
"""Apply prefix to a prompt. Changes name which updates .key."""
new_name = self._add_tool_prefix(prompt.name)
if new_name != prompt.name:
return prompt.model_copy(update={"name": new_name})
return prompt
# -------------------------------------------------------------------------
# Tool methods

56
src/fastmcp/server/proxy.py
View file

@ -280,11 +280,21 @@ class ProxyTool(Tool, MirroredComponent):
"""
task_config: TaskConfig = TaskConfig(mode="forbidden")
_backend_name: str | None = None
def __init__(self, client: Client, **kwargs: Any):
super().__init__(**kwargs)
self._client = client
def model_copy(self, **kwargs: Any) -> ProxyTool:
"""Override to preserve _backend_name when name changes."""
update = kwargs.get("update", {})
if "name" in update and self._backend_name is None:
# First time name is being changed, preserve original for backend calls
update = {**update, "_backend_name": self.name}
kwargs["update"] = update
return super().model_copy(**kwargs) # type: ignore[return-value]
@classmethod
def from_mcp_tool(cls, client: Client, mcp_tool: mcp.types.Tool) -> ProxyTool:
"""Factory method to create a ProxyTool from a raw MCP tool schema."""
@ -331,7 +341,7 @@ class ProxyTool(Tool, MirroredComponent):
)
result = await self._client.call_tool_mcp(
name=self.name, arguments=arguments, meta=meta
name=self._backend_name or self.name, arguments=arguments, meta=meta
)
if result.isError:
raise ToolError(cast(mcp.types.TextContent, result.content[0]).text)
@ -351,6 +361,7 @@ class ProxyResource(Resource, MirroredComponent):
task_config: TaskConfig = TaskConfig(mode="forbidden")
_client: Client
_cached_content: ResourceContent | None = None
_backend_uri: str | None = None
def __init__(
self,
@ -363,6 +374,15 @@ class ProxyResource(Resource, MirroredComponent):
self._client = client
self._cached_content = _cached_content
def model_copy(self, **kwargs: Any) -> ProxyResource:
"""Override to preserve _backend_uri when uri changes."""
update = kwargs.get("update", {})
if "uri" in update and self._backend_uri is None:
# First time uri is being changed, preserve original for backend calls
update = {**update, "_backend_uri": str(self.uri)}
kwargs["update"] = update
return super().model_copy(**kwargs) # type: ignore[return-value]
@classmethod
def from_mcp_resource(
cls,
@ -390,10 +410,13 @@ class ProxyResource(Resource, MirroredComponent):
if self._cached_content is not None:
return self._cached_content
backend_uri = self._backend_uri or str(self.uri)
async with self._client:
result = await self._client.read_resource(self.uri)
result = await self._client.read_resource(backend_uri)
if not result:
raise ResourceError(f"Remote server returned empty content for {self.uri}")
raise ResourceError(
f"Remote server returned empty content for {backend_uri}"
)
if isinstance(result[0], TextResourceContents):
return ResourceContent(
content=result[0].text,
@ -416,11 +439,21 @@ class ProxyTemplate(ResourceTemplate, MirroredComponent):
"""
task_config: TaskConfig = TaskConfig(mode="forbidden")
_backend_uri_template: str | None = None
def __init__(self, client: Client, **kwargs: Any):
super().__init__(**kwargs)
self._client = client
def model_copy(self, **kwargs: Any) -> ProxyTemplate:
"""Override to preserve _backend_uri_template when uri_template changes."""
update = kwargs.get("update", {})
if "uri_template" in update and self._backend_uri_template is None:
# First time uri_template is being changed, preserve original for backend
update = {**update, "_backend_uri_template": self.uri_template}
kwargs["update"] = update
return super().model_copy(**kwargs) # type: ignore[return-value]
@classmethod
def from_mcp_template( # type: ignore[override]
cls, client: Client, mcp_template: mcp.types.ResourceTemplate
@ -451,7 +484,8 @@ class ProxyTemplate(ResourceTemplate, MirroredComponent):
# don't use the provided uri, because it may not be the same as the
# uri_template on the remote server.
# quote params to ensure they are valid for the uri_template
parameterized_uri = self.uri_template.format(
backend_template = self._backend_uri_template or self.uri_template
parameterized_uri = backend_template.format(
**{k: quote(v, safe="") for k, v in params.items()}
)
async with self._client:
@ -497,11 +531,21 @@ class ProxyPrompt(Prompt, MirroredComponent):
task_config: TaskConfig = TaskConfig(mode="forbidden")
_client: Client
_backend_name: str | None = None
def __init__(self, client: Client, **kwargs):
super().__init__(**kwargs)
self._client = client
def model_copy(self, **kwargs: Any) -> ProxyPrompt:
"""Override to preserve _backend_name when name changes."""
update = kwargs.get("update", {})
if "name" in update and self._backend_name is None:
# First time name is being changed, preserve original for backend calls
update = {**update, "_backend_name": self.name}
kwargs["update"] = update
return super().model_copy(**kwargs) # type: ignore[return-value]
@classmethod
def from_mcp_prompt(
cls, client: Client, mcp_prompt: mcp.types.Prompt
@ -531,7 +575,9 @@ class ProxyPrompt(Prompt, MirroredComponent):
async def render(self, arguments: dict[str, Any]) -> PromptResult: # type: ignore[override]
"""Render the prompt by making a call through the client."""
async with self._client:
result = await self._client.get_prompt(self.name, arguments)
result = await self._client.get_prompt(
self._backend_name or self.name, arguments
)
# Convert GetPromptResult to PromptResult, preserving runtime meta from the result
# (not the static prompt meta which includes fastmcp tags)
return PromptResult(

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

@ -442,14 +442,14 @@ class FastMCP(Generic[LifespanResultT]):
isinstance(resource, FunctionResource)
and resource.task_config.mode != "forbidden"
):
docket.register(resource.fn, names=[resource.name])
docket.register(resource.fn, names=[resource.key])
for template in self._resource_manager._templates.values():
if (
isinstance(template, FunctionResourceTemplate)
and template.task_config.mode != "forbidden"
):
docket.register(template.fn, names=[template.name])
docket.register(template.fn, names=[template.key])
# Register provider components
for provider in self._providers:
@ -458,9 +458,9 @@ class FastMCP(Generic[LifespanResultT]):
for tool in tasks.tools:
docket.register(tool.fn, names=[tool.key])
for resource in tasks.resources:
docket.register(resource.fn, names=[resource.name])
docket.register(resource.fn, names=[resource.key])
for template in tasks.templates:
docket.register(template.fn, names=[template.name])
docket.register(template.fn, names=[template.key])
for prompt in tasks.prompts:
docket.register(
cast(Callable[..., Awaitable[Any]], prompt.fn),
@ -2755,32 +2755,30 @@ class FastMCP(Generic[LifespanResultT]):
)
# Import tools from the server
for key, tool in (await server.get_tools()).items():
for tool in (await server.get_tools()).values():
if prefix:
tool = tool.model_copy(key=f"{prefix}_{key}")
tool = tool.model_copy(update={"name": f"{prefix}_{tool.name}"})
self._tool_manager.add_tool(tool)
# Import resources and templates from the server
for key, resource in (await server.get_resources()).items():
for resource in (await server.get_resources()).values():
if prefix:
resource_key = add_resource_prefix(key, prefix)
resource = resource.model_copy(
update={"name": f"{prefix}_{resource.name}"}, key=resource_key
)
new_uri = add_resource_prefix(str(resource.uri), prefix)
resource = resource.model_copy(update={"uri": new_uri})
self._resource_manager.add_resource(resource)
for key, template in (await server.get_resource_templates()).items():
for template in (await server.get_resource_templates()).values():
if prefix:
template_key = add_resource_prefix(key, prefix)
new_uri_template = add_resource_prefix(template.uri_template, prefix)
template = template.model_copy(
update={"name": f"{prefix}_{template.name}"}, key=template_key
update={"uri_template": new_uri_template}
)
self._resource_manager.add_template(template)
# Import prompts from the server
for key, prompt in (await server.get_prompts()).items():
for prompt in (await server.get_prompts()).values():
if prefix:
prompt = prompt.model_copy(key=f"{prefix}_{key}")
prompt = prompt.model_copy(update={"name": f"{prefix}_{prompt.name}"})
self._prompt_manager.add_prompt(prompt)
if server._lifespan != default_lifespan:

11
src/fastmcp/server/tasks/handlers.py
View file

@ -99,7 +99,7 @@ async def handle_tool_as_task(
# Don't let notification failures break task creation
await ctx.session.send_notification(notification) # type: ignore[arg-type]
# Queue function to Docket by name (result storage via execution_ttl)
# Queue function to Docket by key (result storage via execution_ttl)
# Use tool.key which matches what was registered - prefixed for mounted tools
await docket.add(
tool.key,
@ -205,7 +205,7 @@ async def handle_prompt_as_task(
with suppress(Exception):
await ctx.session.send_notification(notification) # type: ignore[arg-type]
# Queue function to Docket by name (result storage via execution_ttl)
# Queue function to Docket by key (result storage via execution_ttl)
# Use prompt.key which matches what was registered - prefixed for mounted prompts
await docket.add(
prompt.key,
@ -309,20 +309,19 @@ async def handle_resource_as_task(
with suppress(Exception):
await ctx.session.send_notification(notification) # type: ignore[arg-type]
# Queue function to Docket by name (result storage via execution_ttl)
# Use resource.name which matches what was registered - prefixed for mounted resources
# Queue function to Docket by key (result storage via execution_ttl)
# For templates, extract URI params and pass them to the function
from fastmcp.resources.template import FunctionResourceTemplate, match_uri_template
if isinstance(resource, FunctionResourceTemplate):
params = match_uri_template(uri, resource.uri_template) or {}
await docket.add(
resource.name,
resource.key,
key=task_key,
)(**params)
else:
await docket.add(
resource.name,
resource.key,
key=task_key,
)()

44
src/fastmcp/utilities/components.py
View file

@ -1,7 +1,7 @@
from __future__ import annotations
from collections.abc import Sequence
from typing import Annotated, Any, TypedDict, cast
from typing import Annotated, Any, TypedDict
from mcp.types import Icon
from pydantic import BeforeValidator, Field, PrivateAttr
@ -56,21 +56,16 @@ class FastMCPComponent(FastMCPBaseModel):
description="Whether the component is enabled.",
)
_key: str | None = PrivateAttr()
def __init__(self, *, key: str | None = None, **kwargs: Any) -> None:
super().__init__(**kwargs)
self._key = key
@property
def key(self) -> str:
"""The lookup key for this component. Returns name by default.
Subclasses override this to return different identifiers:
- Tools/Prompts: name
- Resources: str(uri)
- Templates: uri_template
"""
The key of the component. This is used for internal bookkeeping
and may reflect e.g. prefixes or other identifiers. You should not depend on
keys having a certain value, as the same tool loaded from different
hierarchies of servers may have different keys.
"""
return self._key or self.name
return self.name
def get_meta(
self, include_fastmcp_meta: bool | None = None
@ -96,29 +91,6 @@ class FastMCPComponent(FastMCPBaseModel):
return meta or None
def model_copy( # type: ignore[override]
self,
*,
update: dict[str, Any] | None = None,
deep: bool = False,
key: str | None = None,
) -> Self:
"""
Create a copy of the component.
Args:
update: A dictionary of fields to update.
deep: Whether to deep copy the component.
key: The key to use for the copy.
"""
# `model_copy` has an `update` parameter but it doesn't work for certain private attributes
# https://github.com/pydantic/pydantic/issues/12116
# So we manually set the private attribute here instead, such as _key
copy = super().model_copy(update=update, deep=deep)
if key is not None:
copy._key = key
return cast(Self, copy)
def __eq__(self, other: object) -> bool:
if type(self) is not type(other):
return False

5
tests/deprecated/__init__.py
View file

@ -1,4 +1 @@
import pytest
# reset deprecation warnings for this module
pytestmark = pytest.mark.filterwarnings("default::DeprecationWarning")
# Tests for deprecated features - deprecation warnings are suppressed via conftest.py

9
tests/deprecated/conftest.py Normal file
View file

@ -0,0 +1,9 @@
"""Conftest for deprecated tests - suppresses deprecation warnings."""
import pytest
def pytest_collection_modifyitems(items):
"""Add filterwarnings marker to all tests in this directory."""
for item in items:
item.add_marker(pytest.mark.filterwarnings("ignore::DeprecationWarning"))

60
tests/server/test_import_server.py → tests/deprecated/test_import_server.py
View file

@ -28,7 +28,8 @@ async def test_import_basic_functionality():
# Verify the original tool still exists in the sub-app
tool = await main_app._tool_manager.get_tool("sub_sub_tool")
assert tool is not None
assert tool.name == "sub_tool"
# 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)
@ -589,8 +590,8 @@ async def test_import_conflict_resolution_with_prefix():
assert result.data == "Second app tool"
async def test_import_server_resource_name_prefixing():
"""Test that resource names are prefixed when using import_server."""
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")
@ -602,14 +603,14 @@ async def test_import_server_resource_name_prefixing():
main_server = FastMCP("MainServer")
await main_server.import_server(sub_server, prefix="imported")
# Get resources and verify name prefixing
# Get resources and verify URI prefixing (name should NOT be prefixed)
resources = await main_server.get_resources()
resource = resources["resource://imported/test_resource"]
assert resource.name == "imported_test_resource"
assert resource.name == "test_resource"
async def test_import_server_resource_template_name_prefixing():
"""Test that resource template names are prefixed when using import_server."""
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")
@ -621,7 +622,48 @@ async def test_import_server_resource_template_name_prefixing():
main_server = FastMCP("MainServer")
await main_server.import_server(sub_server, prefix="imported")
# Get resource templates and verify name prefixing
# Get resource templates and verify URI prefixing (name should NOT be prefixed)
templates = await main_server.get_resource_templates()
template = templates["resource://imported/data/{item_id}"]
assert template.name == "imported_data_template"
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.get_resources()
templates = await target_server.get_resource_templates()
assert "resource://imported/test-resource" in resources
assert "resource://imported//absolute/path" in resources
assert "resource://imported/{param}/template" 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 result[0].text == "Resource content" # type: ignore[attr-defined]
result = await client.read_resource("resource://imported//absolute/path")
assert result[0].text == "Absolute resource content" # type: ignore[attr-defined]
result = await client.read_resource("resource://imported/param-value/template")
assert result[0].text == "Template resource with param-value" # type: ignore[attr-defined]

117
tests/resources/test_resource_manager.py
View file

@ -450,123 +450,6 @@ class TestResourceTags:
assert admin_resources[0].name == "system_data"
class TestCustomResourceKeys:
"""Test adding resources and templates with custom keys."""
def test_add_resource_with_custom_key(self, temp_file: Path):
"""Test adding a resource with a custom key different from its URI."""
manager = ResourceManager()
original_uri = "data://test/resource"
custom_key = "custom://resource/key"
# Create a function resource instead of file resource to avoid path issues
async def get_data():
return "Test data"
resource = FunctionResource(
uri=AnyUrl(original_uri),
name="test_resource",
fn=get_data,
)
# Use model_copy to create a new resource with the custom key
resource_with_custom_key = resource.model_copy(key=custom_key)
manager.add_resource(resource_with_custom_key)
# Resource should be accessible via custom key
assert custom_key in manager._resources
# But not via its original URI
assert original_uri not in manager._resources
# The resource's internal URI remains unchanged
assert str(manager._resources[custom_key].uri) == original_uri
def test_add_template_with_custom_key(self):
"""Test adding a template with a custom key different from its URI template."""
manager = ResourceManager()
def template_fn(id: str) -> str:
return f"Template {id}"
original_uri_template = "test://{id}"
custom_key = "custom://{id}/template"
template = ResourceTemplate.from_function(
fn=template_fn,
uri_template=original_uri_template,
name="test_template",
)
# Use model_copy to create a new template with the custom key
template_with_custom_key = template.model_copy(key=custom_key)
manager.add_template(template_with_custom_key)
# Template should be accessible via custom key
assert custom_key in manager._templates
# But not via its original URI template
assert original_uri_template not in manager._templates
# The template's internal URI template remains unchanged
assert str(manager._templates[custom_key].uri_template) == original_uri_template
async def test_get_resource_with_custom_key(self, temp_file: Path):
"""Test that get_resource works with resources added with custom keys."""
manager = ResourceManager()
original_uri = "data://test/resource"
custom_key = "custom://resource/path"
# Create a function resource instead of file resource to avoid path issues
async def get_data():
return "Test data"
resource = FunctionResource(
uri=AnyUrl(original_uri),
name="test_resource",
fn=get_data,
)
# Use model_copy to create a new resource with the custom key
resource_with_custom_key = resource.model_copy(key=custom_key)
manager.add_resource(resource_with_custom_key)
# Should be retrievable by the custom key
retrieved = await manager.get_resource(custom_key)
assert retrieved is not None
assert str(retrieved.uri) == original_uri
# Should NOT be retrievable by the original URI
with pytest.raises(NotFoundError, match="Unknown resource"):
await manager.get_resource(original_uri)
async def test_get_resource_from_template_with_custom_key(self):
"""Test that templates with custom keys can create resources."""
manager = ResourceManager()
def greet(name: str) -> str:
return f"Hello, {name}!"
original_template = "greet://{name}"
custom_key = "custom://greet/{name}"
template = ResourceTemplate.from_function(
fn=greet,
uri_template=original_template,
name="custom_greeter",
)
# Use model_copy to create a new template with the custom key
template_with_custom_key = template.model_copy(key=custom_key)
manager.add_template(template_with_custom_key)
# Using a URI that matches the custom key pattern
resource = await manager.get_resource("custom://greet/world")
assert isinstance(resource, FunctionResource)
result = await resource.read()
assert result.content == "Hello, world!"
# Shouldn't work with the original template pattern
with pytest.raises(NotFoundError, match="Unknown resource"):
await manager.get_resource("greet://world")
class TestQueryOnlyTemplates:
"""Test resource templates with only query parameters (no path params)."""

22
tests/server/tasks/test_server_tasks_parameter.py
View file

@ -28,11 +28,12 @@ async def test_server_tasks_true_defaults_all_components():
async with Client(mcp) as client:
# Verify all task-enabled components are registered with docket
# Tools and prompts use .key (which equals name), resources use .key (which is URI)
docket = mcp.docket
assert docket is not None
assert "my_tool" in docket.tasks
assert "my_prompt" in docket.tasks
assert "my_resource" in docket.tasks
assert "test://resource" in docket.tasks
# Tool should support background execution
tool_task = await client.call_tool("my_tool", task=True)
@ -198,17 +199,18 @@ async def test_mixed_explicit_and_inherited():
async with Client(mcp) as client:
# Verify docket registration matches task settings
# Tools/prompts use .key (name), resources use .key (URI)
docket = mcp.docket
assert docket is not None
# task=True (explicit or inherited) means registered
assert "inherited_tool" in docket.tasks
assert "explicit_true_tool" in docket.tasks
assert "inherited_prompt" in docket.tasks
assert "inherited_resource" in docket.tasks
assert "test://inherited" in docket.tasks
# task=False means NOT registered
assert "explicit_false_tool" not in docket.tasks
assert "explicit_false_prompt" not in docket.tasks
assert "explicit_false_resource" not in docket.tasks
assert "test://explicit_false" not in docket.tasks
# Tools
inherited = await client.call_tool("inherited_tool", task=True)
@ -339,8 +341,7 @@ async def test_task_with_custom_tool_name():
async def test_task_with_custom_resource_name():
"""Resources with custom names work correctly as tasks.
When a resource is registered with a custom name different from the function
name, task execution should use the custom name for Docket lookup.
Resources are registered/looked up by their .key (URI), not their name.
"""
mcp = FastMCP("test", tasks=True)
@ -349,10 +350,10 @@ async def test_task_with_custom_resource_name():
return "result from custom-named resource"
async with Client(mcp) as client:
# Verify the resource is registered with its custom name in Docket
# Verify the resource is registered with its key (URI) in Docket
docket = mcp.docket
assert docket is not None
assert "custom-resource-name" in docket.tasks
assert "test://resource" in docket.tasks
# Call the resource as a task
task = await client.read_resource("test://resource", task=True)
@ -364,8 +365,7 @@ async def test_task_with_custom_resource_name():
async def test_task_with_custom_template_name():
"""Resource templates with custom names work correctly as tasks.
When a template is registered with a custom name different from the function
name, task execution should use the custom name for Docket lookup.
Templates are registered/looked up by their .key (uri_template), not their name.
"""
mcp = FastMCP("test", tasks=True)
@ -374,10 +374,10 @@ async def test_task_with_custom_template_name():
return f"result for {item_id}"
async with Client(mcp) as client:
# Verify the template is registered with its custom name in Docket
# Verify the template is registered with its key (uri_template) in Docket
docket = mcp.docket
assert docket is not None
assert "custom-template-name" in docket.tasks
assert "test://{item_id}" in docket.tasks
# Call the template as a task
task = await client.read_resource("test://123", task=True)

20
tests/server/test_mount.py
View file

@ -970,11 +970,11 @@ class TestAsProxyKwarg:
assert lifespan_check == ["start"]
class TestResourceNamePrefixing:
"""Test that resource and resource template names get prefixed when mounted."""
class TestResourceUriPrefixing:
"""Test that resource and resource template URIs get prefixed when mounted (names are NOT prefixed)."""
async def test_resource_name_prefixing(self):
"""Test that resource names are prefixed when mounted."""
async def test_resource_uri_prefixing(self):
"""Test that resource URIs are prefixed when mounted (names are NOT prefixed)."""
# Create a sub-app with a resource
sub_app = FastMCP("SubApp")
@ -993,12 +993,12 @@ class TestResourceNamePrefixing:
# Should have prefixed key (using path format: resource://prefix/resource_name)
assert "resource://prefix/my_resource" in resources
# The resource name should also be prefixed
# The resource name should NOT be prefixed (only URI is prefixed)
resource = resources["resource://prefix/my_resource"]
assert resource.name == "prefix_my_resource"
assert resource.name == "my_resource"
async def test_resource_template_name_prefixing(self):
"""Test that resource template names are prefixed when mounted."""
async def test_resource_template_uri_prefixing(self):
"""Test that resource template URIs are prefixed when mounted (names are NOT prefixed)."""
# Create a sub-app with a resource template
sub_app = FastMCP("SubApp")
@ -1017,9 +1017,9 @@ class TestResourceNamePrefixing:
# Should have prefixed key (using path format: resource://prefix/template_uri)
assert "resource://prefix/user/{user_id}" in templates
# The template name should also be prefixed
# The template name should NOT be prefixed (only URI template is prefixed)
template = templates["resource://prefix/user/{user_id}"]
assert template.name == "prefix_user_template"
assert template.name == "user_template"
class TestParentTagFiltering:

42
tests/server/test_server.py
View file

@ -1335,48 +1335,6 @@ class TestResourcePrefixMounting:
# Test stripping
assert remove_resource_prefix(uri, prefix) == expected_strip
async def test_import_server_with_new_prefix_format(self):
"""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.get_resources()
templates = await target_server.get_resource_templates()
assert "resource://imported/test-resource" in resources
assert "resource://imported//absolute/path" in resources
assert "resource://imported/{param}/template" 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 result[0].text == "Resource content" # type: ignore[attr-defined]
result = await client.read_resource("resource://imported//absolute/path")
assert result[0].text == "Absolute resource content" # type: ignore[attr-defined]
result = await client.read_resource(
"resource://imported/param-value/template"
)
assert result[0].text == "Template resource with param-value" # type: ignore[attr-defined]
class TestShouldIncludeComponent:
def test_no_filters_returns_true(self):

21
tests/tools/test_tool_manager.py
View file

@ -869,27 +869,6 @@ class TestCustomToolNames:
with pytest.raises(NotFoundError, match="Tool 'original_fn' not found"):
await manager.get_tool("original_fn")
async def test_add_tool_object_with_custom_key(self):
"""Test adding a Tool object with a custom key using add_tool()."""
def fn(x: int) -> int:
return x + 1
# Create a tool with a specific name
tool = Tool.from_function(fn, name="my_tool")
manager = ToolManager()
# Use model_copy to create a new tool with the custom key
tool_with_custom_key = tool.model_copy(key="proxy_tool")
manager.add_tool(tool_with_custom_key)
# The tool is accessible under the key
stored = await manager.get_tool("proxy_tool")
assert stored is not None
# But the tool's .name is unchanged
assert stored.name == "my_tool"
# The tool is not accessible under its original name
with pytest.raises(NotFoundError, match="Tool 'my_tool' not found"):
await manager.get_tool("my_tool")
async def test_call_tool_with_custom_name(self):
"""Test calling a tool added with a custom name."""

62
tests/utilities/test_components.py
View file

@ -48,7 +48,7 @@ class TestFastMCPComponent:
name="test_component",
title="Test Component",
description="A test component",
tags=["test", "component"],
tags={"test", "component"},
)
def test_initialization_with_minimal_params(self):
@ -68,7 +68,7 @@ class TestFastMCPComponent:
name="full",
title="Full Component",
description="A fully configured component",
tags=["tag1", "tag2"],
tags={"tag1", "tag2"},
meta=meta,
enabled=False,
)
@ -83,12 +83,6 @@ class TestFastMCPComponent:
"""Test that key property returns name when no custom key is set."""
assert basic_component.key == "test_component"
def test_key_property_with_custom_key(self):
"""Test that key property returns custom key when set."""
component = FastMCPComponent(name="test", key="custom_key")
assert component.key == "custom_key"
assert component.name == "test"
def test_get_meta_without_fastmcp_meta(self, basic_component):
"""Test get_meta without including fastmcp meta."""
basic_component.meta = {"custom": "data"}
@ -110,7 +104,7 @@ class TestFastMCPComponent:
component = FastMCPComponent(
name="test",
meta={"_fastmcp": {"existing": "value"}},
tags=["new_tag"],
tags={"new_tag"},
)
result = component.get_meta(include_fastmcp_meta=True)
assert result is not None
@ -123,14 +117,6 @@ class TestFastMCPComponent:
result = component.get_meta(include_fastmcp_meta=False)
assert result is None
def test_model_copy_creates_copy_with_new_key(self, basic_component):
"""Test that model_copy with key creates a copy with a new key."""
new_component = basic_component.model_copy(key="new_key")
assert new_component.key == "new_key"
assert new_component.name == basic_component.name
assert new_component is not basic_component # Should be a copy
assert basic_component.key == "test_component" # Original unchanged
def test_equality_same_components(self):
"""Test that identical components are equal."""
comp1 = FastMCPComponent(name="test", description="desc")
@ -182,17 +168,17 @@ class TestFastMCPComponent:
assert basic_component.name == "test_component"
def test_tags_deduplication(self):
"""Test that tags are deduplicated."""
"""Test that tags are deduplicated when passed as a sequence."""
component = FastMCPComponent(
name="test",
tags=["tag1", "tag2", "tag1", "tag2"],
tags=["tag1", "tag2", "tag1", "tag2"], # type: ignore[arg-type]
)
assert component.tags == {"tag1", "tag2"}
def test_validation_error_for_invalid_data(self):
"""Test that validation errors are raised for invalid data."""
with pytest.raises(ValidationError):
FastMCPComponent() # Missing required name field
FastMCPComponent() # type: ignore[call-arg]
def test_extra_fields_forbidden(self):
"""Test that extra fields are not allowed."""
@ -290,10 +276,6 @@ class TestMirroredComponent:
# Test key property
assert mirrored_component.key == "mirrored"
# Test model_copy with key
with_key = mirrored_component.model_copy(key="new_key")
assert with_key.key == "new_key"
# Test get_meta
mirrored_component.tags = {"tag1"}
meta = mirrored_component.get_meta(include_fastmcp_meta=True)
@ -323,7 +305,7 @@ class TestEdgeCasesAndIntegration:
def test_empty_tags_conversion(self):
"""Test that empty tags are handled correctly."""
component = FastMCPComponent(name="test", tags=[])
component = FastMCPComponent(name="test", tags=set())
assert component.tags == set()
def test_tags_with_none_values(self):
@ -353,17 +335,17 @@ class TestEdgeCasesAndIntegration:
component = FastMCPComponent(name="test", meta=complex_meta)
assert component.meta == complex_meta
def test_model_copy_with_key_preserves_all_attributes(self):
"""Test that model_copy with key preserves all component attributes."""
def test_model_copy_preserves_all_attributes(self):
"""Test that model_copy preserves all component attributes."""
component = FastMCPComponent(
name="test",
title="Title",
description="Description",
tags=["tag1", "tag2"],
tags={"tag1", "tag2"},
meta={"key": "value"},
enabled=False,
)
new_component = component.model_copy(key="new_key")
new_component = component.model_copy()
assert new_component.name == component.name
assert new_component.title == component.title
@ -371,7 +353,7 @@ class TestEdgeCasesAndIntegration:
assert new_component.tags == component.tags
assert new_component.meta == component.meta
assert new_component.enabled == component.enabled
assert new_component.key == "new_key"
assert new_component.key == component.key
def test_mirrored_component_copy_chain(self):
"""Test creating copies of copies for mirrored components."""
@ -390,30 +372,34 @@ class TestEdgeCasesAndIntegration:
assert copy1.name == "copy1"
assert copy2.name == "copy2"
def test_model_copy_with_update_and_key(self):
"""Test that model_copy works with both update dict and key parameter."""
def test_model_copy_with_update(self):
"""Test that model_copy works with update dict."""
component = FastMCPComponent(
name="test",
title="Original Title",
description="Original Description",
tags=["tag1"],
tags={"tag1"},
enabled=True,
)
# Test with both update and key
# Test with update (including name which affects .key)
updated_component = component.model_copy(
update={"title": "New Title", "description": "New Description"},
key="new_key",
update={
"name": "new_name",
"title": "New Title",
"description": "New Description",
},
)
assert updated_component.name == "test" # Not in update, unchanged
assert updated_component.name == "new_name" # Updated
assert updated_component.title == "New Title" # Updated
assert updated_component.description == "New Description" # Updated
assert updated_component.tags == {"tag1"} # Not in update, unchanged
assert updated_component.enabled is True # Not in update, unchanged
assert updated_component.key == "new_key" # Custom key set
assert updated_component.key == "new_name" # .key is computed from name
# Original should be unchanged
assert component.name == "test"
assert component.title == "Original Title"
assert component.description == "Original Description"
assert component.key == "test" # Uses name as key