How-To Integrate a 3rd Party Tool - Part 5: API Wrapper


Now that we understand the benefit of Connection Info objects and have created an Admin extension, its time to discuss executing remote calls to the 3rd party tool. This is most commonly done with a REST API and that is what this article will cover. We consider the “Admin Page” UI Extension (XUI) to be the home for credentials and general settings of our integration. It is also the perfect place to insert common code that will be used in the extension itself or called from CloudBolt (CB) plugins. We are going to build an API Wrapper “” in the XUI so we do not have to duplicate this code in multiple locations. Lets get started.

This is also a great example for any general common code you might need for plugins. Consider creating an “Admin Page” XUI that simply stores common code that is imported by your plugins.


  • You must be familiar with REST APIs

  • You will need access to the API development documentation for the 3rd party tool

  • Not all REST APIs are the same and this also goes for the authentication method used by the API. The most common type is basic auth and bearer token. Basic auth is used to get a token and the token is used for all additional requests.

  • An intermediate knowledge with Python and the Requests library is ideal

  • The provided sample will support the OOTB PROXY and SSL features

Locate The Admin Extension

  1. Go to Admin > Manage UI Extensions

  2. Expand the extension to see the file path and edit the files. Browse to this file path on the CloudBolt instance.

Add “”

This is a sample only. You will need to follow the API documentation of the 3rd party tool to update the login method, base_url, etc.

Sample API Wrapper
import requests
from utilities.helpers import get_ssl_verification
from utilities.models import ConnectionInfo
from utilities.logger import ThreadLogger

logger = ThreadLogger(__name__)
conn_name = 'Example Integration'

class XUIAPIWrapper(object):

    def __init__(self):
        Very simply initialization for the class.
        No arguements are required, the Connection Info object is looked up.
        # get the Connection Info object
        ci, _ = ConnectionInfo.objects.get_or_create(name=conn_name)
        self.connection_info = ci
        # configure the base url from the attributes in the Connection Info object
        self.BASE_URL = f'{ci.protocol}://{ci.ip}:{ci.port}/api/v1'
        # the headers can also be defined in the Connection Info object
        self.headers = {'Content-Type': 'application/json'}
        # create a placeholder to store the token
        self.token = ''

     def _run(self, method, url, auth=None, json=None, params=None, headers=None):
         Generic run method for the Requests library

         :param url: URL for the new :class:`Request` object.
         :param auth: (optional)Basic auth `Tuple` object.
         :param json: (optional) json data to send in the body of the :class:`Request`.
         :param params: (optional) dictionary of parameters for the request.
         :param headers: (optional) dictionary of header options for the request.
         :return: :class:`Response <Response>` object
         if not headers:
             headers = self.headers
         response = requests.request(
             method, url, auth=auth, json=json, params=params,
             headers=headers, verify=get_ssl_verification())
         if response.status_code not in [200, 201, 204]:
             # The API documentation of your 3rd party tool should define the error codes and associated messaging.
             # Add additional codes below for customized error handling messages
             if response.status_code in [400]:
                 raise Exception(
                     f'{response.text} The specified request returned a bad request error, requires an argument or \
                     requires only a single argument'
             raise Exception(
                 f'Return Code: {response.status_code}, Response Text: {response.text}s'

         return response

     def _get(self, url, params=None, headers=None):
         return self._run('GET', url, params=params, headers=headers)

     def _post(self, url, json=None, params=None, headers=None):
         return self._run('POST', url, json=json, params=params, headers=headers)

     def _patch(self, url, json=None, params=None, headers=None):
         return self._run('PATCH', url, json=json, params=params, headers=headers)

     def _delete(self, url, json=None, headers=None):
         return self._run('DELETE', url, json=json, headers=headers)

     def login(self):
         url = self.AUTH_URL
         # create a copy of the header and append the auth specific header options from the API documentation
         headers = self.headers.copy()
         headers['Content-Type'] = 'application/x-www-form-urlencoded'
         auth = (self.connection_info.username, self.connection_info.password)
         response = self._run('POST', url=url, auth=auth, headers=headers)
         r_json = response.json()
         self.token = r_json.get('token')
         self.headers['Authorization'] = f'Bearer {self.token}'
         return response

     def list_objects(self, obj_type):
         url = f'{self.BASE_URL}/{obj_type}'
         resources = []
         params = {"order_by": "name"}
         response = self._get(url, params=params).json()
         for resource in response['resources']:
         next = response['pagination']['next']
         while next:
             response = self._get(next, params=params).json()
             for resource in response['resources']:
         return resources

     def create_object(self, obj_type, name=None, params={}):
         url = f'{self.BASE_URL}/{obj_type}'
         if name:
             params['name'] = name
         response = self._post(url, params=params).json()
         return response

     def update_object(self, obj_type, name, guid, params={}):
         url = f'{self.BASE_URL}/{obj_type}/{guid}'
         params['name'] = name
         response = self._patch(url, params=params).json()
         return response

     def delete_object(self, obj_type, guid):
         url = f'{self.BASE_URL}/{obj_type}/{guid}'
         response = self._delete(url).json()
         return response

     def list_organizations(self):
         List ALL Organizations
         return self.list_objects(obj_type='organizations')
     def create_organization(self, name):
             name: (STR) Required: New Organization name
         return self.create_object(obj_type='organizations', name=name)

Example plugin usage of “”

from xui.example_extension.api_wrapper import XUIAPIWrapper

client = XUIAPIWrapper()

Additional information

CloudBolt UI Extension documentation :!cloudbolt-latest-docs/ui-extensions

Have more questions? Submit a request


Please sign in to leave a comment.