Python Worker SDK

overview »

Humatron provides a Python Worker SDK designed for building AI workers and applications using Python programming language. This SDK is essentially a convenient wrapper around the underlying Worker API. You can access the latest version of the Python Worker SDK on https://pypi.org/project/humatron-python-sdk.

The Python Worker SDK is logically divided into two distinct modules: the Worker SDK and the REST Channel SDK. While both modules are bundled into a single library, they are fully decoupled, allowing developers to use each independently based on their needs. Whether you're building autonomous AI workers or integrating complex REST-based workflows, Python Worker SDK provides the necessary APIs.

Installation »

To install the latest version of Python Worker SDK:

pip install humatron-python-sdk locked-dict flask requests

NOTE: flask dependency is optional and used for examples only.

worker SDK »

Python Worker SDK’s primary goal is to offer a convenient and efficient framework for building AI workers in Python, while simplifying and abstracting the underlying Worker API protocol complexities. Refer to the protocol details and terminology nomenclature here.

requests »

The communication between your AI worker and the Humatron service is based on the REST protocol. Humatron initiates the interaction by sending HTTP POST requests to the designated endpoint of your AI worker. The AI worker must listen for these requests, process the data, and return an appropriate response to Humatron. The detailed request-handling workflow is described here.

To handle these REST requests, the AI worker, built using the provided Python Worker SDK, must be integrated with a web server capable of receiving REST HTTP requests. The web server's role is to forward the incoming requests to the Python client and return the client’s response as an HTTP response. Although you are free to use any web server, we provide a simple reference utility using Flask Server (which we use in the exemple below).

REST request	description
`register`	This request is sent when a new worker instance is hired and just before the onboarding commences.
`unregister`	This request is sent when a worker instance is terminated.
`pause`	This request is sent when a worker instance is temporarily paused.
`resume`	This request is sent when a worker instance is resumed from paused state.
`message`	This request is used to send one or more messages from the outside world to the worker instance.
`interview`	A interview request is used to conduct an interview for the worker template (i.e. specialist).

main classes »

Python Worker SDK provides abstract classes that simplify request processing, along with utilities to integrate these classes into your web server. Python Worker SDK provides two abstract client classes to help to handle required request:

class description

HumatronWorkerApi The class mirrors the REST API, providing a single method process_request to implement. This method accepts a request object as input and returns a response object. You will need to manage the request state and handle all asynchronous logic yourself. This approach is flexible but requires additional effort.

HumatronAsyncWorker The class is designed to handle multiple requests concurrently in asynchronous mode. This class inherits from HumatronWorkerApi, but in the process_payload_part method it implements, the input request payload is divided into separate payload parts. This design allows each part to be processed independently without the need to manage asynchronous execution explicitly. All these response payload parts are processed in asynchronous mode. This class also provides a push_response method to push responses to the client without initiating a request. This response will be sent to the AI worker along with the next portion of responses.

After implementing one of these classes, you can integrate it into your web server by simply providing an HTTP request as input and returning the result. If you use the HumatronWorkerApi implementation, the results will be returned immediately. If you use the HumatronAsyncWorker implementation, the accumulated results will be returned with the next request.

To integrate the Python client with a web server using Flask Server, you can use the specialized utility we provide. Below is an example of how to use this utility for simple integration.

example »

Lets review the minimal implementation of the AI worker.

In our implementation we will forward all the messages that we receive from the outside world to OpenAI LLM and will send the LLM response back to the same channel the request was received on.

spec_impl.py

hide

# AI worker Example.
from uuid import uuid4
from humatron.worker.client import *
from humatron.worker.utils import make_default_response_payload
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# Define the chat prompt template.
_prompt = ChatPromptTemplate.from_messages([('system', 'You are a helpful assistant.'), ('user', '{input}')])

