Create Custom Integrations

LogicHub provides the following ways to include integrations in your playbooks.

  • Add an Integration from the LogicHub library.
  • Create your own integration from a starter .py integration file.

Let's look at how to create an integration from the starter.py file.

Download the Starter Integration File

  1. On the left navigation, go to Automations > Integrations.
  2. Search for the integration that you want to use to start your custom integration.
  3. Click the More (...) icon and select Download as zip.
  4. Unzip the downloaded starter zip file from your local directory.
  5. On your development machine, run pip install lhub_integ to get the LogicHub integration packages.

Create a Custom Integration

  1. On the left navigation, go to Automations > Integrations.
  2. Click Create Custom Integration from the top-right corner of the pop-up window.
  3. Click Start with Starter Integration.
    • The configuration page opens to show the Python editor on the left and the Details panel on the right. The Details panel shows the connection parameters and action parameters specified in the Python file.
  1. Perform any of the following actions on this page.
    • Change the integration name. Click the StarterIntegration label, enter a new name, and press Return.
    • Modify the python code. Make edits to the code, then press Refresh in the Details panel to update the information there.
    • Add new files. Click + in the top tab area.
    • Rename files. Click the file name, enter the new name, and press Return.

Add External Packages

You can include any external packages that are needed to run the integration. To access the options:

  1. On the Starter Integration page, click Packages in the top menu bar.
  2. To specify packages manually, click Enter Manually and specify packages with name and version number and click Save.
  3. To upload a requirements file, click Upload requirements.txt or bundle.tar.bz2 and select the file from the browser to upload.

Manage Integration Files

  1. On the Starter Integration page, click Files in the top menu bar.
    • The files that were created using the Python editor are shown, and you can edit or delete them. You can also upload files to include in the package.

Test the Integration

To validate the integration with a test file:

  1. On the Starter Integration page, click Test Data.
    • A window opens with a test file. You can modify the file as needed for testing.
  2. Click Save and Exit to run the test and display the results in the right-side panel.

To test the integration, set up a temporary connection to a data source.

  1. On the Starter Integration page, click Configure.
    • Enter the following details in the configuration window.
      • Enter a label to identify the test in the UI.
      • Select the remote agent to run the integration test, or select None if you don’t want to use a remote agent.
      • To use an SSL certificate to verify the connection, select Verify SSL Certificate. To skip the verification, select Skip Verifying SSL Certificate.
      • Enter a parameter for the connection and click Next.
  2. Choose the action to test. The Additional settings are shown.
  3. Enter the column to map to the input string.
  4. Enter the action parameter
  5. Specify any other optional criteria for the text, including timeout, rows, and threads to process.
  6. Click Submit.

The integration and associated action run and the results are shown in the output Data tab in the Results pane.

You can save the integration at any time by clicking Save at the bottom of the screen. When you’re ready to make the integration available to your organization, click Publish.

Custom Integration Features

The following file contains instructions for creating a custom integration.

# A top level __doc__ string in any of the files can be used to set meta information for the entire integration.
# Information in required format must be specified in only one of the files (any one file).
# Other files can contain __doc__ strings but it shouldn't be in the following format so that meta information is picked up in a predicatable manner.
"""
name: Custom Integration
description: Custom Integration Description
logoUrl: https://s3.amazonaws.com/lhub-public/integrations/default-integration-logo.svg
"""

# `lhub_integ` is already installed as a dependency in the container
# For testing, you can do `pip install -e <path to backend/custom-integrations/lhub-integ`
from lhub_integ.params import ConnectionParam, ActionParam
from lhub_integ import action

# ConnectionParams and ActionParams must be declared at the top level.
# To get the value of the Parameter call param_name.read()

# ConnectionParam and ActionParam can take the values `optional=[True/False]` and `default="some_default_string"`
"""
ConnectionParam and ActionParam must be called with an identifier
Other optional named parameters:
description="description shown in the UI"
label="label shown for parameter in UI"
optional=[True/False]
default="some_default_string"

Advanced usage:
data_type=DataType.[STRING/COLUMN/NUMBER] - defaults to STRING
input_type=InputType.[TEXT/TEXT_AREA/EMAIL/PASSWORD/SELECT/COLUMN_SELECT] - defaults to TEXT
options= [list of options in a drop down]

"""

PASSWORD = ConnectionParam("PASSWORD", description="Password for JIRA")
URL = ConnectionParam("URL", description="URL for the server")

# ActionParam should be called with an ‘action’ argument to specify what action it is used for
PROJECT = ActionParam("PROJECT", description="Jira Project", action="process")

# accepts "process,process1" or ["process","process1"] or ("process","process1")
SHARED_AP = ActionParam("ACTION_PARAM", description="Parameter for your action", action=["process", "process1"])


