Maintenance and cleanup operations
maintenanceRead-only
Perform cleanup operations to manage storage and database hygiene for image generation workflows, including removing expired files, checking quota usage, and resolving inconsistencies.
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
- Core handler function for the 'maintenance' tool. Includes the @server.tool decorator, input schema via Pydantic Field annotations, operation validation, dispatching to MaintenanceService methods, and formatted ToolResult return.
@server.tool( annotations={ "title": "Maintenance and cleanup operations", "readOnlyHint": True, "openWorldHint": True, } ) 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[ int | None, Field( description="For local cleanup: maximum age in hours (default: 168 = 1 week)", ge=1, le=8760, ), ] = None, keep_count: Annotated[ int | None, 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 - Pydantic-based input schema for the maintenance tool parameters: operation (str enum), dry_run (bool), max_age_hours (int|None), keep_count (int|None).
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[ int | None, Field( description="For local cleanup: maximum age in hours (default: 168 = 1 week)", ge=1, le=8760, ), ] = None, keep_count: Annotated[ int | None, Field( description="For local cleanup: minimum number of recent files to keep", ge=1, le=1000, ), ] = None, ctx: Context = None, ) -> ToolResult: - banana_image_mcp/tools/maintenance.py:22-24 (registration)The register_maintenance_tool function that defines the nested handler and registers it via @server.tool decorator.
def register_maintenance_tool(server: FastMCP): """Register the maintenance tool with the FastMCP server.""" - banana_image_mcp/core/server.py:39-46 (registration)Server initialization calls register_maintenance_tool(self.server) to register the tool.
from ..tools.maintenance import register_maintenance_tool from ..tools.output_stats import register_output_stats_tool from ..tools.upload_file import register_upload_file_tool register_generate_image_tool(self.server) register_upload_file_tool(self.server) register_output_stats_tool(self.server) register_maintenance_tool(self.server) - Key helper method in MaintenanceService implementing the full cleanup cycle, orchestrating all maintenance operations called by the tool handler.
def full_maintenance_cycle( self, dry_run: bool = True, max_age_hours: int = 168, keep_count: int = 10 ) -> dict[str, Any]: """ Run complete maintenance cycle with all operations. Args: dry_run: If True, only report what would be done max_age_hours: For local cleanup keep_count: For local cleanup Returns: Dictionary with results from all operations """ try: self.logger.info(f"Starting full maintenance cycle (dry_run={dry_run})") results = {} # 1. Clean up expired Files API entries results["expired_cleanup"] = self.cleanup_expired_files(dry_run=dry_run) # 2. Clean up old local files results["local_cleanup"] = self.cleanup_local_files( dry_run=dry_run, max_age_hours=max_age_hours, keep_count=keep_count ) # 3. Check storage quota results["quota_check"] = self.check_storage_quota() # 4. Database hygiene results["database_hygiene"] = self.database_hygiene(dry_run=dry_run) self.logger.info("Full maintenance cycle complete") return results except Exception as e: self.logger.error(f"Full maintenance cycle failed: {e}") return {"error": str(e)}