# Worker implementation based on `python-sdk` library.
class HumatronWorkerChatGpt(HumatronAsyncWorker):
    def __init__(self, openai_api_key: str):
        super().__init__()
        # Create the processing chain.
        self._chain = _prompt | ChatOpenAI(openai_api_key=openai_api_key) | StrOutputParser()

    # Implement the `process_payload_part` method.
    def process_payload_part(self, rpp: RequestPayloadPart, _: Storage) -> ResponsePayloadPart:
        # Process different types of request commands.
        match rpp.body:
            case RequestDataMessage(_) as data:
                # To simplify the example, we skip the check for sending a message to oneself, etc.
                match data.message:
                    case RequestMessageEmail(_) as email:
                        resp = self._chain.invoke({'input': email.text})
                        resp_email = ResponseMessageEmail.make(sender=email.to, to=email.sender, subj='Demo', text=resp)
                        return ResponseDataMessage.make(data.instance.id, data.resource_id, resp_email, data.payload_id)
                    case RequestMessageSms(_) as sms:
                        resp = self._chain.invoke({'input': sms.text})
                        resp_sms = ResponseMessageSms.make(sender=sms.receiver, receiver=sms.sender, text=resp)
                        return ResponseDataMessage.make(data.instance.id, data.resource_id, resp_sms, data.payload_id)
                    case RequestMessageSlack(_) as slack:
                        resp = self._chain.invoke({'input': slack.body['text']})
                        resp_slack = ResponseMessageSlack.make(channel=slack.body['channel'], text=resp)
                        return ResponseDataMessage.make(data.instance.id, data.resource_id, resp_slack, data.payload_id)
                    case _:
                        raise ValueError(f'Unexpected request: {data.message}')
            case _:
                # We skip all `interview`, `register`, `unregister`, `pause` and `resume` logic.
                return make_default_response_payload(req_cmd=rpp.req_cmd, req_payload_part=rpp.body)

Comments below:

Lines 3, 4 - import classes from the Python SDK library. This library provides asynchronous request handling as well as a number of utility methods.
Line 17 - initialize OpenAI LLM and create the request processing chain.
Lines 13 - define the HumatronWorkerChatGpt class, which inherits from HumatronAsyncWorker, part of the Python SDK library.
Line 20 - implement the abstract process_payload_part method from the HumatronAsyncWorker ABC class.
Lines 27, 31, 35 - send requests to the LLM, with the text field value extracted from the request body.
Line 26 - handle the message request with channel_type equal to email. We return an email response to the question contained in the email, sent to the sender's address.
Line 30 - handle the message request with channel_type equal to SMS. We return an SMS response to the question contained in the message, sent to the sender's phone number.
Line 34 - handle the message request with channel_type equal to slack. We return a Slack message response to the question contained in the Slack request, sent to the channel from which the request was received.

The final step in completing your AI worker is to wrap its implementation within a web server. Note that we use Flask Server in this example for demonstration purposes - you can use any web server of your choosing:

spec_rest.py

hide

# Web Server Integration Example. 
import os
from dotenv import load_dotenv
from humatron.worker.rest.flask.flask_server import start_flask_server
from demo import HumatronWorkerChatGpt

# Start the REST server.
def start() -> None:
    # Load the environment variables.
    load_dotenv()

    # Get the tokens from the environment.
    req_token, resp_token = os.environ['HUMATRON_REQUEST_TOKEN'], os.environ['HUMATRON_RESPONSE_TOKEN']
    openai_api_key = os.environ['OPENAI_API_KEY']
    host, port, url = os.environ['REST_HOST'], int(os.environ['REST_PORT']), os.environ['REST_URL_WORKER']

    worker = HumatronWorkerChatGpt(openai_api_key=openai_api_key)
    start_flask_server(worker, req_token, resp_token, host, port, url, None, None, lambda: worker.close())

if __name__ == '__main__':
    start()

Comments below:

Line 4 - import classes from the Python SDK library.
Lines 5, 18 - import and create a new instance of HumatronWorkerChatGpt we developed above.

