With the emergence of AI tools, automation has gained a new level of momentum, particularly in the business world. The ability to integrate multiple tools into one cohesive business process has become increasingly appealing, and there are numerous open-source systems available that offer automation capabilities. However, many of these systems lack the ability to store customer data, which can be a significant disadvantage for businesses that rely on such data. One system that stands out in this regard is Tracardi, which offers advanced automation capabilities and allows businesses to easily store and manage customer data. With Tracardi, businesses can streamline their operations and make data-driven decisions, ultimately leading to increased efficiency and profitability.

One aspect that I particularly appreciate about Tracardi is its extensibility. By incorporating Tracardi at the core of your business process, you can avoid being locked into a single vendor since its simple code allows for easy integration with other solution providers. This extensibility enables businesses to add value to their processes and seamlessly switch between different solution providers as needed.

TLDR

In this article I am going to describe how to extend tracardi and give it a new shiny plugin.

Getting started - development environment

To start working with the system, we need to prepare a development environment. Refer to the Tracardi documentation to get started.

Introduction

Tracardi is an event-based system that processes customer actions. Events in computer science are actions or occurrences detected by software that can trigger a response, such as user interactions, system alerts, sensor readings, and messages from other software systems.

Event handling in Tracardi is based on a workflow, which consists of individual actions visualized as nodes in the workflow. Workflow control when each action/node should be triggered. The action consists of an input, a program that computes the input data, and an output (the result of the program computation). In Tracardi, an action can have one input and many outputs. In addition, the action has a configuration, it is a set of data that define how the program should behave, Let's assume that we want to connect to external resources of some database, it is in the configuration that we will have information about where this database is, and what username and password to use to connect to the resource.

Tracardi allows for multiple outputs, so it's important to specify which output should receive the data. To accomplish this, Tracardi uses the concept of ports. Ports determine where the data will be returned, and a port that doesn't receive data will return None, preventing the workflow from being executed on that branch.

Plugin life-cycle

Plugins go through a life-cycle in which they are created, executed and recycled. Workflow controls this process. When a workflow is created, the system recognizes which classes will be needed to start the process defined in the data flow graph.

The process is as follows. The workflow checks what class is assigned to the node in the graph and checks if it exists. It then creates its instances by running the parameter-less __init__ method of that class.

It then checks to see if there is an async set_up method. It passes the plug-in configuration to it. The configuration is stored inside the node and is defined during plug-in registration (more on that in a moment). Then the workflow executes nodes in the graph one by one and runs the run(self, payload: dict, in_edge=None) method, passing to it the parameters that appeared at the input to the node and the additional information on the connection from the previous node.

When the workflow exits, it executes the close method on each node.

__init__()  # (1)
async set_up(config)  # (2)
async run(input_payload)  # (3)
async close()  # (4)

1. Inits the plugin object
2. Set-ups the configuration and async resources
3. Gets the input payload as dictionary and runs the plugin, also returns results on ports
4. Closes async resources

Our first plugin

We already have all the information so let's try to write the first plugin.

All plugins inherit from the ActionRunner class. This class stores the internal state of the workflow, i.e. elements such as event data, profiles, etc. They can be useful for us while writing our plugin. Let's assume the simplest case, we would like our workflow to react to the type of event that is sent to our system. It will check if the event is of the type "my-event" and then it will return data from the input on the output named "MyEvent" otherwise it will return empty data on the port "NotMyEvent". Of-course we could use the built-in IF node, but we want to write our own.

Let’s begin

We don't need to complete all the methods mentioned above in our case. Since we don't have configuration, the set_up method is unnecessary, we don't have any connection to external systems, so the close method is not required, and we don't have an internal class state, so __init__ can be left empty.

To get started, we need to create a file where we will write the code. Tracardi plugins are located in the directory /tracardi/process_engine/action/v1. You can either create your own directory or use an existing one. In this example, we will create a directory called my_plugin_folder and place the my_plugin.py file inside it.

Now the code.

Our plugin could look like this:

File: /tracardi/process_engine/action/v1/my_plugin_folder/my_plugin.py

from tracardi.service.plugin.runner import ActionRunner
from tracardi.service.plugin.domain.result import Result

class MyPlugin(ActionRunner):  # (1)
    async def run(self, payload: dict, in_edge=None):  # (2)
       if self.event.type == "my-event":
          return Result(port="MyEvent", value=payload)  # (3)
       else:
          return Result(port="NotMyEvent", value={})  # (4)

