Taiwan Stock Agent

Instructions

Implementation Reference

• mcp_server.py:307-337 (handler)

  MCP tool handler for 'invalidate_cache' that delegates to resource_manager.invalidate_cache and formats the response with cache 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
              }
          }

• tw_stock_agent/services/mcp_resource_service.py:307-332 (helper)

  Core implementation of cache invalidation in ResourceManager class, clearing all or pattern-matched cache entries.

  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)