Repository: camunda-community-hub/pyzeebe
Issue category: Feature Request

Issue:
Use environment variables to configure Zeebe
**Is your feature request related to a problem? Please describe.**
The zeebe-client-node library has a "[Zero-Conf constructor](https://github.com/zeebe-io/zeebe-client-node-js#zero-conf)" that allows to use environment variables to set up Zeebe. 

This project could benefit from taking the same approach. 

It already supports specifying `hostname:port` with `ZEEBE_ADDRESS`. 

**Describe the solution you'd like**
To quote from zeebe-client-node docs:

> With no relevant environment variables set, it will default to localhost on the default port with no TLS.
> 
> The following environment variable configurations are possible with the Zero-conf constructor:
> 
> Camunda Cloud:
> 
> ```
> ZEEBE_ADDRESS
> ZEEBE_AUTHORIZATION_SERVER_URL
> ZEEBE_CLIENT_SECRET
> ZEEBE_CLIENT_ID
> ```
> 
> Self-hosted or local broker (no TLS or OAuth):
> 
> ```
> ZEEBE_ADDRESS
> ```
> 
> Self-hosted or local broker with OAuth + TLS:
> 
> ```
> ZEEBE_CLIENT_ID
> ZEEBE_CLIENT_SECRET
> ZEEBE_TOKEN_AUDIENCE
> ZEEBE_AUTHORIZATION_SERVER_URL
> ZEEBE_ADDRESS
> ```
> 
> Basic Auth:
> 
> ```
> ZEEBE_BASIC_AUTH_PASSWORD
> ZEEBE_BASIC_AUTH_USERNAME
> ```


**Describe alternatives you've considered**
- Having a function that's responsible for loading config and pass it to ZeebeWorker/ZeebeClient on instantiation.

E.g.

```py
zeebe_client = ZeebeClient(**get_zeebe_config())
# where `get_zeebe_config` would return a dict
# e.g. {"secure_connection": True, "hostname": "localhost", "port": 26500}
```

---

If this is a direction we'd like to take, I'd be happy to submit a PR for this. 
Use environment variables to configure Zeebe
**Is your feature request related to a problem? Please describe.**
The zeebe-client-node library has a "[Zero-Conf constructor](https://github.com/zeebe-io/zeebe-client-node-js#zero-conf)" that allows to use environment variables to set up Zeebe. 

This project could benefit from taking the same approach. 

It already supports specifying `hostname:port` with `ZEEBE_ADDRESS`. 

**Describe the solution you'd like**
To quote from zeebe-client-node docs:

> With no relevant environment variables set, it will default to localhost on the default port with no TLS.
> 
> The following environment variable configurations are possible with the Zero-conf constructor:
> 
> Camunda Cloud:
> 
> ```
> ZEEBE_ADDRESS
> ZEEBE_AUTHORIZATION_SERVER_URL
> ZEEBE_CLIENT_SECRET
> ZEEBE_CLIENT_ID
> ```
> 
> Self-hosted or local broker (no TLS or OAuth):
> 
> ```
> ZEEBE_ADDRESS
> ```
> 
> Self-hosted or local broker with OAuth + TLS:
> 
> ```
> ZEEBE_CLIENT_ID
> ZEEBE_CLIENT_SECRET
> ZEEBE_TOKEN_AUDIENCE
> ZEEBE_AUTHORIZATION_SERVER_URL
> ZEEBE_ADDRESS
> ```
> 
> Basic Auth:
> 
> ```
> ZEEBE_BASIC_AUTH_PASSWORD
> ZEEBE_BASIC_AUTH_USERNAME
> ```


**Describe alternatives you've considered**
- Having a function that's responsible for loading config and pass it to ZeebeWorker/ZeebeClient on instantiation.

E.g.

```py
zeebe_client = ZeebeClient(**get_zeebe_config())
# where `get_zeebe_config` would return a dict
# e.g. {"secure_connection": True, "hostname": "localhost", "port": 26500}
```

---

If this is a direction we'd like to take, I'd be happy to submit a PR for this. 


Hints:
I definitely think this is something that we want in pyzeebe. It would be great if you could submit a PR.
Should I add this issue to the Pyzeebe 3.0.0 milestone @kbakk?
@JonatanMartens I assume #186 was accidentally closed (because the branch it was made against was deleted)? Should I try to rebase this against master/development?
@kbakk, sorry for the late response. 

It would be great if you could rebase this agains master :)
I definitely think this is something that we want in pyzeebe. It would be great if you could submit a PR.
Should I add this issue to the Pyzeebe 3.0.0 milestone @kbakk?
@JonatanMartens I assume #186 was accidentally closed (because the branch it was made against was deleted)? Should I try to rebase this against master/development?
@kbakk, sorry for the late response. 

It would be great if you could rebase this agains master :)

Patch:
diff --git a/docs/channels.rst b/docs/channels.rst
index 7a5cbbf4..0d45ebcc 100644
--- a/docs/channels.rst
+++ b/docs/channels.rst
@@ -6,4 +6,5 @@ Channels
     :name: channels
 
     Quickstart <channels_quickstart>
+    Configuration <channels_configuration>
     Reference <channels_reference>
