Python zeebe-grpc and OAuth

Hi there,

I’m trying to test my Python gRPC Library ( with Camunda Cloud but I’m a little bit lost in space with OAuth 2.0.
I would like to integrate oauthlib ( and gRPC. Does anyone have know-how on this topic?

Yo @gizmo84 , “OAuth” is a red herring. See the Node client implementation here:

You just send the env vars over REST to the specified validator service (also in the env vars), and it will give you back a token.

You then put that token in the headers for the gRPC calls. See here:

So if you are googling for how to implement this in Python, you would never touch an OAuth lib. You would be googling for:

  • How to read an env var in Python
  • How to do a REST POST in Python
  • How to set a metadata header in gRPC in Python

That’s pretty much it!


Thanks Josh! You helped me a lot. The two first points are easy. I will see how to manage the third one.

1 Like

@gizmo84 How did it go? Were you able to use zeebe_python_grpc with Camunda Cloud (with OAuth) and would you be able to share your solution?

Hi @datakurre,

I didn’t had time to work on it lately but last time I tried it following code worked for me:

import requests
import grpc
from zeebe_grpc import gateway_pb2, gateway_pb2_grpc


audience = ZEEBE_ADDRESS.split(":")[0]
p = {
    "audience": audience,
    "client_id": ZEEBE_CLIENT_ID,
    "client_secret": ZEEBE_CLIENT_SECRET,

r =, json=p)
access_token = r.json().get("access_token")

class AuthGateway(grpc.AuthMetadataPlugin):
    def __call__(self, context, callback):
        callback((("authorization", f"Bearer {access_token}"),), None)

call_credentials = grpc.metadata_call_credentials(AuthGateway())

creds = grpc.ssl_channel_credentials()

composite_credentials = grpc.composite_channel_credentials(creds, call_credentials)

with grpc.secure_channel(ZEEBE_ADDRESS, credentials=composite_credentials) as channel:
    stub = gateway_pb2_grpc.GatewayStub(channel)

    topology = stub.Topology(gateway_pb2.TopologyRequest())

Let me know if it’s working for you! :slight_smile:

@gizmo84 Thanks!

For the record, I finally ended up with something like

        headers = {"Accept": "application/json",
                   "Content-Type": "application/json"}
        data = {
            "audience": ZEEBE_ADDRESS.rsplit(":", 1)[0],
            "client_id": ZEEBE_CLIENT_ID,
            "client_secret": ZEEBE_CLIENT_SECRET,
            "grant_type": "client_credentials",            
        response =
        token = response.json()["access_token"]  
        auth_creds = grpc.access_token_call_credentials(token)
        ssl_creds = grpc.ssl_channel_credentials()
        channel_creds = grpc.composite_channel_credentials(ssl_creds, auth_creds)
        with grpc.secure_channel(ZEEBE_ADDRESS, channel_creds) as channel:
1 Like