refactor utilities

src/fastmcp/server/context.py

@@ -17,8 +17,8 @@ from pydantic import BaseModel, ConfigDict
 from pydantic.networks import AnyUrl
 from starlette.requests import Request
 
+from fastmcp.server.http import get_current_starlette_request
 from fastmcp.server.server import FastMCP
-from fastmcp.utilities.http import get_current_starlette_request
 from fastmcp.utilities.logging import get_logger
 
 logger = get_logger(__name__)

src/fastmcp/server/server.py

@@ -56,10 +56,11 @@ from fastmcp.prompts import Prompt, PromptManager
 from fastmcp.prompts.prompt import PromptResult
 from fastmcp.resources import Resource, ResourceManager
 from fastmcp.resources.template import ResourceTemplate
+from fastmcp.server.http import RequestMiddleware
 from fastmcp.tools import ToolManager
 from fastmcp.tools.tool import Tool
+from fastmcp.utilities.cache import TimedCache
 from fastmcp.utilities.decorators import DecoratedFunction
-from fastmcp.utilities.http import RequestMiddleware
 from fastmcp.utilities.logging import configure_logging, get_logger
 
 if TYPE_CHECKING:
@@ -70,99 +71,6 @@ if TYPE_CHECKING:
 
 logger = get_logger(__name__)
 
-NOT_FOUND = object()
-
-
-class MountedServer:
-    def __init__(
-        self,
-        prefix: str,
-        server: FastMCP,
-        tool_separator: str | None = None,
-        resource_separator: str | None = None,
-        prompt_separator: str | None = None,
-    ):
-        if tool_separator is None:
-            tool_separator = "_"
-        if resource_separator is None:
-            resource_separator = "+"
-        if prompt_separator is None:
-            prompt_separator = "_"
-
-        _validate_resource_prefix(f"{prefix}{resource_separator}")
-
-        self.server = server
-        self.prefix = prefix
-        self.tool_separator = tool_separator
-        self.resource_separator = resource_separator
-        self.prompt_separator = prompt_separator
-
-    async def get_tools(self) -> dict[str, Tool]:
-        tools = await self.server.get_tools()
-        return {
-            f"{self.prefix}{self.tool_separator}{key}": tool
-            for key, tool in tools.items()
-        }
-
-    async def get_resources(self) -> dict[str, Resource]:
-        resources = await self.server.get_resources()
-        return {
-            f"{self.prefix}{self.resource_separator}{key}": resource
-            for key, resource in resources.items()
-        }
-
-    async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
-        templates = await self.server.get_resource_templates()
-        return {
-            f"{self.prefix}{self.resource_separator}{key}": template
-            for key, template in templates.items()
-        }
-
-    async def get_prompts(self) -> dict[str, Prompt]:
-        prompts = await self.server.get_prompts()
-        return {
-            f"{self.prefix}{self.prompt_separator}{key}": prompt
-            for key, prompt in prompts.items()
-        }
-
-    def match_tool(self, key: str) -> bool:
-        return key.startswith(f"{self.prefix}{self.tool_separator}")
-
-    def strip_tool_prefix(self, key: str) -> str:
-        return key.removeprefix(f"{self.prefix}{self.tool_separator}")
-
-    def match_resource(self, key: str) -> bool:
-        return key.startswith(f"{self.prefix}{self.resource_separator}")
-
-    def strip_resource_prefix(self, key: str) -> str:
-        return key.removeprefix(f"{self.prefix}{self.resource_separator}")
-
-    def match_prompt(self, key: str) -> bool:
-        return key.startswith(f"{self.prefix}{self.prompt_separator}")
-
-    def strip_prompt_prefix(self, key: str) -> str:
-        return key.removeprefix(f"{self.prefix}{self.prompt_separator}")
-
-
-class TimedCache:
-    def __init__(self, expiration: datetime.timedelta):
-        self.expiration = expiration
-        self.cache: dict[Any, tuple[Any, datetime.datetime]] = {}
-
-    def set(self, key: Any, value: Any) -> None:
-        expires = datetime.datetime.now() + self.expiration
-        self.cache[key] = (value, expires)
-
-    def get(self, key: Any) -> Any:
-        value = self.cache.get(key)
-        if value is not None and value[1] > datetime.datetime.now():
-            return value[0]
-        else:
-            return NOT_FOUND
-
-    def clear(self) -> None:
-        self.cache.clear()
-
 
 @asynccontextmanager
 async def default_lifespan(server: FastMCP) -> AsyncIterator[Any]:
@@ -325,7 +233,7 @@ class FastMCP(Generic[LifespanResultT]):
 
     async def get_tools(self) -> dict[str, Tool]:
         """Get all registered tools, indexed by registered key."""
-        if (tools := self._cache.get("tools")) is NOT_FOUND:
+        if (tools := self._cache.get("tools")) is self._cache.NOT_FOUND:
             tools = {}
             for server in self._mounted_servers.values():
                 server_tools = await server.get_tools()
