Create Custom Integrations

Suggest Edits

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

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

Create a Custom Integration

On the left navigation, go to Automations > Integrations.
Click Create Custom Integration from the top-right corner of the pop-up window.
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.

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:

On the Starter Integration page, click Packages in the top menu bar.
To specify packages manually, click Enter Manually and specify packages with name and version number and click Save.
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

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:

On the Starter Integration page, click Test Data.
- A window opens with a test file. You can modify the file as needed for testing.
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.

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.
Choose the action to test. The Additional settings are shown.
Enter the column to map to the input string.
Enter the action parameter
Specify any other optional criteria for the text, including timeout, rows, and threads to process.
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, including the integration name, description, and logo URL, can be defined in a top-level doc string within any one of the .py files. If specified in multiple files, the platform will select any one of them.

"""
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 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.

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


@action(name="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"
    }

Updated about 1 year ago

What's Next

Share Use Cases with Others