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

Add key to component

Browse source
This commit is contained in:
Jeremiah Lowin 2025-06-18 21:05:11 -04:00
commit a2bb2314c7
6 changed files with 81 additions and 85 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -101,6 +101,16 @@ class Resource(FastMCPComponent, abc.ABC):
def __repr__(self) -> str:
return f"{self.__class__.__name__}(uri={self.uri!r}, name={self.name!r}, description={self.description!r}, tags={self.tags})"
@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)
class FunctionResource(Resource):
"""A resource that defers data loading by wrapping a function.

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

@ -128,6 +128,16 @@ class ResourceTemplate(FastMCPComponent):
}
return MCPResourceTemplate(**kwargs | overrides)
@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
class FunctionResourceTemplate(ResourceTemplate):
"""A template for dynamically creating resources."""

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

@ -330,11 +330,11 @@ class FastMCP(Generic[LifespanResultT]):
server_tools = await mounted_server.server.get_tools()
# Apply prefix to each tool key if prefix exists and is not empty
if mounted_server.prefix:
server_tools = {
f"{mounted_server.prefix}_{key}": tool
for key, tool in server_tools.items()
}
tools.update(server_tools)
for tool in server_tools.values():
tool = tool.with_key(f"{mounted_server.prefix}_{tool.key}")
tools[tool.key] = tool
else:
tools.update(server_tools)
except Exception as e:
logger.warning(
f"Failed to get tools from mounted server '{mounted_server.prefix}': {e}"
@ -361,15 +361,17 @@ class FastMCP(Generic[LifespanResultT]):
server_resources = await mounted_server.server.get_resources()
# Apply prefix to each resource key if prefix exists
if mounted_server.prefix:
server_resources = {
add_resource_prefix(
key,
mounted_server.prefix,
mounted_server.server.resource_prefix_format,
): resource
for key, resource in server_resources.items()
}
resources.update(server_resources)
for resource in server_resources.values():
resource = resource.with_key(
add_resource_prefix(
resource.key,
mounted_server.prefix,
self.resource_prefix_format,
)
)
resources[resource.key] = resource
else:
resources.update(server_resources)
except Exception as e:
logger.warning(
f"Failed to get resources from mounted server '{mounted_server.prefix}': {e}"
@ -400,15 +402,17 @@ class FastMCP(Generic[LifespanResultT]):
)
# Apply prefix to each template key if prefix exists
if mounted_server.prefix:
server_templates = {
add_resource_prefix(
key,
mounted_server.prefix,
mounted_server.server.resource_prefix_format,
): template
for key, template in server_templates.items()
}
templates.update(server_templates)
for template in server_templates.values():
template = template.with_key(
add_resource_prefix(
template.key,
mounted_server.prefix,
self.resource_prefix_format,
)
)
templates[template.key] = template
else:
templates.update(server_templates)
except Exception as e:
logger.warning(
"Failed to get resource templates from mounted server "
@ -429,6 +433,7 @@ class FastMCP(Generic[LifespanResultT]):
"""
List all available prompts.
"""
if (prompts := self._cache.get("prompts")) is self._cache.NOT_FOUND:
prompts: dict[str, Prompt] = {}
@ -438,11 +443,13 @@ class FastMCP(Generic[LifespanResultT]):
server_prompts = await mounted_server.server.get_prompts()
# Apply prefix to each prompt key if prefix exists
if mounted_server.prefix:
server_prompts = {
f"{mounted_server.prefix}_{key}": prompt
for key, prompt in server_prompts.items()
}
prompts.update(server_prompts)
for prompt in server_prompts.values():
prompt = prompt.with_key(
f"{mounted_server.prefix}_{prompt.key}"
)
prompts[prompt.key] = prompt
else:
prompts.update(server_prompts)
except Exception as e:
logger.warning(
f"Failed to get prompts from mounted server '{mounted_server.prefix}': {e}"
@ -524,6 +531,7 @@ class FastMCP(Generic[LifespanResultT]):
"""
resources = await self.get_resources()
mcp_resources: list[MCPResource] = []
for key, resource in resources.items():
if self._should_enable_component(resource):
@ -666,12 +674,12 @@ class FastMCP(Generic[LifespanResultT]):
if has_resource_prefix(
str(resource_uri),
mounted_server.prefix,
mounted_server.server.resource_prefix_format,
self.resource_prefix_format,
):
resource_uri = remove_resource_prefix(
str(resource_uri),
mounted_server.prefix,
mounted_server.server.resource_prefix_format,
self.resource_prefix_format,
)
else:
continue

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

@ -1,7 +1,8 @@
from collections.abc import Sequence
from typing import Annotated, TypeVar
from typing import Annotated, Any, TypeVar
from pydantic import BeforeValidator, Field
from pydantic import BeforeValidator, Field, PrivateAttr
from typing_extensions import Self
from fastmcp.utilities.types import FastMCPBaseModel
@ -37,6 +38,25 @@ 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 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
def with_key(self, key: str) -> Self:
return self.model_copy(update={"_key": key})
def __eq__(self, other: object) -> bool:
if type(self) is not type(other):
return False