Using third party APIs often requires some boilerplate like:
Using connectors, you let Zenaton handle all of this for you so you can focus on what matter: your business workflow.
If you need to interact with an API for which we do not have a connector yet or because it's one of your internal APIs.
It's easy as wrap the logic into a Zenaton task skeleton. So then you can re-use it in many places.
In our example, we will create a simple Task to send Slack message using the Slack SDK.
If the API does not have an SDK, then you would need to add the http calls to the API in your task.
Here are the steps we will follow:
Let's walk through an example of creating a task to the Slack API so that we can send a message to slack using the Slack SDK.
Add a Zenaton task boilerplate, into a file:
from zenaton.abstracts.task import Task from zenaton.traits.zenatonable import Zenatonable class SlackTask(Task, Zenatonable): def handle(self): # Your task implementation
We will use the Slack SDK, so we will need to add it to our codebase dependency.
pip3 install slackclient
From their doc , we simply copy their snippet:
import os import slack client = slack.WebClient(token=os.environ['SLACK_API_TOKEN']) response = client.chat_postMessage(channel='#random', text="Hello world!")
And turn it into a our new Zenaton task by integrating the snippet into your task file.
The final code is:
import os import slack from zenaton.abstracts.task import Task from zenaton.traits.zenatonable import Zenatonable class SlackTask(Task, Zenatonable): def handle(self): client = slack.WebClient(token=os.environ['SLACK_TOKEN']) client.chat_postMessage(channel='#my-channel', text="Hello from a Zenaton task!")
We will add a
SLACK_TOKEN environment variable to the .env file used by our agent.
Note that we will need to unlisten and then listen to take this new parameter into account.
ZENATON_APP_ID=XXXXXX ZENATON_API_TOKEN=XXXXXXXXXXXXXXXXXXXX ZENATON_APP_ENV=dev SLACK_TOKEN=xoxp-xxxxxxxx
Now we can test our task by running this launch script
import client from tasks.slack_task import SlackTask SlackTask().dispatch()
Update your boot file:
... from tasks.slack_task import SlackTask
Then run the task.
To make it more reusable, we should add some parameters to our SlackTask so that we can customize the slack message and the channel.
... class SlackTask(Task, Zenatonable): def __init__(self, text, channel): self.text = text self.channel = channel def handle(self): client = slack.WebClient(token=os.environ['SLACK_TOKEN']) client.chat_postMessage(channel=self.channel, text=self.text)
So, now we can use it this way:
SlackTask("Hello from My SlackTask", "my-channel").dispatch()
We can update our
launch_slack_task.py to test the task:
import client from tasks.slack_task import SlackTask SlackTask("Hello from My SlackTask", "my-channel").dispatch()
Then run the task
We can now re-use this task in many of our workflows !
Automatic retry — As APIs may have downtime, we could add an automatic retry to our API call so that it will automatically try again if it fails to connect.
Scheduling — And, if we plan to execute our API call on a recurring basis, we could set up a schedule for it using cron expressions.