maintenance
Perform maintenance operations to manage storage and database hygiene, including cleanup of expired files, local file removal, quota checking, and database consistency fixes.
Instructions
Perform maintenance operations following workflows.md patterns.
Available operations:
cleanup_expired: Remove expired Files API entries from database
cleanup_local: Clean old local files based on age/LRU
check_quota: Check Files API storage usage vs. ~20GB budget
database_hygiene: Clean up database inconsistencies
full_cleanup: Run all cleanup operations in sequence
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| operation | Yes | Maintenance operation to perform: 'cleanup_expired', 'cleanup_local', 'check_quota', 'database_hygiene', 'full_cleanup' | |
| dry_run | No | If true, only report what would be done without making changes | |
| max_age_hours | No | For local cleanup: maximum age in hours (default: 168 = 1 week) | |
| keep_count | No | For local cleanup: minimum number of recent files to keep |
Implementation Reference
- The main handler function 'maintenance' for the MCP tool, which parses the operation parameter, fetches the maintenance service, executes the appropriate cleanup or check operation, formats a detailed text summary, and returns a ToolResult with structured content.def maintenance( operation: Annotated[ str, Field( description="Maintenance operation to perform: " "'cleanup_expired', 'cleanup_local', 'check_quota', 'database_hygiene', 'full_cleanup'" ), ], dry_run: Annotated[ bool, Field(description="If true, only report what would be done without making changes"), ] = True, max_age_hours: Annotated[ Optional[int], Field( description="For local cleanup: maximum age in hours (default: 168 = 1 week)", ge=1, le=8760, ), ] = None, keep_count: Annotated[ Optional[int], Field( description="For local cleanup: minimum number of recent files to keep", ge=1, le=1000, ), ] = None, ctx: Context = None, ) -> ToolResult: """ Perform maintenance operations following workflows.md patterns. Available operations: - cleanup_expired: Remove expired Files API entries from database - cleanup_local: Clean old local files based on age/LRU - check_quota: Check Files API storage usage vs. ~20GB budget - database_hygiene: Clean up database inconsistencies - full_cleanup: Run all cleanup operations in sequence """ logger = logging.getLogger(__name__) try: logger.info(f"Maintenance operation: {operation}, dry_run={dry_run}") # Get services (would be injected in real implementation) maintenance_service = _get_maintenance_service() # Validate operation valid_operations = [ "cleanup_expired", "cleanup_local", "check_quota", "database_hygiene", "full_cleanup", ] if operation not in valid_operations: raise ValidationError( f"Invalid operation. Must be one of: {', '.join(valid_operations)}" ) # Execute maintenance operation if operation == "cleanup_expired": result = maintenance_service.cleanup_expired_files(dry_run=dry_run) summary = _format_expired_cleanup_summary(result, dry_run) elif operation == "cleanup_local": result = maintenance_service.cleanup_local_files( dry_run=dry_run, max_age_hours=max_age_hours or 168, # 1 week default keep_count=keep_count or 10, # Keep at least 10 recent files ) summary = _format_local_cleanup_summary(result, dry_run) elif operation == "check_quota": result = maintenance_service.check_storage_quota() summary = _format_quota_summary(result) elif operation == "database_hygiene": result = maintenance_service.database_hygiene(dry_run=dry_run) summary = _format_database_hygiene_summary(result, dry_run) elif operation == "full_cleanup": result = maintenance_service.full_maintenance_cycle( dry_run=dry_run, max_age_hours=max_age_hours or 168, keep_count=keep_count or 10 ) summary = _format_full_cleanup_summary(result, dry_run) else: # This shouldn't happen due to validation above raise ValidationError(f"Unhandled operation: {operation}") content = [TextContent(type="text", text=summary)] structured_content = { "operation": operation, "dry_run": dry_run, "workflow": "workflows.md_maintenance_sequence", "result": result, "parameters": {"max_age_hours": max_age_hours, "keep_count": keep_count}, } logger.info(f"Maintenance operation {operation} completed successfully") return ToolResult(content=content, structured_content=structured_content) except ValidationError as e: logger.error(f"Validation error in maintenance: {e}") raise except Exception as e: logger.error(f"Unexpected error in maintenance: {e}") raise
- Input schema definitions using Annotated and Field for the tool parameters: operation (str, enum-like), dry_run (bool), max_age_hours (Optional[int]), keep_count (Optional[int]).operation: Annotated[ str, Field( description="Maintenance operation to perform: " "'cleanup_expired', 'cleanup_local', 'check_quota', 'database_hygiene', 'full_cleanup'" ), ], dry_run: Annotated[ bool, Field(description="If true, only report what would be done without making changes"), ] = True, max_age_hours: Annotated[ Optional[int], Field( description="For local cleanup: maximum age in hours (default: 168 = 1 week)", ge=1, le=8760, ), ] = None, keep_count: Annotated[ Optional[int], Field( description="For local cleanup: minimum number of recent files to keep", ge=1, le=1000, ), ] = None, ctx: Context = None, ) -> ToolResult:
- nanobanana_mcp_server/core/server.py:39-44 (registration)Registration of the maintenance tool: imports register_maintenance_tool from tools.maintenance and calls it with the FastMCP server instance in the _register_tools method.from ..tools.maintenance import register_maintenance_tool register_generate_image_tool(self.server) register_upload_file_tool(self.server) register_output_stats_tool(self.server) register_maintenance_tool(self.server)
- Helper function _get_maintenance_service that imports and calls get_maintenance_service() to obtain the service instance used by the handler.def _get_maintenance_service(): """Get the maintenance service instance.""" from ..services import get_maintenance_service return get_maintenance_service()
- MaintenanceService class providing the backend methods (cleanup_expired_files, cleanup_local_files, etc.) called by the tool handler.class MaintenanceService: """Service for maintenance and cleanup operations following workflows.md patterns.""" def __init__( self, files_api_service: FilesAPIService, db_service: ImageDatabaseService, out_dir: str ): """ Initialize maintenance service. Args: files_api_service: Files API service db_service: Database service out_dir: Output directory path """ self.files_api = files_api_service self.db_service = db_service self.out_dir = out_dir self.logger = logging.getLogger(__name__) def cleanup_expired_files(self, dry_run: bool = True) -> Dict[str, Any]: """ Clean up expired Files API entries from database. Implements: "Scan DB for Files API expirations (~48h TTL)" Args: dry_run: If True, only report what would be cleaned Returns: Dictionary with cleanup statistics """ try: self.logger.info(f"Starting expired Files API cleanup (dry_run={dry_run})") # Delegate to FilesAPIService result = self.files_api.cleanup_expired_files(dry_run=dry_run) self.logger.info(f"Expired cleanup complete: {result}") return result except Exception as e: self.logger.error(f"Expired files cleanup failed: {e}") return {"expired_count": 0, "cleared_count": 0, "errors": [str(e)]} def cleanup_local_files( self, dry_run: bool = True, max_age_hours: int = 168, # 1 week keep_count: int = 10, ) -> Dict[str, Any]: """ Clean up old local files based on age and LRU policy. Implements: "Local LRU/age-based cleanup of OUT_DIR" Args: dry_run: If True, only report what would be cleaned max_age_hours: Files older than this are candidates for removal keep_count: Always keep at least this many recent files Returns: Dictionary with cleanup statistics """ try: self.logger.info( f"Starting local file cleanup (dry_run={dry_run}, " f"max_age_hours={max_age_hours}, keep_count={keep_count})" ) stats = { "total_files": 0, "removed_count": 0, "kept_count": 0, "freed_mb": 0.0, "errors": [], } # Get all image files in output directory image_files = [] for ext in ["*.jpg", "*.jpeg", "*.png", "*.webp"]: image_files.extend(Path(self.out_dir).glob(f"**/{ext}")) # Sort by modification time (newest first) image_files.sort(key=lambda p: p.stat().st_mtime, reverse=True) stats["total_files"] = len(image_files) # Cutoff time for old files cutoff_time = datetime.now() - timedelta(hours=max_age_hours) cutoff_timestamp = cutoff_time.timestamp() removed_count = 0 freed_bytes = 0 for i, file_path in enumerate(image_files): try: file_stat = file_path.stat() # Always keep the most recent files if i < keep_count: continue # Check if file is old enough to remove if file_stat.st_mtime > cutoff_timestamp: continue # Check if file is still referenced in database db_record = self.db_service.get_by_path(str(file_path)) if db_record and db_record.file_id: # File is still referenced in Files API, keep it self.logger.debug(f"Keeping referenced file: {file_path}") continue # File is eligible for removal file_size = file_stat.st_size if not dry_run: file_path.unlink() # Also remove corresponding thumbnail if it exists thumb_path = file_path.with_name(file_path.stem + "_thumb.jpeg") if thumb_path.exists(): thumb_size = thumb_path.stat().st_size thumb_path.unlink() file_size += thumb_size