Add cache TTL

.env.example

@@ -7,3 +7,6 @@ IPINFO_TOKEN=
 
 # Base URL for the legacy ipinfo.io endpoints, e.g. /me (default: https://ipinfo.io)
 # IPINFO_LEGACY_BASE_URL=https://ipinfo.io
+
+# How long cached IP results stay fresh, in seconds (default: 3600)
+# IPINFO_CACHE_TTL=3600

README.md

@@ -38,6 +38,7 @@ IPINFO_TRANSPORT=http IPINFO_HOST=0.0.0.0 IPINFO_PORT=8000 uv run ipinfo-mcp-ser
 | `IPINFO_TOKEN`        |                         | API token                             |
 | `IPINFO_API_BASE_URL` | `https://api.ipinfo.io` | Base URL for `api.ipinfo.io` endpoints |
 | `IPINFO_LEGACY_BASE_URL` | `https://ipinfo.io`  | Base URL for legacy `ipinfo.io` endpoints (e.g. `/me`) |
+| `IPINFO_CACHE_TTL`    | `3600`                  | Seconds a cached IP result stays fresh |
 | `IPINFO_TRANSPORT`    | `stdio`                 | Transport type (`stdio` or `http`)    |
 | `IPINFO_HOST`         | `0.0.0.0`               | HTTP host (only for `http` transport) |
 | `IPINFO_PORT`         | `8000`                  | HTTP port (only for `http` transport) |

src/ipinfo_mcp/cache.py

@@ -1,10 +1,14 @@
 import hashlib
+import time
 
 from ipinfo_mcp.types import LiteResponse, LookupResponse, ResproxyResponse
 
 # Any single-IP response that can be cached
 CachedResponse = LiteResponse | LookupResponse | ResproxyResponse | dict[str, object]
 
+# How long an entry stays fresh, in seconds.
+DEFAULT_TTL = 3600.0
+
 
 class IPCache:
     """
@@ -15,10 +19,16 @@ class IPCache:
     Entries are scoped also using the token that fetched them. The cache is process
     wide, so if we don't use the token as part of the key we would hand one token's
     results to another.
+
+    Entries expire after ttl seconds. Expiry is checked on read, and an expired
+    entry is dropped as we find it, so nothing sweeps the store in the background.
     """
 
-    def __init__(self) -> None:
-        self._store: dict[tuple[str, str, str], CachedResponse] = {}
+    def __init__(self, ttl: float = DEFAULT_TTL) -> None:
+        self._ttl = ttl
+        # Each entry is (stored_at, data), stored_at coming from a monotonic clock
+        # so that a system clock change can't make an entry look fresh forever.
+        self._store: dict[tuple[str, str, str], tuple[float, CachedResponse]] = {}
 
     @staticmethod
     def _hash_token(token: str) -> str:
@@ -29,21 +39,34 @@ def _hash_token(token: str) -> str:
         """
         return hashlib.sha256(token.encode()).hexdigest()[:32]
 
+    def _get_fresh(self, key: tuple[str, str, str]) -> CachedResponse | None:
+        """Read an entry by key, dropping and reporting it as a miss if it expired."""
+        entry = self._store.get(key)
+        if entry is None:
+            return None
+
+        stored_at, data = entry
+        if time.monotonic() > stored_at + self._ttl:
+            del self._store[key]
+            return None
+
+        return data
+
     def get(self, token: str, namespace: str, ip: str) -> CachedResponse | None:
         """Get this token's cached data for an IP in a namespace. Returns None on miss."""
-        return self._store.get((self._hash_token(token), namespace, ip))
+        return self._get_fresh((self._hash_token(token), namespace, ip))
 
     def put(self, token: str, namespace: str, ip: str, data: CachedResponse) -> None:
         """Store data for an IP in a namespace, scoped to this token."""
-        self._store[(self._hash_token(token), namespace, ip)] = data
+        self._store[(self._hash_token(token), namespace, ip)] = (time.monotonic(), data)
 
     def get_many(self, token: str, namespace: str, ips: list[str]) -> tuple[dict[str, CachedResponse], list[str]]:
         """Look up multiple IPs for this token. Returns (cached_results, cache_misses)."""
         token_hash = self._hash_token(token)
         cached: dict[str, CachedResponse] = {}
         misses: list[str] = []
         for ip in ips:
-            data = self._store.get((token_hash, namespace, ip))
+            data = self._get_fresh((token_hash, namespace, ip))
             if data is not None:
                 cached[ip] = data
             else:
