nanobanana-mcp-server

Instructions

Implementation Reference

• nanobanana_mcp_server/tools/maintenance.py:23-142 (handler)

  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)

• nanobanana_mcp_server/tools/maintenance.py:24-59 (schema)

  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:

• nanobanana_mcp_server/tools/maintenance.py:263-267 (helper)

  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()

• nanobanana_mcp_server/services/maintenance_service.py:251-290 (helper)

  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)}