# A main function for the integraion must be defined. By default, we'll look for a function called `process` in `main.py`
# but this can be overriden.

# Every method with `@action` will be converted into an action, adding (“...”) will define a name for the action 
@action(“Process”)
# adding a type int will cause the shim to coerce types properly
def process(summary, issue_type: int):
    """
    File a JIRA issue
    :param summary: Summary of the JIRA issue
    :label issue_type: Issue Type
    :param issue_type: The type of JIRA issue
    :return:
    """
    # ^^^ An optional doc string may be provided in the standard Python docstring format.
    # If provided, it will be used to set the descriptions and labels for arguments in the integration
    
   
    # ... actually file a JIRA issue or something
    
    # The function must return either:
    # 1. A Python dictionary that is JSON serializable
    # 2. A list of Python dictionaries that are JSON serializable
    return {
        "status": "OK",
        "url": URL.read(),   # reading from the parameter
        "shared_action_param": SHARED_AP.read()
    }

@action("Process1")
def process():
    return {
      "shared_action_param": SHARED_AP.read()
    }

Write your Own Custom Integration

Let's write a Slack Integration from scratch.

Integration Metadata

Integration metadata such as the integration name, description, and logoUrl can be specified as a top-level doc string in any one of the .py files.

"""
name: Slack
description: Slack is a cloud-based set of proprietary team collaboration tools and services.
logoUrl: https://s3.amazonaws.com/lhub-public/integrations/slack.svg
"""

ConnectionParams and ActionParams

  • ConnectionParam: This class is defined in the lhub_integ.params module. It helps you define connection parameters/inputs, such as API-Key or Host-URL, which are required to create a connection for your custom integration.
  • ActionParam: This class is defined in the lhub_integ.params module. It helps you define action parameters/inputs, such as the channel name to post messages for Slack, which are required for the integration action.
  • Bonus: You can share an ActionParam with multiple actions. Supply an array of function_names to the action **kwargs (Keyword Arguments).
from lhub_integ.params import ConnectionParam, ActionParam, DataType, InputType


SLACK_INCOMING_WEBHOOK = ConnectionParam(
    id="slack_incoming_webhook",
    label="Incoming Webhook URL",
    description="Incoming Webhook URL for Slack",
    optional=False,
    options=None,
    data_type=DataType.STRING,
    input_type=InputType.PASSWORD
)

SLACK_CHANNEL_NAME = ActionParam(
    id="slack_channel_name",
    label="Channel Name",
    description="Incoming Webhook has a default channel, but it can be overridden e.g. #general, #dev etc.",
    action=['process_post_message'],
    optional=True,
    options=None,
    data_type=DataType.STRING,
    input_type=InputType.TEXT
)
SLACK_TIME_BETWEEN_CONSECUTIVE_REQUESTS_MILLISECONDS = ActionParam(
    id="slack_time_between_consecutive_requests_milliseconds",
    label="Time between consecutive API requests (in millis)",
    description="Time to wait between consecutive API requests in milliseconds (Default is 0 millisecond)",
    action=['process_post_message'],
    optional=True,
    options=None,
    data_type=DataType.INT,
    input_type=InputType.TEXT,
    default="0"
)

Define a Connection Validator (optional, but recommended)

Write a function and annotate it with @connection_validator. This option checks the validity of the connection. For example, you can check whether the input credentials entered by the integration user are valid.

For Slack Integrations, you can make an API call (using the requests library) to check whether the value entered by the user for Incoming Webhook URL is valid.

📘

Note

Utilities are available to perform validity or sanity checks, including validations, input_helpers, and helpers. You can explore them or you can write your own sanity checks.

import requests
from lhub_integ.common import helpers, validations
from lhub_integ import connection_validator


def post_message_on_slack(incoming_webhook, message, channel_name=''):
    payload = {'text': message}
    if channel_name:
        payload['channel'] = channel_name
    response = requests.post(incoming_webhook, json=payload, headers={'Content-Type': 'application/json'})
    response.raise_for_status()


@connection_validator
def validate_connections():
    if not SLACK_INCOMING_WEBHOOK.read():
        return [ValidationError(message="Parameter must be defined", param=SLACK_INCOMING_WEBHOOK)]

    try:
        post_message_on_slack(SLACK_INCOMING_WEBHOOK.read(), 'This is a test message from LogicHub, Inc.')
    except Exception as ex:
        return [ValidationError(message=f"Incorrect Parameter: {repr(ex)}", param=SLACK_INCOMING_WEBHOOK)]

Define an Integration Action

For Slack integration, an example action is to post a message on a specified channel.

import time
from lhub_integ.params import JinjaTemplatedStr
from lhub_integ import action


def validate_post_message():
    pass
 