@@ -53,5 +76,6 @@ def get_many(self, token: str, namespace: str, ips: list[str]) -> tuple[dict[str
     def put_many(self, token: str, namespace: str, items: dict[str, CachedResponse]) -> None:
         """Store multiple IP results in a namespace, scoped to this token."""
         token_hash = self._hash_token(token)
+        stored_at = time.monotonic()
         for ip, data in items.items():
-            self._store[(token_hash, namespace, ip)] = data
+            self._store[(token_hash, namespace, ip)] = (stored_at, data)

src/ipinfo_mcp/server.py

@@ -11,7 +11,7 @@
 from starlette.responses import HTMLResponse, JSONResponse
 from starlette.types import ASGIApp, Receive, Scope, Send
 
-from ipinfo_mcp.cache import IPCache
+from ipinfo_mcp.cache import DEFAULT_TTL, IPCache
 from ipinfo_mcp.client import IPinfoClient
 from ipinfo_mcp.logging import setup_logging
 from ipinfo_mcp.tools.asn import register_asn
@@ -28,13 +28,15 @@ class Settings(TypedDict):
     api_token: str | None
     api_base_url: str
     legacy_base_url: str
+    cache_ttl: float
 
 
 def _settings() -> Settings:
     return {
         "api_token": os.environ.get("IPINFO_TOKEN"),
         "api_base_url": os.environ.get("IPINFO_API_BASE_URL", "https://api.ipinfo.io"),
         "legacy_base_url": os.environ.get("IPINFO_LEGACY_BASE_URL", "https://ipinfo.io"),
+        "cache_ttl": float(os.environ.get("IPINFO_CACHE_TTL", DEFAULT_TTL)),
     }
 
 
@@ -52,10 +54,11 @@ async def lifespan(_: FastMCP) -> AsyncIterator[ContextData]:
         base_url=settings["api_base_url"],
         legacy_base_url=settings["legacy_base_url"],
     ) as client:
-        cache = IPCache()
+        cache = IPCache(ttl=settings["cache_ttl"])
         logger.info(
-            "IPinfo MCP server started (token=%s)",
+            "IPinfo MCP server started (token=%s cache_ttl=%s)",
             "configured" if settings["api_token"] else "anonymous",
+            settings["cache_ttl"],
         )
         yield {"client": client, "cache": cache, "api_token": settings["api_token"]}
     logger.info("IPinfo MCP server stopped")

tests/test_cache.py

@@ -1,9 +1,32 @@
+import pytest
+
+from ipinfo_mcp import cache as cache_module
 from ipinfo_mcp.cache import IPCache
 
 TOKEN = "token_a"
 OTHER_TOKEN = "token_b"
 
 
+class FakeClock:
+    """Controllable stand-in for time.monotonic, so TTL tests need no sleeps."""
+
+    def __init__(self, now: float = 1000.0) -> None:
+        self.now = now
+
+    def __call__(self) -> float:
+        return self.now
+
+    def advance(self, seconds: float) -> None:
+        self.now += seconds
+
+
+@pytest.fixture
+def clock(monkeypatch: pytest.MonkeyPatch) -> FakeClock:
+    fake = FakeClock()
+    monkeypatch.setattr(cache_module.time, "monotonic", fake)
+    return fake
+
+
 class TestIPCachePutGet:
     def test_get_returns_none_for_missing_key(self) -> None:
         cache = IPCache()
@@ -124,3 +147,63 @@ def test_put_many_is_scoped_to_token(self) -> None:
         cache.put_many(TOKEN, "lite", {"8.8.8.8": {"ip": "8.8.8.8"}})
 
         assert cache.get(OTHER_TOKEN, "lite", "8.8.8.8") is None
+
+
+class TestIPCacheTTL:
+    def test_entry_is_fresh_before_ttl(self, clock: FakeClock) -> None:
+        cache = IPCache(ttl=60.0)
+        cache.put(TOKEN, "lite", "8.8.8.8", {"ip": "8.8.8.8"})
+
+        clock.advance(59.0)
+        assert cache.get(TOKEN, "lite", "8.8.8.8") == {"ip": "8.8.8.8"}
+
+    def test_entry_is_fresh_exactly_at_ttl(self, clock: FakeClock) -> None:
+        cache = IPCache(ttl=60.0)
+        cache.put(TOKEN, "lite", "8.8.8.8", {"ip": "8.8.8.8"})
+
+        clock.advance(60.0)
+        assert cache.get(TOKEN, "lite", "8.8.8.8") == {"ip": "8.8.8.8"}
+
+    def test_entry_expires_after_ttl(self, clock: FakeClock) -> None:
+        cache = IPCache(ttl=60.0)
+        cache.put(TOKEN, "lite", "8.8.8.8", {"ip": "8.8.8.8"})
+
+        clock.advance(60.1)
+        assert cache.get(TOKEN, "lite", "8.8.8.8") is None
+
+    def test_expired_entry_is_dropped_from_the_store(self, clock: FakeClock) -> None:
+        """Expiry cleans as it reads, so an expired entry doesn't linger."""
+        cache = IPCache(ttl=60.0)
+        cache.put(TOKEN, "lite", "8.8.8.8", {"ip": "8.8.8.8"})
+
+        clock.advance(61.0)
+        _ = cache.get(TOKEN, "lite", "8.8.8.8")
+
+        assert cache._store == {}
+
+    def test_expired_entry_is_a_miss_in_get_many(self, clock: FakeClock) -> None:
+        cache = IPCache(ttl=60.0)
+        cache.put(TOKEN, "lite", "8.8.8.8", {"ip": "8.8.8.8"})
+        clock.advance(61.0)
+        cache.put(TOKEN, "lite", "1.1.1.1", {"ip": "1.1.1.1"})
+
+        cached, misses = cache.get_many(TOKEN, "lite", ["8.8.8.8", "1.1.1.1"])
+        assert cached == {"1.1.1.1": {"ip": "1.1.1.1"}}
+        assert misses == ["8.8.8.8"]
+
+    def test_put_resets_the_ttl(self, clock: FakeClock) -> None:
+        cache = IPCache(ttl=60.0)
+        cache.put(TOKEN, "lite", "8.8.8.8", {"old": True})
+
+        clock.advance(59.0)
+        cache.put(TOKEN, "lite", "8.8.8.8", {"new": True})
+
+        clock.advance(59.0)
+        assert cache.get(TOKEN, "lite", "8.8.8.8") == {"new": True}
+
+    def test_put_many_entries_expire(self, clock: FakeClock) -> None:
+        cache = IPCache(ttl=60.0)
+        cache.put_many(TOKEN, "lite", {"8.8.8.8": {"ip": "8.8.8.8"}})
+
+        clock.advance(61.0)
+        assert cache.get(TOKEN, "lite", "8.8.8.8") is None