REST channel SDK »

The REST channel is a special communication channel that enables REST interactions between an AI worker instance and the end user. The primary goal of the Python REST Channel SDK is to provide a convenient and efficient way to build custom applications based on the REST Channel API using the Python programming language.

main classes »

Python REST Channel SDK provides three abstract client classes to help to communicate via REST channel.

class	description
`RestChannelClient`	The class mirrors the REST Channel API, providing a single method `post` to implement. This method accepts a request object as input and returns an optional response object. Note that due to the asynchronous nature of the REST channel, this `post` method may return a response to any previous request, the current request, or nothing at all if there is no data to return. Note that you will need to manage the request state and handle all asynchronous logic yourself. This approach is flexible but requires additional effort.
`RestChannelAsyncClient`	The class is designed to operate in asynchronous mode, where you send requests and listen for responses. This class is initialized with a callback parameter, which handles the asynchronous response parts. Requests are sent asynchronously using the `post` method.
`RestChannelSyncClient`	This is the synchronous version of the REST channel client. It features a single post method for sending requests and returning responses. Its API is similar to the `RestChannelClient` described above, with one significant difference: while `RestChannelClient` `post` method returns an optional currently accessible response object, `RestChannelSyncClient` `post` method waits for a response to its request and returns its result.

REST channel SDK example »

Below is a simple example demonstrating how to use the client classes described above.

rest_channel_client_example.py

hide

# REST Channel Base Client Example.
from humatron.channels.rest.client import *

client = RestChannelClient('https://humatron.ai/restchannel', 'your_token')

while True:
    req_txt = input('Request: ')
    req_payload = RequestMessagePayloadPart.make(req_txt, 'sender', 'receiver')
    req = RestChannelRequest.make_message(req_payload)
    resp = client.post(req)

    # A response to the provided payload can be received with either the initial or subsequent response.
    while not resp or resp.payload[0].ref_payload_id != req_payload.payload_id:
        time.sleep(1)
        resp = client.post(RestChannelRequest.make_heartbeat())

    resp_txt = resp.payload[0].text
    print(f'Response: {resp_txt}')

NOTE: that you must send heartbeats to ensure you receive a response to your request.

rest_channel_async_client_example.py

hide

# REST Channel Asynchronous Client Example.
chat = dict()

def callback(resp_part: ResponseMessagePayloadPart) -> None:
    r = chat.get(resp_part.ref_payload_id, 'Unknown request')
    print(f'>>Response `{resp_part.text}` received in request `{r}`.')

with RestChannelAsyncClient('https://humatron.ai/restchannel', 'your_token', callback) as client:
    while True:
        req_txt = input('Request: ')
        payload = RequestMessagePayloadPart.make(req_txt, 'sender', 'receiver')
        chat[payload.payload_id] = payload.text
        req = RestChannelRequest.make_message(payload)
        # The response to the provided payload will be received asynchronously.
        client.post(req)

NOTE: input and output messages may be mixed in the console output.

rest_channel_sync_client_example.py

hide

# REST Channel Synchronous Client Example.
from humatron.channels.rest.client import *

client = RestChannelSyncClient('https://humatron.ai/restchannel', 'your_token')

while True:
    req_txt = input('Request: ')
    req = RestChannelRequest.make_message(RequestMessagePayloadPart.make(req_txt, 'sender', 'receiver'))
    # The response to the provided payload is received in a synchronous manner.
    resp = client.post(req)
    resp_txt = resp.payload[0].text
    print(f'Response: {resp_txt}')

NOTE: this class handles a single request payload at a time, one by one.

Featured Workers

APIs

Worker SDKs

Platform SDKs

Featured Workers

APIs

Worker SDKs

Platform SDKs

overview »

Installation »

worker SDK »

requests »

main classes »

example »

REST channel SDK »

main classes »

REST channel SDK example »