invalidate_cache
Clear cached data for specific resources to ensure access to the most recent Taiwan stock market information. Specify resource patterns to refresh targeted data.
Instructions
Invalidate resource cache for fresh data
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| resource_pattern | No |
Implementation Reference
- mcp_server.py:307-337 (handler)MCP tool handler for 'invalidate_cache' that invalidates the resource cache by calling resource_manager.invalidate_cache and returns statistics.@mcp.tool(name="invalidate_cache", description="Invalidate resource cache for fresh data") async def invalidate_cache_tool(resource_pattern: Optional[str] = None) -> Dict[str, Any]: """Invalidate resource cache.""" try: invalidated_count = resource_manager.invalidate_cache(resource_pattern) cache_stats = resource_manager.get_cache_stats() return { "pattern": resource_pattern or "all", "invalidated_entries": invalidated_count, "cache_statistics": cache_stats, "_metadata": { "source": "tw-stock-agent", "timestamp": datetime.now().isoformat(), "data_type": "cache_management" } } except Exception as e: logger.error(f"Failed to invalidate cache: {e}") return { "pattern": resource_pattern or "all", "invalidated_entries": 0, "error": str(e), "_metadata": { "source": "tw-stock-agent", "timestamp": datetime.now().isoformat(), "data_type": "cache_management", "has_error": True } }
- Implementation of invalidate_cache in ResourceManager class that clears all or pattern-matched cache entries from the TTLCache.def invalidate_cache(self, resource_pattern: Optional[str] = None) -> int: """ Invalidate cached resources. Args: resource_pattern: Optional pattern to match specific resources Returns: Number of cache entries invalidated """ if resource_pattern is None: # Clear entire cache count = len(self.cache) self.cache.clear() logger.info(f"Invalidated entire resource cache ({count} entries)") return count else: # Clear matching entries keys_to_remove = [ key for key in self.cache.keys() if resource_pattern in key ] for key in keys_to_remove: del self.cache[key] logger.info(f"Invalidated {len(keys_to_remove)} cache entries matching '{resource_pattern}'") return len(keys_to_remove)
- mcp_server.py:307-337 (registration)Registration of the 'invalidate_cache' tool using @mcp.tool decorator in the MCP server.@mcp.tool(name="invalidate_cache", description="Invalidate resource cache for fresh data") async def invalidate_cache_tool(resource_pattern: Optional[str] = None) -> Dict[str, Any]: """Invalidate resource cache.""" try: invalidated_count = resource_manager.invalidate_cache(resource_pattern) cache_stats = resource_manager.get_cache_stats() return { "pattern": resource_pattern or "all", "invalidated_entries": invalidated_count, "cache_statistics": cache_stats, "_metadata": { "source": "tw-stock-agent", "timestamp": datetime.now().isoformat(), "data_type": "cache_management" } } except Exception as e: logger.error(f"Failed to invalidate cache: {e}") return { "pattern": resource_pattern or "all", "invalidated_entries": 0, "error": str(e), "_metadata": { "source": "tw-stock-agent", "timestamp": datetime.now().isoformat(), "data_type": "cache_management", "has_error": True } }