MCP Server Hero

"""Base caching framework.""" import asyncio import hashlib import time from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional, Set from dataclasses import dataclass from ..utils.logging import get_logger logger = get_logger(__name__) @dataclass class CacheEntry: """Cache entry with metadata.""" key: str value: Any created_at: float accessed_at: float hit_count: int = 0 ttl: Optional[float] = None def is_expired(self) -> bool: """Check if entry is expired. Returns: True if expired """ if self.ttl is None: return False return time.time() > self.created_at + self.ttl def update_access(self) -> None: """Update access statistics.""" self.accessed_at = time.time() self.hit_count += 1 class BaseCache(ABC): """Base class for cache implementations.""" def __init__(self, name: str, max_size: int = 1000) -> None: """Initialize cache. Args: name: Cache name max_size: Maximum number of entries """ self.name = name self.max_size = max_size self.enabled = True self._stats = { "hits": 0, "misses": 0, "sets": 0, "deletes": 0, "evictions": 0, } self._logger = get_logger(f"{__name__}.{name}") @abstractmethod async def get(self, key: str) -> Optional[Any]: """Get value from cache. Args: key: Cache key Returns: Cached value or None """ pass @abstractmethod async def set(self, key: str, value: Any, ttl: Optional[float] = None) -> None: """Set value in cache. Args: key: Cache key value: Value to cache ttl: Time to live in seconds """ pass @abstractmethod async def delete(self, key: str) -> bool: """Delete value from cache. Args: key: Cache key Returns: True if key existed """ pass @abstractmethod async def clear(self) -> None: """Clear all cache entries.""" pass @abstractmethod async def size(self) -> int: """Get current cache size. Returns: Number of entries in cache """ pass @abstractmethod async def keys(self) -> List[str]: """Get all cache keys. Returns: List of cache keys """ pass def get_stats(self) -> Dict[str, Any]: """Get cache statistics. Returns: Statistics dictionary """ total_requests = self._stats["hits"] + self._stats["misses"] hit_rate = (self._stats["hits"] / total_requests * 100) if total_requests > 0 else 0 return { **self._stats, "hit_rate": hit_rate, "total_requests": total_requests, "enabled": self.enabled, "max_size": self.max_size, } def reset_stats(self) -> None: """Reset cache statistics.""" for key in self._stats: self._stats[key] = 0 self._logger.info(f"Cache {self.name} statistics reset") def _record_hit(self) -> None: """Record cache hit.""" self._stats["hits"] += 1 def _record_miss(self) -> None: """Record cache miss.""" self._stats["misses"] += 1 def _record_set(self) -> None: """Record cache set.""" self._stats["sets"] += 1 def _record_delete(self) -> None: """Record cache delete.""" self._stats["deletes"] += 1 def _record_eviction(self) -> None: """Record cache eviction.""" self._stats["evictions"] += 1 class CacheManager: """Manager for multiple cache instances.""" def __init__(self) -> None: """Initialize cache manager.""" self.caches: Dict[str, BaseCache] = {} self._logger = get_logger(__name__) def add_cache(self, cache: BaseCache) -> None: """Add cache to manager. Args: cache: Cache instance to add """ self.caches[cache.name] = cache self._logger.info(f"Added cache: {cache.name}") def remove_cache(self, cache_name: str) -> None: """Remove cache from manager. Args: cache_name: Name of cache to remove """ if cache_name in self.caches: del self.caches[cache_name] self._logger.info(f"Removed cache: {cache_name}") def get_cache(self, cache_name: str) -> Optional[BaseCache]: """Get cache by name. Args: cache_name: Cache name Returns: Cache instance or None """ return self.caches.get(cache_name) async def clear_all(self) -> None: """Clear all caches.""" for cache in self.caches.values(): await cache.clear() self._logger.info("Cleared all caches") def get_all_stats(self) -> Dict[str, Dict[str, Any]]: """Get statistics for all caches. Returns: Dictionary mapping cache names to their statistics """ return {name: cache.get_stats() for name, cache in self.caches.items()} def reset_all_stats(self) -> None: """Reset statistics for all caches.""" for cache in self.caches.values(): cache.reset_stats() self._logger.info("Reset all cache statistics") def enable_all(self) -> None: """Enable all caches.""" for cache in self.caches.values(): cache.enabled = True self._logger.info("Enabled all caches") def disable_all(self) -> None: """Disable all caches.""" for cache in self.caches.values(): cache.enabled = False self._logger.info("Disabled all caches") def generate_cache_key(*args: Any, prefix: str = "") -> str: """Generate cache key from arguments. Args: *args: Arguments to include in key prefix: Optional prefix for key Returns: Generated cache key """ # Create string representation of all arguments key_parts = [str(arg) for arg in args] key_string = "|".join(key_parts) # Hash the key for consistent length and safety key_hash = hashlib.sha256(key_string.encode()).hexdigest()[:16] if prefix: return f"{prefix}:{key_hash}" return key_hash def cache_result(cache_name: str, ttl: Optional[float] = None, key_prefix: str = ""): """Decorator to cache function results. Args: cache_name: Name of cache to use ttl: Time to live for cached result key_prefix: Prefix for cache key Returns: Decorator function """ def decorator(func): async def wrapper(*args, **kwargs): # Get cache manager from first argument (should be server instance) cache_manager = None if args and hasattr(args[0], "cache_manager"): cache_manager = args[0].cache_manager if not cache_manager: # No cache available, execute function normally return await func(*args, **kwargs) cache = cache_manager.get_cache(cache_name) if not cache or not cache.enabled: # Cache not available or disabled return await func(*args, **kwargs) # Generate cache key cache_key = generate_cache_key(*args, **kwargs, prefix=key_prefix) # Try to get from cache cached_result = await cache.get(cache_key) if cached_result is not None: return cached_result # Execute function and cache result result = await func(*args, **kwargs) await cache.set(cache_key, result, ttl) return result return wrapper return decorator