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”.