Remove unused TimedCache (#1303)

docs/python-sdk/fastmcp-utilities-cache.mdx

@@ -1,30 +0,0 @@
----
-title: cache
-sidebarTitle: cache
----
-
-# `fastmcp.utilities.cache`
-
-## Classes
-
-### `TimedCache` <sup><a href="https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/utilities/cache.py#L7" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
-
-**Methods:**
-
-#### `set` <sup><a href="https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/utilities/cache.py#L14" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
-
-```python
-set(self, key: Any, value: Any) -> None
-```
-
-#### `get` <sup><a href="https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/utilities/cache.py#L18" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
-
-```python
-get(self, key: Any) -> Any
-```
-
-#### `clear` <sup><a href="https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/utilities/cache.py#L25" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
-
-```python
-clear(self) -> None
-```

src/fastmcp/server/server.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import datetime
 import inspect
 import re
 import warnings
@@ -63,7 +62,6 @@ from fastmcp.settings import Settings
 from fastmcp.tools import ToolManager
 from fastmcp.tools.tool import FunctionTool, Tool, ToolResult
 from fastmcp.tools.tool_transform import ToolTransformConfig
-from fastmcp.utilities.cache import TimedCache
 from fastmcp.utilities.cli import log_server_banner
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.logging import get_logger
@@ -174,9 +172,6 @@ class FastMCP(Generic[LifespanResultT]):
             resource_prefix_format or fastmcp.settings.resource_prefix_format
         )
 