@action(name="Post Message", validator=validate_post_message)
def process_post_message(message: JinjaTemplatedStr):
    """
    Post a message on Slack

    :param message: Jinja-templated message string that will be posted on slack. Eg: '{{message_col}}'
    :label message: Message Template
    :optional message: False
    """
    post_message_on_slack(incoming_webhook=SLACK_INCOMING_WEBHOOK.read(), message=message, channel_name=SLACK_CHANNEL_NAME.read())

    time_delay = helpers.convert_milliseconds_to_seconds(SLACK_TIME_BETWEEN_CONSECUTIVE_REQUESTS_MILLISECONDS.read())
    time.sleep(time_delay)
    return {
        "result": "Successfully posted message to Slack"
    }
  • validator: an optional function (similar to connection_validator) that validates the inputs provided to the action.
  • :param, :label and :optional are provided on the argument of the action function specifies the input's metadata.

Advanced Usage

If an integration requires dynamic descriptors based on connection, they can be implemented (in your main.py file) as shown in the below example:

import json
from lhub_integ.common import input_helpers
from lhub_integ import connection_validator

def override_action_descriptors():
    descriptor_json = input_helpers.get_stripped_env_string('__integration_descriptor')

    # Custom value to be overridden at connection validation
    new_description = "My Custom Dynamic Description"

    descriptor = json.loads(descriptor_json)
    actions = descriptor['actions']
    actions[0]['instantiation']['steps'][0]['inputs'][0]['description'] = new_description
    return {"actionOverrides": actions}


@connection_validator
def validate_connections():
    result = json.dumps(override_action_descriptors())
    print("[result] {}".format(result))

Summing It All Up

"""
name: Slack
description: Slack is a cloud-based set of proprietary team collaboration tools and services.
logoUrl: https://s3.amazonaws.com/lhub-public/integrations/slack.svg
"""

import time
import requests
from lhub_integ.common import helpers
from lhub_integ.params import ConnectionParam, ActionParam, DataType, InputType, JinjaTemplatedStr, ValidationError
from lhub_integ import action, connection_validator


SLACK_INCOMING_WEBHOOK = ConnectionParam(
    id="slack_incoming_webhook",
    label="Incoming Webhook URL",
    description="Incoming Webhook URL for Slack",
    optional=False,
    options=None,
    data_type=DataType.STRING,
    input_type=InputType.PASSWORD
)

SLACK_CHANNEL_NAME = ActionParam(
    id="slack_channel_name",
    label="Channel Name",
    description="Incoming Webhook has a default channel, but it can be overridden e.g. #general, #dev etc.",
    action=['process_post_message'],
    optional=True,
    options=None,
    data_type=DataType.STRING,
    input_type=InputType.TEXT
)
SLACK_TIME_BETWEEN_CONSECUTIVE_REQUESTS_MILLISECONDS = ActionParam(
    id="slack_time_between_consecutive_requests_milliseconds",
    label="Time between consecutive API requests (in millis)",
    description="Time to wait between consecutive API requests in milliseconds (Default is 0 millisecond)",
    action=['process_post_message'],
    optional=True,
    options=None,
    data_type=DataType.INT,
    input_type=InputType.TEXT,
    default="0"
)


def post_message_on_slack(incoming_webhook, message, channel_name=''):
    payload = {'text': message}
    if channel_name:
        payload['channel'] = channel_name
    response = requests.post(incoming_webhook, json=payload, headers={'Content-Type': 'application/json'})
    response.raise_for_status()


@connection_validator
def validate_connections():
    if not SLACK_INCOMING_WEBHOOK.read():
        return [ValidationError(message="Parameter must be defined", param=SLACK_INCOMING_WEBHOOK)]

    try:
        post_message_on_slack(SLACK_INCOMING_WEBHOOK.read(), 'This is a test message from LogicHub, Inc.')
    except Exception as ex:
        return [ValidationError(message=f"Incorrect Parameter: {repr(ex)}", param=SLACK_INCOMING_WEBHOOK)]

def validate_post_message():
    pass


@action(name="Post Message", validator=validate_post_message)
def process_post_message(message: JinjaTemplatedStr):
    """
    Post a message on Slack

    :param message: Jinja-templated message string that will be posted on slack. Eg: '{{message_col}}'
    :label message: Message Template
    :optional message: False
    """
    post_message_on_slack(incoming_webhook=SLACK_INCOMING_WEBHOOK.read(), message=message, channel_name=SLACK_CHANNEL_NAME.read())

    time_delay = SLACK_TIME_BETWEEN_CONSECUTIVE_REQUESTS_MILLISECONDS.read() / 1000.0
    time.sleep(time_delay)
    return {
        "result": "Successfully posted message to Slack"
    }

Did this page help you?