@@ -336,7 +244,7 @@ class FastMCP(Generic[LifespanResultT]):
 
     async def get_resources(self) -> dict[str, Resource]:
         """Get all registered resources, indexed by registered key."""
-        if (resources := self._cache.get("resources")) is NOT_FOUND:
+        if (resources := self._cache.get("resources")) is self._cache.NOT_FOUND:
             resources = {}
             for server in self._mounted_servers.values():
                 server_resources = await server.get_resources()
@@ -347,7 +255,9 @@ class FastMCP(Generic[LifespanResultT]):
 
     async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
         """Get all registered resource templates, indexed by registered key."""
-        if (templates := self._cache.get("resource_templates")) is NOT_FOUND:
+        if (
+            templates := self._cache.get("resource_templates")
+        ) is self._cache.NOT_FOUND:
             templates = {}
             for server in self._mounted_servers.values():
                 server_templates = await server.get_resource_templates()
@@ -360,7 +270,7 @@ class FastMCP(Generic[LifespanResultT]):
         """
         List all available prompts.
         """
-        if (prompts := self._cache.get("prompts")) is NOT_FOUND:
+        if (prompts := self._cache.get("prompts")) is self._cache.NOT_FOUND:
             prompts = {}
             for server in self._mounted_servers.values():
                 server_prompts = await server.get_prompts()
@@ -1145,3 +1055,74 @@ def _validate_resource_prefix(prefix: str) -> None:
         raise ValueError(
             f"Resource prefix or separator would result in an invalid resource URI: {e}"
         )
+
+
+class MountedServer:
+    def __init__(
+        self,
+        prefix: str,
+        server: FastMCP,
+        tool_separator: str | None = None,
+        resource_separator: str | None = None,
+        prompt_separator: str | None = None,
+    ):
+        if tool_separator is None:
+            tool_separator = "_"
+        if resource_separator is None:
+            resource_separator = "+"
+        if prompt_separator is None:
+            prompt_separator = "_"
+
+        _validate_resource_prefix(f"{prefix}{resource_separator}")
+
+        self.server = server
+        self.prefix = prefix
+        self.tool_separator = tool_separator
+        self.resource_separator = resource_separator
+        self.prompt_separator = prompt_separator
+
+    async def get_tools(self) -> dict[str, Tool]:
+        tools = await self.server.get_tools()
+        return {
+            f"{self.prefix}{self.tool_separator}{key}": tool
+            for key, tool in tools.items()
+        }
+
+    async def get_resources(self) -> dict[str, Resource]:
+        resources = await self.server.get_resources()
+        return {
+            f"{self.prefix}{self.resource_separator}{key}": resource
+            for key, resource in resources.items()
+        }
+
+    async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
+        templates = await self.server.get_resource_templates()
+        return {
+            f"{self.prefix}{self.resource_separator}{key}": template
+            for key, template in templates.items()
+        }
+
+    async def get_prompts(self) -> dict[str, Prompt]:
+        prompts = await self.server.get_prompts()
+        return {
+            f"{self.prefix}{self.prompt_separator}{key}": prompt
+            for key, prompt in prompts.items()
+        }
+
+    def match_tool(self, key: str) -> bool:
+        return key.startswith(f"{self.prefix}{self.tool_separator}")
+
+    def strip_tool_prefix(self, key: str) -> str:
+        return key.removeprefix(f"{self.prefix}{self.tool_separator}")
+
+    def match_resource(self, key: str) -> bool:
+        return key.startswith(f"{self.prefix}{self.resource_separator}")
+
+    def strip_resource_prefix(self, key: str) -> str:
+        return key.removeprefix(f"{self.prefix}{self.resource_separator}")
+
+    def match_prompt(self, key: str) -> bool:
+        return key.startswith(f"{self.prefix}{self.prompt_separator}")
+
+    def strip_prompt_prefix(self, key: str) -> str:
+        return key.removeprefix(f"{self.prefix}{self.prompt_separator}")

src/fastmcp/utilities/cache.py

@@ -0,0 +1,24 @@
+import datetime
+from typing import Any
+
+
+class TimedCache:
+    NOT_FOUND = object()
+
+    def __init__(self, expiration: datetime.timedelta):
+        self.expiration = expiration
+        self.cache: dict[Any, tuple[Any, datetime.datetime]] = {}
+
+    def set(self, key: Any, value: Any) -> None:
+        expires = datetime.datetime.now() + self.expiration
+        self.cache[key] = (value, expires)
+
+    def get(self, key: Any) -> Any:
+        value = self.cache.get(key)
+        if value is not None and value[1] > datetime.datetime.now():
+            return value[0]
+        else:
+            return self.NOT_FOUND
+
+    def clear(self) -> None:
+        self.cache.clear()