Skip to content

Setup a runtime

Overview

Runtime connects to a Zeebe instance and listens to service tasks handled by the connectors.

Runtime does not discover connectors automatically. You need to let runtime know which connectors you want it to handle by passing connector classes through outbound_connectors argument.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from python_camunda_sdk import CamundaRuntime, InsecureConfig

from log import LogConnector

config = InsecureConfig(
    hostname="127.0.0.1"
)

runtime = CamundaRuntime(
    config=config,
    outbound_connectors=[LogConnector]
)

if __name__ == "__main__":
    runtime.start()

from pydantic import BaseModel, Field
from loguru import logger

from python_camunda_sdk import OutboundConnector


class StatusModel(BaseModel):
    status: str


class LogConnector(OutboundConnector):
    message: str = Field(description="Message to log")

    async def run(self) -> StatusModel:
        logger.info(f"LogConnector: {self.message}")

        return StatusModel(status="ok")

    class ConnectorConfig:
        name = "LogConnector"
        type = "log"

from .log import LogConnector

Start the runtime with

$ python example/runtime.py
2023-08-03 20:44:09.887 | INFO     | python_camunda_sdk.runtime.runtime:main:114 - Loading LogConnector (log)
2023-08-03 20:44:09.887 | INFO     | python_camunda_sdk.runtime.runtime:main:117 - Starting runtime

Runtime has successfully started and loaded LogConnector. It is now ready to handle 'log' type service tasks.

Configuration

In addition to outbound_connectors you need to give runtime credentials to Zeebe instance.

With config object

You can pass configuration object with the details of Zeebe instance you want to connect. The examples of supported configs are below.

from python_camunda_sdk import InsecureConfig

config = InsecureConfig(
    host='127.0.0.1',
    port=26500
)

from python_camunda_sdk import SecureConfig

config = SecureConfig(
    host='127.0.0.1',
    port=26500,
    root_certificates='''
    -----BEGIN CERTIFICATE-----
    xxx
    -----BEGIN CERTIFICATE-----
    '''
    private_key='''
    -----BEGIN RSA PRIVATE KEY-----
    xxx
    -----BEGIN RSA PRIVATE KEY-----
    '''
    certificate_chain=''
)

from python_camunda_sdk import CloudConfig

config = CloudConfig(
    client_id='jYsgv.SryJYQlcpobk-tZZP~2R60xpNY',
    client_secret='55kddTbk~yZBFb2NH5GtebWHkSoK1z.TG7G1Hn-n.mH_f4ihpZAUop1-sryxHnyV',
    cluster_id='7bc802fc-7bf4-4800-b84a-596628d1ed08',
    region='bru-2'
)

With environmental variables

If you don't pass config to the runtime, it will try to collect connection configuration from the environmental variables specified below.

Variable Description
CAMUNDA_CONNECTION_TYPE Connection type
Connection type value Description
INSECURE Insecure connection to a self-hosted Zeebe
SECURE Secure connection to a self-hosted Zeebe
CAMUNDA_CLOUD Conneection to Camunda Cloud
Variable Description
ZEBEE_HOSTNAME Hostname or IP address of Zeebe
ZEBEE_PORT The port Zeebe is listening on
Variable Description
ZEBEE_HOSTNAME Hostname or IP address of Zeebe
ZEBEE_PORT The port Zeebe is listening on
SSL_ROOT_CA SSL certificate
SSL_PRIVATE_KEY SSL private key
SSL_CERTIFICATE_CHAIN SSL certificate chain
Variable Description
CAMUNDA_CLIENT_ID Client id
CAMUNDA_CLIENT_SECRET Client seecret
CAMUNDA_CLUSTER_ID Camunda cluster id
CAMUNDA_REGION Camunda cluster region