Maintenance and cleanup operations
maintenanceRead-only
Perform maintenance operations to manage storage and database health, including cleaning expired files, checking quotas, 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 for the 'maintenance' tool: decorated with @server.tool(), defines input schema, dispatches to MaintenanceService based on 'operation' param, formats markdown summaries, returns structured ToolResult.
@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[ 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 - nanobanana_mcp_server/core/server.py:39-44 (registration)Registration point where register_maintenance_tool is imported and invoked to add the maintenance tool to the FastMCP server instance.
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) - Tool schema: @server.tool annotations (title, hints) and Pydantic input parameters with Field descriptions, validation (ge/le), defaults.
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[ 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: - Helper function to retrieve the singleton MaintenanceService instance used by the handler for actual cleanup operations.
def _get_maintenance_service(): """Get the maintenance service instance.""" from ..services import get_maintenance_service return get_maintenance_service() - Key helper method in MaintenanceService: full_maintenance_cycle orchestrates all maintenance operations called by the tool's 'full_cleanup' operation.
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(f"Full maintenance cycle complete") return results except Exception as e: self.logger.error(f"Full maintenance cycle failed: {e}") return {"error": str(e)}