-        self._cache = TimedCache(
-            expiration=datetime.timedelta(seconds=cache_expiration_seconds or 0)
-        )
         self._additional_http_routes: list[BaseRoute] = []
         self._tool_manager = ToolManager(
             duplicate_behavior=on_duplicate_tools,
@@ -816,7 +811,6 @@ class FastMCP(Generic[LifespanResultT]):
             The tool instance that was added to the server.
         """
         self._tool_manager.add_tool(tool)
-        self._cache.clear()
 
         # Send notification if we're in a request context
         try:
@@ -839,7 +833,6 @@ class FastMCP(Generic[LifespanResultT]):
             NotFoundError: If the tool is not found
         """
         self._tool_manager.remove_tool(name)
-        self._cache.clear()
 
         # Send notification if we're in a request context
         try:
@@ -1033,7 +1026,6 @@ class FastMCP(Generic[LifespanResultT]):
             The resource instance that was added to the server.
         """
         self._resource_manager.add_resource(resource)
-        self._cache.clear()
 
         # Send notification if we're in a request context
         try:
@@ -1105,7 +1097,6 @@ class FastMCP(Generic[LifespanResultT]):
             mime_type=mime_type,
             tags=tags,
         )
-        self._cache.clear()
 
     def resource(
         self,
@@ -1254,7 +1245,6 @@ class FastMCP(Generic[LifespanResultT]):
             The prompt instance that was added to the server.
         """
         self._prompt_manager.add_prompt(prompt)
-        self._cache.clear()
 
         # Send notification if we're in a request context
         try:
@@ -1803,8 +1793,6 @@ class FastMCP(Generic[LifespanResultT]):
         self._resource_manager.mount(mounted_server)
         self._prompt_manager.mount(mounted_server)
 
-        self._cache.clear()
-
     async def import_server(
         self,
         server: FastMCP[LifespanResultT],
@@ -1927,8 +1915,6 @@ class FastMCP(Generic[LifespanResultT]):
         else:
             logger.debug(f"Imported server {server.name}")
 
-        self._cache.clear()
-
     @classmethod
     def from_openapi(
         cls,

src/fastmcp/utilities/cache.py

@@ -1,26 +0,0 @@
-import datetime
-from typing import Any
-
-UTC = datetime.timezone.utc
-
-
-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(UTC) + 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(UTC):
-            return value[0]
-        else:
-            return self.NOT_FOUND
-
-    def clear(self) -> None:
-        self.cache.clear()

tests/server/test_mount.py

@@ -629,7 +629,6 @@ class TestDynamicChanges:
 
         # The tool should no longer be accessible
         # Refresh the cache by clearing it
-        main_app._cache.cache.clear()
         tools = await main_app.get_tools()
         assert "sub_temp_tool" not in tools
 

tests/utilities/test_cache.py

@@ -1,233 +0,0 @@
-"""Tests for the cache.py module."""
-
-import datetime
-import time
-from unittest.mock import patch
-
-from fastmcp.utilities.cache import TimedCache
-
-
-class TestTimedCache:
-    """Tests for the TimedCache class."""
-
-    def test_init(self):
-        """Test that a TimedCache can be initialized with an expiration."""
-        expiration = datetime.timedelta(seconds=10)
-        cache = TimedCache(expiration)
-        assert cache.expiration == expiration
-        assert isinstance(cache.cache, dict)
-        assert len(cache.cache) == 0
-
-    def test_set(self):
-        """Test that values can be set in the cache."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-        key, value = "test_key", "test_value"
-
-        with patch("datetime.datetime") as mock_datetime:
-            now = datetime.datetime(2023, 1, 1, tzinfo=datetime.timezone.utc)
-            mock_datetime.now.return_value = now
-
-            cache.set(key, value)
-
-            # Check that the value is stored with the correct expiration
-            assert key in cache.cache
-            stored_value, expiration = cache.cache[key]
-            assert stored_value == value
-            assert expiration == now + datetime.timedelta(seconds=10)
-
-    def test_get_found(self):
-        """Test retrieving a value that exists and has not expired."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-        key, value = "test_key", "test_value"
-
-        # Set a future expiration time
-        future = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
-            seconds=30
-        )
-        cache.cache[key] = (value, future)
-
-        # The value should be returned
-        assert cache.get(key) == value
-
-    def test_get_expired(self):
-        """Test retrieving a value that exists but has expired."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-        key, value = "test_key", "test_value"
-
-        # Set a past expiration time
-        past = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(
-            seconds=1
-        )
-        cache.cache[key] = (value, past)
-
-        # Should return NOT_FOUND
-        assert cache.get(key) is TimedCache.NOT_FOUND
-
-    def test_get_not_found(self):
-        """Test retrieving a value that doesn't exist in the cache."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-
-        # Key doesn't exist
-        assert cache.get("nonexistent_key") is TimedCache.NOT_FOUND
-
-    def test_clear(self):
-        """Test that the cache can be cleared."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-
-        # Add some items
-        cache.set("key1", "value1")
-        cache.set("key2", "value2")
-        assert len(cache.cache) == 2
-
-        # Clear the cache
-        cache.clear()
-        assert len(cache.cache) == 0
-
-    def test_real_expiration(self):
-        """Test that values actually expire after the specified time."""
-        # Use a very short expiration for the test
-        cache = TimedCache(datetime.timedelta(milliseconds=50))
-        key, value = "test_key", "test_value"
-
-        cache.set(key, value)
-        # Value should be available immediately
-        assert cache.get(key) == value
-
-        # Wait for expiration
-        time.sleep(0.06)  # 60 milliseconds, slightly longer than expiration
-
-        # Value should now be expired
-        assert cache.get(key) is TimedCache.NOT_FOUND
-
-    def test_overwrite_value(self):
-        """Test that setting a key that already exists overwrites the old value."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-        key = "test_key"
-
-        # Set initial value
-        cache.set(key, "initial_value")
-        assert cache.get(key) == "initial_value"
-
-        # Overwrite with new value
-        cache.set(key, "new_value")
-        assert cache.get(key) == "new_value"
-
-    def test_extends_expiration_on_overwrite(self):
-        """Test that overwriting a key extends its expiration time."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-        key = "test_key"
-
-        with patch("datetime.datetime") as mock_datetime:
-            # Set initial value at t=0
-            initial_time = datetime.datetime(2023, 1, 1, tzinfo=datetime.timezone.utc)
-            mock_datetime.now.return_value = initial_time
-            cache.set(key, "initial_value")
-
-            initial_expiration = cache.cache[key][1]
-            assert initial_expiration == initial_time + datetime.timedelta(seconds=10)
-
-            # Overwrite at t=5
-            later_time = initial_time + datetime.timedelta(seconds=5)
-            mock_datetime.now.return_value = later_time
-            cache.set(key, "new_value")
-
-            # Expiration should be extended
-            new_expiration = cache.cache[key][1]
-            assert new_expiration == later_time + datetime.timedelta(seconds=10)
-
-    def test_different_key_types(self):
-        """Test that different types of keys can be used."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-
-        # Test various key types
-        keys_and_values = [
-            (42, "int_value"),
-            (3.14, "float_value"),
-            ((1, 2), "tuple_value"),
-            (frozenset({1, 2, 3}), "frozenset_value"),
-        ]
-
-        for key, value in keys_and_values:
-            cache.set(key, value)
-            assert cache.get(key) == value
-
-    def test_none_value(self):
-        """Test that None can be stored as a value."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-        key = "none_key"
-
-        cache.set(key, None)
-        # The stored value is None, but get() should return None, not NOT_FOUND
-        assert cache.get(key) is None
-
-    def test_edge_case_zero_expiration(self):
-        """Test with a zero expiration time."""
-        cache = TimedCache(datetime.timedelta(seconds=0))
-        key, value = "test_key", "test_value"
-
-        cache.set(key, value)
-        # The value might already be expired by the time we call get()
-        # We can't make strong assertions here due to timing variability
-        retrieved = cache.get(key)
-        assert retrieved in (value, TimedCache.NOT_FOUND)
-
-    def test_negative_expiration(self):
-        """Test with a negative expiration time."""
-        cache = TimedCache(datetime.timedelta(seconds=-1))
-        key, value = "test_key", "test_value"
-
-        cache.set(key, value)
-        # Value should be immediately expired
-        assert cache.get(key) is TimedCache.NOT_FOUND
-
-    def test_cache_consistency(self):
-        """Test cache consistency with multiple operations."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-
-        # Add multiple items
-        cache.set("key1", "value1")
-        cache.set("key2", "value2")
-        cache.set("key3", "value3")
-
-        # Check all items
-        assert cache.get("key1") == "value1"
-        assert cache.get("key2") == "value2"
-        assert cache.get("key3") == "value3"
-
-        # Overwrite one item
-        cache.set("key2", "updated_value")
-
-        # Check again
-        assert cache.get("key1") == "value1"
-        assert cache.get("key2") == "updated_value"
-        assert cache.get("key3") == "value3"
-
-        # Clear and verify all items are gone
-        cache.clear()
-        assert cache.get("key1") is TimedCache.NOT_FOUND
-        assert cache.get("key2") is TimedCache.NOT_FOUND
-        assert cache.get("key3") is TimedCache.NOT_FOUND
-
-    def test_large_expiration(self):
-        """Test with a very large expiration time."""
-        # One year expiration
-        cache = TimedCache(datetime.timedelta(days=365))
-        key, value = "test_key", "test_value"
-
-        cache.set(key, value)
-        assert cache.get(key) == value
-
-    def test_many_items(self):
-        """Test cache with many items."""
-        cache = TimedCache(datetime.timedelta(seconds=10))
-
-        # Add 1000 items
-        for i in range(1000):
-            cache.set(f"key{i}", f"value{i}")
-
-        # Check size
-        assert len(cache.cache) == 1000
-
-        # Check some random items
-        for i in [0, 123, 456, 789, 999]:
-            assert cache.get(f"key{i}") == f"value{i}"