Skip to main content

Zeebe client api

Project description

Coverage Status Test pyzeebe Integration test pyzeebe

GitHub tag (latest by date) PyPI PyPI - Python Version

Documentation Status


pyzeebe is a python grpc client for Zeebe.

Zeebe version support:

Pyzeebe version Tested Zeebe versions
3.x.x 1.0.0
2.x.x 0.23, 0.24, 0.25, 0.26
1.x.x 0.23, 0.24

Getting Started

To install:

pip install pyzeebe

For full documentation please visit:



The ZeebeWorker class uses threading to get and run jobs.

import asyncio

from pyzeebe import ZeebeWorker, Job, create_insecure_channel

channel = create_insecure_channel(hostname="localhost", port=26500) # Create grpc channel
worker = ZeebeWorker(channel) # Create a zeebe worker

async def on_error(exception: Exception, job: Job):
    on_error will be called when the task fails
    await job.set_error_status(f"Failed to handle job {job}. Error: {str(exception)}")

@worker.task(task_type="example", exception_handler=on_error)
def example_task(input: str) -> dict:
    return {"output": f"Hello world, {input}!"}

@worker.task(task_type="example2", exception_handler=on_error)
async def another_example_task(name: str) -> dict: # Tasks can also be async
    return {"output": f"Hello world, {name} from async task!"}

loop = asyncio.get_running_loop()
loop.run_until_complete( # Now every time that a task with type `example` or `example2` is called, the corresponding function will be called

Stop a worker:

await zeebe_worker.stop() # Stops worker after all running jobs have been completed


from pyzeebe import ZeebeClient, create_insecure_channel

# Create a zeebe client
channel = create_insecure_channel(hostname="localhost", port=26500)
zeebe_client = ZeebeClient(channel)

# Run a Zeebe process instance
process_instance_key = await zeebe_client.run_process(bpmn_process_id="My zeebe process", variables={})

# Run a process and receive the result
process_instance_key, process_result = await zeebe_client.run_process_with_result(
    bpmn_process_id="My zeebe process",

# Deploy a BPMN process definition
await zeebe_client.deploy_process("process.bpmn")

# Cancel a running process
await zeebe_client.cancel_process_instance(process_instance_key=12345)

# Publish message
await zeebe_client.publish_message(name="message_name", correlation_key="some_id")


Use the package manager pip to install pyzeebe

pytest tests/unit


Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.

Please make sure to update tests as appropriate.


We use SemVer for versioning. For the versions available, see the tags on this repository.

In order to bump the current version run:

$ bump2version <part>

where part is the part that will be bumped (major/minor/patch/rc).

This will bump the version in all relevant files as well as create a git commit.


We use the MIT license, see for details

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

pyzeebe-3.0.4.tar.gz (21.0 kB view hashes)

Uploaded Source

Built Distribution

pyzeebe-3.0.4-py3-none-any.whl (31.0 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page