1. Extends ActionRunner class
2. Runs the plugin
3. Returns the input payload data on the "MyEvent" port
4. Returns the empty dictionary on the "NotMyEvent" port

This plugin class extends the ActionRunner class and defines the run method that processes input data and returns it on one of the two output ports: "MyEvent" or "NotMyEvent", depending on the type of the event.

The Result class is used to wrap the output data and its destination port.

The only thing left for us to do is to describe the plugin in the system. This is done by defining a function called register. It contains the specification of the plugin that we wrote (it returns the plugin class) and the type metadata, with the input and output ports, the name of the plugin, etc.

The register function can be placed in the same file as the plugin or in any other file. I am placing it in the same file.

Example:

File: /tracardi/process_engine/action/v1/my_plugin_folder/my_plugin.py

    from tracardi.service.plugin.domain.register import Plugin, Spec, MetaData

    def register() -> Plugin:
        return Plugin(   # (1)
            start=False,
            spec=Spec(  # (2)
                module=__name__,  # (4)
                className=MyPlugin,
                inputs=["payload"],
                outputs=["MyEvent", "NotMyEvent"],
                version='0.1',
                license="MIT",
                author="Your Name"
            ),
            metadata=MetaData(  # (3)
                name="My first plugin",
                desc='Checks if the event type is equal to my-event.',
                group=["Test plugin"]
            )
        )

1. Returns Plugin class
2. Sets spec property as Spec class
3. Sets metadata property as Metadata class
4. Sets __name__ because refister is in the same file as plugin: e.i. /tracardi/process_engine/action/v1/my_plugin_folder/my_plugin.py

Let's analyze this code. It returns a plugin class that has the following properties.

• start - sets whether the workflow can start from this node. In 99% of cases, we put False here. The startup nodes are already built into the system.
• spec - describes the plugin specification, i.e. what class is to be run and what ports it contains on input and output. There can only be one port on input. In our case, we have the following data:
• module - where is the class package. __name__ means that the class is in the same file as the register method. If we separate the plugin and register function, then you need to enter the package name of the plugin here, e.g. tracardi.process_engine.action.v1.my_plugin_folder
• className - the name of the class. We named it MyPlugin. See class MyPlugin (ActionRunner)
• inputs - list with the names of the input ports. There can only be one input port.
• outputs - list with names of output ports. Here we define what ports we have. Port names can have any name you like. Remember, however, that they must correspond to what the code returns and our code returns, one time: Result (port="MyEvent", value=payload) and another time Result (port="NotMyEvent", value={}), i.e. possible output ports are ["MyEvent", "NotMyEvent"]
• version - enter the plugin version here
• license - license type, Tracardi is able to attach the plug-in only under the MIT or Apache 2.0 license
• author - author's first and last name
• metadata - contains additional data about the plugin.
• name - the name of the plugin to be displayed on the workflow graph
• desc - a short description of what the plugin does
• group - plugins are displayed in groups. The name of the group in which the plug-in is to be displayed on the plug-in list. The name can be any or one of the existing names.

Automatic plug-in loading.

The only thing left is to register the plug-in on the list of available plug-ins for installation. We do this by pointing to the file with the register function.

To do that, go to the directory: /tracardi/service/setup and find the file setup_plugins.py This is the list of all available plugins in the system.

At the top of this file you will find the variable installed_plugins: Dict [str, PluginTestTemplate] which is a dictionary where the key is the location of the register function. The value is an object of the PluginTestTemplate type, it is responsible for the test data for the plug-in. We will not write tests, so our block of code should look like this:

File: /tracardi/service/setup/setup_plugins.py

    "tracardi.process_engine.action.v1.my_plugin_folder.my_plugin": PluginTestTemplate(  # (1)
        init=None,
        resource=None
    ),

1. Key is the package of the register function

Type this into the installed_plugins dictionary and we are ready to install the plugin.

Restart the Tracardi API so the changes are activated and go to Processing/Workflows, open any workflow and click the Reinstall Plugins button. Alternatively you can go to Maintenance/Plug-ins and click the Reinstall Plugins button.

Wrap-up

And this concludes the first part of the tutorial. We added the first plugin and installed it. In the second part we will extend our plugin with the configuration form.

Follow up

1. Part 2: Configuring the plugin