Python Client for Zeebe API

Hello,
So far, I have worked with Java Spring Framework to create clients for Zeebe API. Now I would like to create python clients for Camunda 8.7 (self-managed) and newer versions. I found some documentation in the camunda official documentation but it does not help me till now. Here is the link: Generating a Zeebe-Python Client Stub in Less Than An Hour | Camunda
I found also a GitHub repo that talk about it ( Stéphane Ludwig / zeebe-python-grpc · GitLab ). I took also a look on the official documentation of pyzeebe - the python library to create a camunda client over grpc protocol - but it still not working.

When I run my code, I get always a Runtime exception, from which I don’t know where to start debugging. Could someone has some idea or already build any zeebe client using python or Django ?
I am working on a windows 11 machine.

Code:

import asyncio

from pyzeebe import ZeebeClient, ZeebeWorker, Job, JobController, create_insecure_channel

channel = create_insecure_channel(grpc_address=“host.com:26500”) # Create grpc channel
worker = ZeebeWorker(channel)
client = ZeebeClient(channel)

async def main():
      process_instance_key, process_result = await client.run_process_with_result(
      bpmn_process_id=“pythonTestProcess”,
      timeout=10000)

  return process_instance_key, process_result

if **name** == “\_\_**main_\_**”:
      asyncio.run(main())

Exception:

RuntimeError: Task <Task pending name=‘Task-1’ coro=<main() running at C:\Users…\from-doc.py:13> cb=[_run_until_complete_cb() at C:\Users…\AppData\Local\Programs\Python\Python313\Lib\asyncio\base_events.py:181]> got Future <Task pending name=‘Task-2’ coro=<UnaryUnaryCall._invoke() running at C:\Users…\Desktop\camunda-python-client.venv\Lib\site-packages\grpc\aio_call.py:573>> attached to a different loop
Task was destroyed but it is pending!

Hi @Maurice_Tchangue, welcome to the forums! This appears to be more about how to use asyncio than pyzeebe. Unfortunately I am not a Python developer, but I would try the following:

  • instead of using asyncio.run(main()) what if you just do await main()?
  • the pyzeebe examples use asyncio.get_running_loop() - perhaps try that?
  • the client example just awaits the different calls, you could do some testing with that approach
  • you can also open an issue in the pyzeebe GitHub repository

Sorry I couldn’t be more help, my familiarity with async in Python is very limited. Perhaps someone else can jump in! :crossed_fingers:

@Maurice_Tchangue I struggled with the pyzeebe and the python SDK as well when trying to build a Robot framework test suite. Although I can not help you solve your problem I can over you some hope. According to the roadmap roadmap.camunda.com/tabs/27–roadmap there should/could be an official “Camunda Python Client for Orchestration Cluster API” and even “Python SDK for AI Agent Development”.

This topic was automatically closed 90 days after the last reply. New replies are no longer allowed.