MCP Hub

Instructions

Implementation Reference

• app.py:457-473 (handler)

  Handler function that executes the get_cache_status tool logic, checks for advanced features, and delegates to cache_manager.get_cache_status()
  def get_cache_status() -> Dict[str, Any]: """Get cache status and statistics.""" if not ADVANCED_FEATURES_AVAILABLE: return { "status": "basic_mode", "message": "Cache monitoring not available. Install 'pip install psutil aiohttp' to enable cache statistics.", "cache_info": { "caching_available": False, "recommendation": "Install advanced features for intelligent caching" } } try: from mcp_hub.cache_utils import cache_manager return cache_manager.get_cache_status() except Exception as e: return {"error": f"Cache status failed: {str(e)}"}
• app.py:1078-1083 (registration)

  Gradio MCP tool registration for get_cache_status_service which calls the handler fn=get_cache_status
  cache_btn.click( fn=get_cache_status, inputs=[], outputs=cache_output, api_name="get_cache_status_service" )
• mcp_hub/cache_utils.py:144-184 (helper)

  CacheManager.get_cache_status method providing file-based cache statistics used by the main handler
  def get_cache_status(self) -> Dict[str, Any]: """Get detailed status information about the cache system.""" try: # Count cache files cache_files = list(self.cache_dir.glob("*.cache")) cache_count = len(cache_files) # Calculate cache directory size total_size = sum(f.stat().st_size for f in cache_files) # Count expired files expired_count = 0 current_time = datetime.now() for cache_file in cache_files: try: with open(cache_file, 'rb') as f: cache_data = pickle.load(f) if current_time > cache_data['expires_at']: expired_count += 1 except Exception: expired_count += 1 # Count corrupted files as expired # Get cache stats return { "status": "healthy", "cache_dir": str(self.cache_dir), "total_files": cache_count, "expired_files": expired_count, "total_size_bytes": total_size, "total_size_mb": round(total_size / (1024 * 1024), 2), "default_ttl_seconds": self.default_ttl, "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"Failed to get cache status: {str(e)}") return { "status": "error", "error": str(e), "timestamp": datetime.now().isoformat() }
• mcp_hub/redis_cache.py:275-320 (helper)

  RedisCacheBackend.get_cache_status method for Redis-based cache statistics, used if Redis backend is configured
  def get_cache_status(self) -> Dict[str, Any]: """Get detailed status information about the cache. Returns: Dictionary with cache status information """ try: # Get Redis info info = self.client.info() # Count keys with our prefix pattern = f"{self.key_prefix}*" cursor = 0 key_count = 0 while True: cursor, keys = self.client.scan(cursor, match=pattern, count=100) key_count += len(keys) if cursor == 0: break # Get memory usage memory_used = info.get("used_memory", 0) memory_used_human = info.get("used_memory_human", "0B") return { "status": "healthy", "backend": "redis", "redis_version": info.get("redis_version", "unknown"), "connected_clients": info.get("connected_clients", 0), "total_keys": key_count, "memory_used_bytes": memory_used, "memory_used_human": memory_used_human, "default_ttl_seconds": self.default_ttl, "key_prefix": self.key_prefix, "timestamp": datetime.now().isoformat(), } except Exception as e: logger.error(f"Failed to get cache status: {e}") return { "status": "error", "backend": "redis", "error": str(e), "timestamp": datetime.now().isoformat(), }