diff --git a/docs/channels_configuration.rst b/docs/channels_configuration.rst
new file mode 100644
index 00000000..ddeab0ce
--- /dev/null
+++ b/docs/channels_configuration.rst
@@ -0,0 +1,123 @@
+======================
+Channels Configuration
+======================
+
+This document describes the environment variables and configurations used for establishing different gRPC channel connections to Camunda (Zeebe) instances, either with or without authentication.
+
+Environment Variables
+---------------------
+
+The following environment variables are used to configure channels. The variables are grouped according to their relevance and usage context in each type of channel.
+
+These variables are only considered if a corresponding argument was not passed (Unset) during initialization of a channel.
+
+Common Variables
+----------------
+
+This variables is used across all types of channels:
+
+**ZEEBE_ADDRESS**
+  :Description:  
+    The default address of the Zeebe Gateway.
+  
+  :Usage:        
+    Used in both secure and insecure channel configurations.
+    :func:`pyzeebe.create_insecure_channel`
+    :func:`pyzeebe.create_secure_channel`
+  
+  :Default:      
+    ``"localhost:26500"``
+
+Common OAuth2 Variables
+-----------------------
+
+These variables are specifically for connecting to generic OAuth2 or Camunda Cloud instances.
+
+**CAMUNDA_CLIENT_ID** / **ZEEBE_CLIENT_ID**
+  :Description:  
+    The client ID required for OAuth2 client credential authentication.
+  
+  :Usage:        
+    Required for OAuth2 and Camunda Cloud channels.
+    :func:`pyzeebe.create_oauth2_client_credentials_channel`
+    :func:`pyzeebe.create_camunda_cloud_channel`
+
+**CAMUNDA_CLIENT_SECRET** / **ZEEBE_CLIENT_SECRET**
+  :Description:  
+    The client secret for the OAuth2 client.
+  
+  :Usage:        
+    Required for OAuth2 and Camunda Cloud channels.
+    :func:`pyzeebe.create_oauth2_client_credentials_channel`
+    :func:`pyzeebe.create_camunda_cloud_channel`
+
+OAuth2 Variables (Self-Managed)
+-------------------------------
+
+These variables are primarily used for OAuth2 authentication in self-managed Camunda 8 instances.
+
+**CAMUNDA_OAUTH_URL** / **ZEEBE_AUTHORIZATION_SERVER_URL**
+  :Description:  
+    Specifies the URL of the authorization server issuing access tokens to the client.
+  
+  :Usage:        
+    Required if channel initialization argument was not specified.
+    :func:`pyzeebe.create_oauth2_client_credentials_channel`
+
+**CAMUNDA_TOKEN_AUDIENCE** / **ZEEBE_TOKEN_AUDIENCE**
+  :Description:  
+    Specifies the audience for the OAuth2 token.
+  
+  :Usage:        
+    Used when creating OAuth2 or Camunda Cloud channels.
+    :func:`pyzeebe.create_oauth2_client_credentials_channel`
+  
+  :Default:      
+    ``None`` if not provided.
+
+Camunda Cloud Variables (SaaS)
+------------------------------
+
+These variables are specifically for connecting to Camunda Cloud instances.
+
+**CAMUNDA_OAUTH_URL** / **ZEEBE_AUTHORIZATION_SERVER_URL**
+  :Description:  
+    Specifies the URL of the authorization server issuing access tokens to the client.
+  
+  :Usage:        
+    Used in the OAuth2 and Camunda Cloud channel configurations.
+    :func:`pyzeebe.create_camunda_cloud_channel`
+  
+  :Default:      
+    ``"https://login.cloud.camunda.io/oauth/token"`` if not specified.
+
+**CAMUNDA_CLUSTER_ID**
+  :Description:  
+    The unique identifier for the Camunda Cloud cluster to connect to.
+  
+  :Usage:        
+    Required if channel initialization argument was not specified.
+    :func:`pyzeebe.create_camunda_cloud_channel`
+
+**CAMUNDA_CLUSTER_REGION**
+  :Description:  
+    The region where the Camunda Cloud cluster is hosted.
+  
+  :Usage:        
+    Required for Camunda Cloud channels.
+    :func:`pyzeebe.create_camunda_cloud_channel`
+  
+  :Default:      
+    ``"bru-2"`` if not provided.
+
+**CAMUNDA_TOKEN_AUDIENCE** / **ZEEBE_TOKEN_AUDIENCE**
+  :Description:  
+    Specifies the audience for the OAuth2 token.
+  
+  :Usage:        
+    Used when creating OAuth2 or Camunda Cloud channels.
+    :func:`pyzeebe.create_camunda_cloud_channel`
+  
+  :Default:      
+    ``"zeebe.camunda.io"`` if not provided.
+
diff --git a/docs/channels_quickstart.rst b/docs/channels_quickstart.rst
index e725c85c..95f22e89 100644
--- a/docs/channels_quickstart.rst
+++ b/docs/channels_quickstart.rst
@@ -19,10 +19,6 @@ This Channel can be configured with the parameters `channel_credentials` and `ch
     You can override the default `channel_options` by passing
     e.g. `channel_options = (("grpc.keepalive_time_ms", 60_000),)` - for a keepalive time of 60 seconds.
 
-
-Pyzeebe provides a couple standard ways to achieve this:
-
-
 Insecure
 --------
 
@@ -151,6 +147,12 @@ Example:
 
 This method use the :py:class:`.Oauth2ClientCredentialsMetadataPlugin` under the hood.
 
+Configuration
+-------------
+
+It is possible to omit any arguments to the channel initialization functions and instead provide environment variables.
+See :doc:`Channels Configuration <channels_configuration>` for additional details.
+
 Custom Oauth2 Authorization Flow
 ---------------------------------
 
diff --git a/docs/channels_reference.rst b/docs/channels_reference.rst
index f9d26799..61e14dc3 100644
--- a/docs/channels_reference.rst
+++ b/docs/channels_reference.rst
@@ -26,3 +26,23 @@ Credentials
     :members:
     :special-members:
     :private-members:
+
+
+Utilities (Environment)
+-----------------------
+
+.. autofunction:: pyzeebe.channel.utils.get_zeebe_address
+
+.. autofunction:: pyzeebe.channel.utils.get_camunda_oauth_url
+
+.. autofunction:: pyzeebe.channel.utils.get_camunda_client_id
+
+.. autofunction:: pyzeebe.channel.utils.get_camunda_client_secret
+
+.. autofunction:: pyzeebe.channel.utils.get_camunda_cluster_id
+
+.. autofunction:: pyzeebe.channel.utils.get_camunda_cluster_region
+
+.. autofunction:: pyzeebe.channel.utils.get_camunda_token_audience
+
+.. autofunction:: pyzeebe.channel.utils.get_camunda_address
diff --git a/pyzeebe/channel/insecure_channel.py b/pyzeebe/channel/insecure_channel.py
index 846cff72..39e86360 100644
--- a/pyzeebe/channel/insecure_channel.py
+++ b/pyzeebe/channel/insecure_channel.py
@@ -3,12 +3,12 @@
 import grpc
 
 from pyzeebe.channel.channel_options import get_channel_options
-from pyzeebe.channel.utils import create_address
-from pyzeebe.types import ChannelArgumentType
+from pyzeebe.channel.utils import get_zeebe_address
+from pyzeebe.types import ChannelArgumentType, Unset
 
 
 def create_insecure_channel(
-    grpc_address: str | None = None, channel_options: ChannelArgumentType | None = None
+    grpc_address: str = Unset, channel_options: ChannelArgumentType | None = None
 ) -> grpc.aio.Channel:
     """
     Create an insecure channel
@@ -22,5 +22,7 @@ def create_insecure_channel(
     Returns:
         grpc.aio.Channel: A GRPC Channel connected to the Zeebe gateway.
     """
-    grpc_address = create_address(grpc_address=grpc_address)
+    if grpc_address is Unset:
+        grpc_address = get_zeebe_address()
+
     return grpc.aio.insecure_channel(target=grpc_address, options=get_channel_options(channel_options))
diff --git a/pyzeebe/channel/oauth_channel.py b/pyzeebe/channel/oauth_channel.py
index d54746a4..a4382cb4 100644
--- a/pyzeebe/channel/oauth_channel.py
+++ b/pyzeebe/channel/oauth_channel.py
@@ -5,52 +5,89 @@
 import grpc
 
 from pyzeebe.channel.channel_options import get_channel_options
+from pyzeebe.channel.utils import (
+    get_camunda_address,
+    get_camunda_client_id,
+    get_camunda_client_secret,
+    get_camunda_cluster_id,
+    get_camunda_cluster_region,
+    get_camunda_oauth_url,
+    get_camunda_token_audience,
+    get_zeebe_address,
+)
 from pyzeebe.credentials.oauth import Oauth2ClientCredentialsMetadataPlugin
-from pyzeebe.types import ChannelArgumentType
+from pyzeebe.types import ChannelArgumentType, Unset
 
 
 def create_oauth2_client_credentials_channel(
-    grpc_address: str,
-    client_id: str,
-    client_secret: str,
-    authorization_server: str,
-    scope: str | None = None,
-    audience: str | None = None,
-    channel_credentials: grpc.ChannelCredentials | None = None,
+    grpc_address: str = Unset,
+    client_id: str = Unset,
+    client_secret: str = Unset,
+    authorization_server: str = Unset,
+    scope: str | None = Unset,
+    audience: str | None = Unset,
+    channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials(),
     channel_options: ChannelArgumentType | None = None,
     leeway: int = 60,
     expire_in: int | None = None,
 ) -> grpc.aio.Channel:
     """Create a gRPC channel for connecting to Camunda 8 (Self-Managed) with OAuth2ClientCredentials.
 
-    - https://oauth.net/2/grant-types/client-credentials/
-    - https://datatracker.ietf.org/doc/html/rfc6749#section-11.2.2
+    https://oauth.net/2/grant-types/client-credentials/
+    https://datatracker.ietf.org/doc/html/rfc6749#section-11.2.2
 
     Args:
-        grpc_address (str): Zeebe Gateway Address.
-
-        client_id (str): The client id.
-        client_secret (str): The client secret.
-        authorization_server (str): The authorization server issuing access tokens
+        grpc_address (str, optional): Zeebe Gateway Address.
+            Defaults to value from ZEEBE_ADDRESS environment variable
+            or "{CAMUNDA_CLUSTER_ID}.{CAMUNDA_CLUSTER_REGION}.zeebe.camunda.io:443"
+            or "localhost:26500".
+        client_id (str, optional): The client id.
+            Defaults to value from CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID environment variable
+        client_secret (str, optional): The client secret.
+            Defaults to value from CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET environment variable
+        authorization_server (str, optional): The authorization server issuing access tokens
             to the client after successfully authenticating the client.
-        scope (Optional[str]): The scope of the access request. Defaults to None.
-        audience (Optional[str]): The audience for authentication. Defaults to None.
+            Defaults to value from CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL environment variable
+        scope (str | None, optional): The scope of the access request.
+        audience (str | None, optional): The audience for authentication.
+            Defaults to value from CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE environment variable
 
         channel_credentials (grpc.ChannelCredentials): The gRPC channel credentials.
             Defaults to grpc.ssl_channel_credentials().
-        channel_options (Optional[ChannelArgumentType]): Additional options for the gRPC channel.
+        channel_options (ChannelArgumentType | None): Additional options for the gRPC channel.
             Defaults to None.
             See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
 
         leeway (int): The number of seconds to consider the token as expired before the actual expiration time.
             Defaults to 60.
-        expire_in (Optional[int]): The number of seconds the token is valid for. Defaults to None.
+        expire_in (int | None): The number of seconds the token is valid for. Defaults to None.
             Should only be used if the token does not contain an "expires_in" attribute.
 
     Returns:
         grpc.aio.Channel: A gRPC channel connected to the Zeebe Gateway.
+
+    Raises:
+        InvalidOAuthCredentialsError: One of the provided camunda credentials is not correct
     """
 
+    if grpc_address is Unset:
+        grpc_address = get_zeebe_address()
+
+    if client_id is Unset:
+        client_id = get_camunda_client_id()
+
+    if client_secret is Unset:
+        client_secret = get_camunda_client_secret()
+
+    if authorization_server is Unset:
+        authorization_server = get_camunda_oauth_url()
+
+    if scope is Unset:
+        scope = None
+
+    if audience is Unset:
+        audience = get_camunda_token_audience()
+
     oauth2_client_credentials = Oauth2ClientCredentialsMetadataPlugin(
         client_id=client_id,
         client_secret=client_secret,
@@ -67,21 +104,23 @@ def create_oauth2_client_credentials_channel(
     )
 
     channel: grpc.aio.Channel = grpc.aio.secure_channel(
-        target=grpc_address, credentials=composite_credentials, options=get_channel_options(channel_options)
+        target=grpc_address,
+        credentials=composite_credentials,
+        options=get_channel_options(channel_options),
     )
 
     return channel
 
 
 def create_camunda_cloud_channel(
-    client_id: str,
-    client_secret: str,
-    cluster_id: str,
-    region: str = "bru-2",
-    authorization_server: str = "https://login.cloud.camunda.io/oauth/token",
-    scope: str | None = None,
-    audience: str | None = "zeebe.camunda.io",
-    channel_credentials: grpc.ChannelCredentials | None = None,
+    client_id: str = Unset,
+    client_secret: str = Unset,
+    cluster_id: str = Unset,
+    region: str = Unset,
+    authorization_server: str = Unset,
+    scope: str | None = Unset,
+    audience: str | None = Unset,
+    channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials(),
     channel_options: ChannelArgumentType | None = None,
     leeway: int = 60,
     expire_in: int | None = None,
@@ -89,32 +128,60 @@ def create_camunda_cloud_channel(
     """Create a gRPC channel for connecting to Camunda 8 Cloud (SaaS).
 
     Args:
-        client_id (str): The client id.
-        client_secret (str): The client secret.
-        cluster_id (str): The ID of the cluster to connect to.
-        region (Optional[str]): The region of the cluster. Defaults to "bru-2".
-        authorization_server (Optional[str]): The authorization server issuing access tokens
+        client_id (str, optional): The client id.
+            Defaults to value from CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID environment variable.
+        client_secret (str, optional): The client secret.
+            Defaults to value from CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET environment variable.
+        cluster_id (str, optional): The ID of the cluster to connect to.
+            Defaults to value from CAMUNDA_CLUSTER_ID environment variable.
+        region (str, optional): The region of the cluster.
+            Defaults to value from CAMUNDA_CLUSTER_REGION environment variable or 'bru-2'.
+        authorization_server (str, optional): The authorization server issuing access tokens
             to the client after successfully authenticating the client.
-            Defaults to "https://login.cloud.camunda.io/oauth/token".
-        scope (Optional[str]): The scope of the access request. Can be set to CAMUNDA_CLUSTER_ID. Defaults to None.
-        audience (Optional[str]): The audience for authentication. Defaults to "zeebe.camunda.io".
+            Defaults to value from CAMUNDA_OAUTH_URL
+            or ZEEBE_AUTHORIZATION_SERVER_URL environment variable
+            or "https://login.cloud.camunda.io/oauth/token".
+        scope (str | None, optional): The scope of the access request.
+        audience (str | None, optional): The audience for authentication.
+            Defaults to value from CAMUNDA_TOKEN_AUDIENCE
+            or ZEEBE_TOKEN_AUDIENCE environment variable
+            or "zeebe.camunda.io".
 
         channel_credentials (grpc.ChannelCredentials): The gRPC channel credentials.
             Defaults to grpc.ssl_channel_credentials().
-        channel_options (Optional[ChannelArgumentType]): Additional options for the gRPC channel.
+        channel_options (ChannelArgumentType | None): Additional options for the gRPC channel.
             Defaults to None.
             See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
 
         leeway (int): The number of seconds to consider the token as expired before the actual expiration time.
             Defaults to 60.
-        expire_in (Optional[int]): The number of seconds the token is valid for. Defaults to None.
+        expire_in (int | None): The number of seconds the token is valid for. Defaults to None.
             Should only be used if the token does not contain an "expires_in" attribute.
 
     Returns:
         grpc.aio.Channel: The gRPC channel for connecting to Camunda Cloud.
     """
 
-    grpc_address = f"{cluster_id}.{region}.zeebe.camunda.io:443"
+    if client_id is Unset:
+        client_id = get_camunda_client_id()
+
+    if client_secret is Unset:
+        client_secret = get_camunda_client_secret()
+
+    if cluster_id is Unset:
+        cluster_id = get_camunda_cluster_id()
+
+    if region is Unset:
+        region = get_camunda_cluster_region("bru-2")
+
+    if authorization_server is Unset:
+        authorization_server = get_camunda_oauth_url("https://login.cloud.camunda.io/oauth/token")
+
+    if scope is Unset:
+        scope = None
+
+    if audience is Unset:
+        audience = get_camunda_token_audience("zeebe.camunda.io")
 
     oauth2_client_credentials = Oauth2ClientCredentialsMetadataPlugin(
         client_id=client_id,
@@ -141,6 +208,8 @@ def create_camunda_cloud_channel(
         channel_credentials or grpc.ssl_channel_credentials(), call_credentials
     )
 
+    grpc_address = get_camunda_address(cluster_id=cluster_id, cluster_region=region)
+
     channel: grpc.aio.Channel = grpc.aio.secure_channel(
         target=grpc_address, credentials=composite_credentials, options=get_channel_options(channel_options)
     )
diff --git a/pyzeebe/channel/secure_channel.py b/pyzeebe/channel/secure_channel.py
index e28b4f91..5ffb53da 100644
--- a/pyzeebe/channel/secure_channel.py
+++ b/pyzeebe/channel/secure_channel.py
@@ -3,12 +3,12 @@
 import grpc
 
 from pyzeebe.channel.channel_options import get_channel_options
-from pyzeebe.channel.utils import create_address
-from pyzeebe.types import ChannelArgumentType
+from pyzeebe.channel.utils import get_zeebe_address
+from pyzeebe.types import ChannelArgumentType, Unset
 
 
 def create_secure_channel(
-    grpc_address: str | None = None,
+    grpc_address: str = Unset,
     channel_options: ChannelArgumentType | None = None,
     channel_credentials: grpc.ChannelCredentials | None = None,
 ) -> grpc.aio.Channel:
@@ -26,7 +26,10 @@ def create_secure_channel(
     Returns:
         grpc.aio.Channel: A GRPC Channel connected to the Zeebe gateway.
     """
-    grpc_address = create_address(grpc_address=grpc_address)
+
+    if grpc_address is Unset:
+        grpc_address = get_zeebe_address()
+
     credentials = channel_credentials or grpc.ssl_channel_credentials()
     return grpc.aio.secure_channel(
         target=grpc_address, credentials=credentials, options=get_channel_options(channel_options)
diff --git a/pyzeebe/channel/utils.py b/pyzeebe/channel/utils.py
index 7b81909d..57fbe015 100644
--- a/pyzeebe/channel/utils.py
+++ b/pyzeebe/channel/utils.py
@@ -5,9 +5,151 @@
 DEFAULT_ZEEBE_ADDRESS = "localhost:26500"
 
 
-def create_address(
-    grpc_address: str | None = None,
-) -> str:
-    if grpc_address:
-        return grpc_address
-    return os.getenv("ZEEBE_ADDRESS", DEFAULT_ZEEBE_ADDRESS)
+def get_zeebe_address(default: str | None = None) -> str:
+    """
+    Get the Zeebe Gateway Address.
+
+    Args:
+        default (str, optional): Default value to be used if no other value was discovered.
+
+    Returns:
+        str: ZEEBE_ADDRESS environment variable or provided default or "localhost:26500"
+    """
+    return os.getenv("ZEEBE_ADDRESS") or default or DEFAULT_ZEEBE_ADDRESS
+
+
+def get_camunda_oauth_url(default: str | None = None) -> str:
+    """
+    Get the Camunda OAuth URL or Zeebe Authorization Server URL.
+
+    Args:
+        default (str, optional): Default value to be used if no other value was discovered.
+
+    Returns:
+        str: CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL environment variable or provided default
+
+    Raises:
+        EnvironmentError: If neither CAMUNDA_OAUTH_URL nor ZEEBE_AUTHORIZATION_SERVER_URL is provided.
+    """
+    r = os.getenv("CAMUNDA_OAUTH_URL") or os.getenv("ZEEBE_AUTHORIZATION_SERVER_URL") or default
+
+    if r is None:
+        raise EnvironmentError("No CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL provided!")
+
+    return r
+
+
+def get_camunda_client_id() -> str:
+    """
+    Get the Camunda Client ID.
+
+    Returns:
+        str: CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID environment variable
+
+    Raises:
+        EnvironmentError: If neither CAMUNDA_CLIENT_ID nor ZEEBE_CLIENT_ID is provided.
+    """
+    r = os.getenv("CAMUNDA_CLIENT_ID") or os.getenv("ZEEBE_CLIENT_ID")
+
+    if r is None:
+        raise EnvironmentError("No CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID provided!")
+
+    return r
+
+
+def get_camunda_client_secret() -> str:
+    """
+    Get the Camunda Client Secret.
+
+    Returns:
+        str: CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET environment variable
+
+    Raises:
+        EnvironmentError: If neither CAMUNDA_CLIENT_SECRET nor ZEEBE_CLIENT_SECRET is provided.
+    """
+    r = os.getenv("CAMUNDA_CLIENT_SECRET") or os.getenv("ZEEBE_CLIENT_SECRET")
+
+    if r is None:
+        raise EnvironmentError("No CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET provided!")
+
+    return r
+
+
+def get_camunda_cluster_id() -> str:
+    """
+    Get the Camunda Cluster ID.
+
+    Returns:
+        str: CAMUNDA_CLUSTER_ID environment variable
+
+    Raises:
+        EnvironmentError: If CAMUNDA_CLUSTER_ID is not provided.
+    """
+    r = os.getenv("CAMUNDA_CLUSTER_ID")
+
+    if r is None:
+        raise EnvironmentError("No CAMUNDA_CLUSTER_ID provided!")
+
+    return r
+
+
+def get_camunda_cluster_region(default: str | None = None) -> str:
+    """
+    Get the Camunda Cluster Region.
+
+    Args:
+        default (str, optional): Default value to be used if no other value was discovered.
+
+    Returns:
+        str: CAMUNDA_CLUSTER_REGION environment variable or provided default
+
+    Raises:
+        EnvironmentError: If CAMUNDA_CLUSTER_REGION is not provided.
+    """
+    r = os.getenv("CAMUNDA_CLUSTER_REGION") or default
+
+    if r is None:
+        raise EnvironmentError("No CAMUNDA_CLUSTER_REGION provided!")
+
+    return r
+
+
+def get_camunda_token_audience(default: str | None = None) -> str:
+    """
+    Get the Camunda Token Audience.
+
+    Args:
+        default (str, optional): Default value to be used if no other value was discovered.
+
+    Returns:
+        str: CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE environment variable or provided default
+
+    Raises:
+        EnvironmentError: If neither CAMUNDA_TOKEN_AUDIENCE nor ZEEBE_TOKEN_AUDIENCE is provided.
+    """
+    r = os.getenv("CAMUNDA_TOKEN_AUDIENCE") or os.getenv("ZEEBE_TOKEN_AUDIENCE") or default
+
+    if r is None:
+        raise EnvironmentError("No CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE provided!")
+
+    return r
+
+
+def get_camunda_address(cluster_id: str | None = None, cluster_region: str | None = None) -> str:
+    """
+    Get the Camunda Cloud gRPC server address.
+
+    Args:
+        cluster_id (str, optional): The Camunda cluster ID provided as parameter.
+        cluster_region (str, optional): The Camunda cluster region provided as parameter.
+
+    Returns:
+        str: The Camunda Cloud gRPC server address.
+
+    Raises:
+        EnvironmentError: If either cluster_id or cluster_region is not provided.
+    """
+    if (cluster_id is None) or (cluster_region is None):
+        raise EnvironmentError("The cluster_id and cluster_region must be provided!")
+
+    return f"{cluster_id}.{cluster_region}.zeebe.camunda.io:443"
diff --git a/pyzeebe/types.py b/pyzeebe/types.py
index 6191be46..696fab30 100644
--- a/pyzeebe/types.py
+++ b/pyzeebe/types.py
@@ -5,5 +5,6 @@
 
 Headers: TypeAlias = Mapping[str, Any]
 Variables: TypeAlias = Mapping[str, Any]
+Unset = str("UNSET")
 
 ChannelArgumentType: TypeAlias = Sequence[tuple[str, Any]]


Test patch:
diff --git a/tests/unit/channel/insecure_channel_test.py b/tests/unit/channel/insecure_channel_test.py
index e76fe6ae..35a07d83 100644
--- a/tests/unit/channel/insecure_channel_test.py
+++ b/tests/unit/channel/insecure_channel_test.py
@@ -1,3 +1,4 @@
+import os
 from unittest.mock import Mock, patch
 
 import grpc
@@ -5,7 +6,7 @@
 
 from pyzeebe import create_insecure_channel
 from pyzeebe.channel.channel_options import get_channel_options
-from pyzeebe.channel.utils import create_address
+from pyzeebe.channel.utils import get_zeebe_address
 
 
 class TestCreateInsecureChannel:
@@ -29,4 +30,14 @@ def test_uses_default_address(self, insecure_channel_mock: Mock):
         create_insecure_channel()
 
         insecure_channel_call = insecure_channel_mock.mock_calls[0]
-        assert insecure_channel_call.kwargs["target"] == create_address()
+        assert insecure_channel_call.kwargs["target"] == get_zeebe_address()
+
+    @patch.dict(
+        os.environ,
+        {"ZEEBE_ADDRESS": "ZEEBE_ADDRESS"},
+    )
+    def test_uses_zeebe_address_environment_variable(self, insecure_channel_mock: Mock):
+        create_insecure_channel()
+
+        insecure_channel_call = insecure_channel_mock.mock_calls[0]
+        assert insecure_channel_call.kwargs["target"] == "ZEEBE_ADDRESS"
diff --git a/tests/unit/channel/oauth_channel_test.py b/tests/unit/channel/oauth_channel_test.py
index 19e122c3..cc6a6b9d 100644
--- a/tests/unit/channel/oauth_channel_test.py
+++ b/tests/unit/channel/oauth_channel_test.py
@@ -1,3 +1,4 @@
+import os
 from unittest import mock
 
 import grpc
@@ -42,6 +43,24 @@ def test_create_oauth2_client_credentials_channel(
     assert isinstance(channel, grpc.aio.Channel)
 
 
+@mock.patch.dict(
+    os.environ,
+    {
+        "ZEEBE_ADDRESS": "ZEEBE_ADDRESS",
+        "CAMUNDA_CLIENT_ID": "CAMUNDA_CLIENT_ID",
+        "CAMUNDA_CLIENT_SECRET": "CAMUNDA_CLIENT_SECRET",
+        "CAMUNDA_OAUTH_URL": "CAMUNDA_OAUTH_URL",
+        "CAMUNDA_TOKEN_AUDIENCE": "CAMUNDA_TOKEN_AUDIENCE",
+    },
+)
+def test_create_oauth2_client_credentials_channel_using_environment_variables(
+    mock_oauth2metadataplugin,
+):
+    channel = create_oauth2_client_credentials_channel()
+
+    assert isinstance(channel, grpc.aio.Channel)
+
+
 def test_create_camunda_cloud_channel(
     mock_oauth2metadataplugin,
 ):
@@ -68,3 +87,20 @@ def test_create_camunda_cloud_channel(
     )
 
     assert isinstance(channel, grpc.aio.Channel)
+
+
+@mock.patch.dict(
+    os.environ,
+    {
+        "CAMUNDA_CLUSTER_ID": "CAMUNDA_CLUSTER_ID",
+        "CAMUNDA_CLUSTER_REGION": "CAMUNDA_CLUSTER_REGION",
+        "CAMUNDA_CLIENT_ID": "CAMUNDA_CLIENT_ID",
+        "CAMUNDA_CLIENT_SECRET": "CAMUNDA_CLIENT_SECRET",
+    },
+)
+def test_create_camunda_cloud_channel_using_environment_variables(
+    mock_oauth2metadataplugin,
+):
+    channel = create_camunda_cloud_channel()
+
+    assert isinstance(channel, grpc.aio.Channel)
diff --git a/tests/unit/channel/secure_channel_test.py b/tests/unit/channel/secure_channel_test.py
index 181f5f80..d9bf419c 100644
--- a/tests/unit/channel/secure_channel_test.py
+++ b/tests/unit/channel/secure_channel_test.py
@@ -1,3 +1,4 @@
+import os
 from unittest.mock import Mock, patch
 
 import grpc
@@ -5,7 +6,7 @@
 
 from pyzeebe import create_secure_channel
 from pyzeebe.channel.channel_options import get_channel_options
-from pyzeebe.channel.utils import create_address
+from pyzeebe.channel.utils import get_zeebe_address
 
 
 class TestCreateSecureChannel:
@@ -35,4 +36,14 @@ def test_uses_default_address(self, secure_channel_mock: Mock):
         create_secure_channel()
 
         secure_channel_call = secure_channel_mock.mock_calls[0]
-        assert secure_channel_call.kwargs["target"] == create_address()
+        assert secure_channel_call.kwargs["target"] == get_zeebe_address()
+
+    @patch.dict(
+        os.environ,
+        {"ZEEBE_ADDRESS": "ZEEBE_ADDRESS"},
+    )
+    def test_uses_zeebe_address_environment_variable(self, secure_channel_mock: Mock):
+        create_secure_channel()
+
+        secure_channel_call = secure_channel_mock.mock_calls[0]
+        assert secure_channel_call.kwargs["target"] == "ZEEBE_ADDRESS"
diff --git a/tests/unit/channel/utils_test.py b/tests/unit/channel/utils_test.py
index f4d0a9ef..0a6537d3 100644
--- a/tests/unit/channel/utils_test.py
+++ b/tests/unit/channel/utils_test.py
@@ -1,25 +1,190 @@
 import os
+from unittest.mock import patch
 from uuid import uuid4
 
-from pyzeebe.channel.utils import DEFAULT_ZEEBE_ADDRESS, create_address
+import pytest
 
+from pyzeebe.channel.utils import (
+    DEFAULT_ZEEBE_ADDRESS,
+    get_camunda_address,
+    get_camunda_client_id,
+    get_camunda_client_secret,
+    get_camunda_cluster_id,
+    get_camunda_cluster_region,
+    get_camunda_oauth_url,
+    get_camunda_token_audience,
+    get_zeebe_address,
+)
 
-class TestCreateAddress:
+
+class TestGetZeebeAddress:
 
     def test_returns_passed_address(self):
         address = str(uuid4())
 
-        assert address == create_address(address)
+        assert address == get_zeebe_address(address)
 
     def test_returns_default_address(self):
-        address = create_address()
+        address = get_zeebe_address()
 
         assert address == DEFAULT_ZEEBE_ADDRESS
 
+    @patch.dict(
+        os.environ,
+        {"ZEEBE_ADDRESS": "ZEEBE_ADDRESS"},
+    )
     def test_returns_env_var_if_provided(self):
-        zeebe_address = str(uuid4())
-        os.environ["ZEEBE_ADDRESS"] = zeebe_address
+        address = get_zeebe_address("zeebe_address")
+
+        assert address == "ZEEBE_ADDRESS"
+
+
+class TestGetCamundaOauthUrl:
+    @patch.dict(
+        os.environ,
+        {"CAMUNDA_OAUTH_URL": "CAMUNDA_OAUTH_URL", "ZEEBE_AUTHORIZATION_SERVER_URL": "ZEEBE_AUTHORIZATION_SERVER_URL"},
+    )
+    def test_camunda_oauth_url_has_highest_priority(self):
+        result = get_camunda_oauth_url("oauth_url")
+
+        assert result == "CAMUNDA_OAUTH_URL"
+
+    @patch.dict(os.environ, {"ZEEBE_AUTHORIZATION_SERVER_URL": "ZEEBE_AUTHORIZATION_SERVER_URL"})
+    def test_zeebe_authorization_server_url_has_second_highest_priority(self):
+        result = get_camunda_oauth_url("oauth_url")
+
+        assert result == "ZEEBE_AUTHORIZATION_SERVER_URL"
+
+    def test_param_has_lowest_priority(self):
+        result = get_camunda_oauth_url("oauth_url")
+
+        assert result == "oauth_url"
+
+    @patch.dict(os.environ, {})
+    def test_none_has_fourth_highest_priority(self):
+        with pytest.raises(EnvironmentError):
+            get_camunda_oauth_url()
+
+
+class TestGetCamundaClientId:
+    @patch.dict(os.environ, {"CAMUNDA_CLIENT_ID": "CAMUNDA_CLIENT_ID", "ZEEBE_CLIENT_ID": "ZEEBE_CLIENT_ID"})
+    def test_is_calculated_from_camunda_environment_variable_as_highest_priority(self):
+        result = get_camunda_client_id()
+
+        assert result == "CAMUNDA_CLIENT_ID"
+
+    @patch.dict(os.environ, {"ZEEBE_CLIENT_ID": "ZEEBE_CLIENT_ID"})
+    def test_is_calculated_from_zeebe_environment_variable_as_second_priority(self):
+        result = get_camunda_client_id()
+
+        assert result == "ZEEBE_CLIENT_ID"
+
+    @patch.dict(os.environ, {})
+    def test_throw_exception_if_not_configured(self):
+        with pytest.raises(EnvironmentError):
+            get_camunda_client_id()
+
+
+class TestGetCamundaClientSecret:
+    @patch.dict(
+        os.environ, {"CAMUNDA_CLIENT_SECRET": "CAMUNDA_CLIENT_SECRET", "ZEEBE_CLIENT_SECRET": "ZEEBE_CLIENT_SECRET"}
+    )
+    def test_is_calculated_from_camunda_environment_variable_as_highest_priority(self):
+        result = get_camunda_client_secret()
+
+        assert result == "CAMUNDA_CLIENT_SECRET"
+
+    @patch.dict(os.environ, {"ZEEBE_CLIENT_SECRET": "ZEEBE_CLIENT_SECRET"})
+    def test_is_calculated_from_zeebe_environment_variable_as_second_priority(self):
+        result = get_camunda_client_secret()
+
+        assert result == "ZEEBE_CLIENT_SECRET"
+
+    @patch.dict(os.environ, {})
+    def test_throw_exception_if_not_configured(self):
+        with pytest.raises(EnvironmentError):
+            get_camunda_client_secret()
+
+
+class TestGetCamundaCloudClusterId:
+    @patch.dict(os.environ, {"CAMUNDA_CLUSTER_ID": "CAMUNDA_CLUSTER_ID"})
+    def test_is_calculated_from_camunda_environment_variable_as_highest_priority(self):
+        result = get_camunda_cluster_id()
+
+        assert result == "CAMUNDA_CLUSTER_ID"
+
+    def test_environment_error(self):
+        with pytest.raises(EnvironmentError):
+            get_camunda_cluster_id()
+
+
+class TestGetCamundaCloudClusterRegion:
+    @patch.dict(os.environ, {"CAMUNDA_CLUSTER_REGION": "CAMUNDA_CLUSTER_REGION"})
+    def test_is_calculated_from_camunda_environment_variable_as_highest_priority(self):
+        result = get_camunda_cluster_region("cluster_region_param")
+
+        assert result == "CAMUNDA_CLUSTER_REGION"
+
+    def test_is_calculated_from_default_value_parameter_as_second_priority(self):
+        result = get_camunda_cluster_region("cluster_region_param")
+
+        assert result == "cluster_region_param"
+
+    @patch.dict(os.environ, {})
+    def test_raises_environment_error_if_no_default_value_as_fallback_provided(self):
+        with pytest.raises(EnvironmentError):
+            get_camunda_cluster_region()
+
+
+class TestGetCamundaTokenAudience:
+    @patch.dict(
+        os.environ,
+        {
+            "CAMUNDA_TOKEN_AUDIENCE": "CAMUNDA_TOKEN_AUDIENCE",
+            "ZEEBE_TOKEN_AUDIENCE": "ZEEBE_TOKEN_AUDIENCE",
+        },
+    )
+    def test_is_calculated_from_camunda_token_audience_as_highest_priority(self):
+        result = get_camunda_token_audience("token_audience_param")
+
+        assert result == "CAMUNDA_TOKEN_AUDIENCE"
+
+    @patch.dict(
+        os.environ,
+        {
+            "ZEEBE_TOKEN_AUDIENCE": "ZEEBE_TOKEN_AUDIENCE",
+        },
+    )
+    def test_is_calculated_from_zeebe_token_audience_as_second_highest_priority(self):
+        result = get_camunda_token_audience("token_audience_param")
+
+        assert result == "ZEEBE_TOKEN_AUDIENCE"
+
+    def test_is_calculated_from_camunda_token_audience_as_third_highest_priority(self):
+        result = get_camunda_token_audience("token_audience_param")
+
+        assert result == "token_audience_param"
+
+    @patch.dict(os.environ, {})
+    def test_raises_environment_error_as_fourth_highest_priority(self):
+        with pytest.raises(EnvironmentError):
+            get_camunda_token_audience()
+
+
+class TestGetCamundaAddress:
+    def test_is_calculated_from_parameters_as_highest_priority(self):
+        result = get_camunda_address("cluster_id_param", "camunda_region_param")
+
+        assert result == f"cluster_id_param.camunda_region_param.zeebe.camunda.io:443"
+
+    def test_raises_error_if_cluster_id_is_none(self):
+        with pytest.raises(EnvironmentError):
+            get_camunda_address(None, "camunda_region_param")
 
-        address = create_address()
+    def test_raises_error_if_cluster_region_is_none(self):
+        with pytest.raises(EnvironmentError):
+            get_camunda_address("cluster_id_param", None)
 
-        assert address == zeebe_address
+    def test_raises_error_if_all_args_are_none(self):
+        with pytest.raises(EnvironmentError):
+            get_camunda_address(None, None)