multi app token workflow (started from allow to create clients with different dapr-api-token )

dapr/aio/clients/__init__.py

@@ -66,6 +66,7 @@ def __init__(
         ] = None,
         http_timeout_seconds: Optional[int] = None,
         max_grpc_message_length: Optional[int] = None,
+        api_token: Optional[str] = None,
     ):
         """Connects to Dapr Runtime and via gRPC and HTTP.
 
@@ -79,8 +80,10 @@ def __init__(
             http_timeout_seconds (int): specify a timeout for http connections
             max_grpc_messsage_length (int, optional): The maximum grpc send and receive
                 message length in bytes.
+            api_token (str, optional): Dapr API token for authentication. If not provided,
+                falls back to DAPR_API_TOKEN environment variable.
         """
-        super().__init__(address, interceptors, max_grpc_message_length)
+        super().__init__(address, interceptors, max_grpc_message_length, api_token=api_token)
         self.invocation_client = None
 
         invocation_protocol = settings.DAPR_API_METHOD_INVOCATION_PROTOCOL.upper()
@@ -89,7 +92,7 @@ def __init__(
             if http_timeout_seconds is None:
                 http_timeout_seconds = settings.DAPR_HTTP_TIMEOUT_SECONDS
             self.invocation_client = DaprInvocationHttpClient(
-                headers_callback=headers_callback, timeout=http_timeout_seconds
+                headers_callback=headers_callback, timeout=http_timeout_seconds, api_token=api_token
             )
         elif invocation_protocol == 'GRPC':
             pass

dapr/aio/clients/grpc/client.py

@@ -141,6 +141,7 @@ def __init__(
         ] = None,
         max_grpc_message_length: Optional[int] = None,
         retry_policy: Optional[RetryPolicy] = None,
+        api_token: Optional[str] = None,
     ):
         """Connects to Dapr Runtime and initialize gRPC client stub.
 
@@ -152,8 +153,13 @@ def __init__(
                 StreamStreamClientInterceptor, optional): gRPC interceptors.
             max_grpc_message_length (int, optional): The maximum grpc send and receive
                 message length in bytes.
+            api_token (str, optional): Dapr API token for authentication. If not provided,
+                falls back to DAPR_API_TOKEN environment variable.
         """
-        DaprHealth.wait_until_ready()
+        self._api_token = api_token
+        # For health check, use explicit token or fall back to global setting
+        health_token = api_token if api_token is not None else settings.DAPR_API_TOKEN
+        DaprHealth.wait_until_ready(api_token=health_token)
         self.retry_policy = retry_policy or RetryPolicy()
 
         useragent = f'dapr-sdk-python/{__version__}'
@@ -184,10 +190,12 @@ def __init__(
         else:
             interceptors.append(DaprClientTimeoutInterceptorAsync())
 
-        if settings.DAPR_API_TOKEN:
+        # Use explicit token if provided, otherwise fall back to global setting
+        token = self._api_token if self._api_token is not None else settings.DAPR_API_TOKEN
+        if token:
             api_token_interceptor = DaprClientInterceptorAsync(
                 [
-                    ('dapr-api-token', settings.DAPR_API_TOKEN),
+                    ('dapr-api-token', token),
                 ]
             )
             interceptors.append(api_token_interceptor)

dapr/clients/__init__.py

@@ -71,6 +71,7 @@ def __init__(
         http_timeout_seconds: Optional[int] = None,
         max_grpc_message_length: Optional[int] = None,
         retry_policy: Optional[RetryPolicy] = None,
+        api_token: Optional[str] = None,
     ):
         """Connects to Dapr Runtime via gRPC and HTTP.
 
@@ -84,8 +85,10 @@ def __init__(
             http_timeout_seconds (int): specify a timeout for http connections
             max_grpc_message_length (int, optional): The maximum grpc send and receive
                 message length in bytes.
+            api_token (str, optional): Dapr API token for authentication. If not provided,
+                falls back to DAPR_API_TOKEN environment variable.
         """
-        super().__init__(address, interceptors, max_grpc_message_length, retry_policy)
+        super().__init__(address, interceptors, max_grpc_message_length, retry_policy, api_token)
         self.invocation_client = None
 
         invocation_protocol = settings.DAPR_API_METHOD_INVOCATION_PROTOCOL.upper()
@@ -94,7 +97,7 @@ def __init__(
             if http_timeout_seconds is None:
                 http_timeout_seconds = settings.DAPR_HTTP_TIMEOUT_SECONDS
             self.invocation_client = DaprInvocationHttpClient(
-                headers_callback=headers_callback, timeout=http_timeout_seconds
+                headers_callback=headers_callback, timeout=http_timeout_seconds, api_token=api_token
             )
         elif invocation_protocol == 'GRPC':
             pass

dapr/clients/grpc/client.py

@@ -132,6 +132,7 @@ def __init__(
         ] = None,
         max_grpc_message_length: Optional[int] = None,
         retry_policy: Optional[RetryPolicy] = None,
+        api_token: Optional[str] = None,
     ):
         """Connects to Dapr Runtime and initializes gRPC client stub.
 
@@ -144,8 +145,13 @@ def __init__(
             max_grpc_message_length (int, optional): The maximum grpc send and receive
                 message length in bytes.
             retry_policy (RetryPolicy optional): Specifies retry behaviour
+            api_token (str, optional): Dapr API token for authentication. If not provided,
+                falls back to DAPR_API_TOKEN environment variable.
         """
-        DaprHealth.wait_until_ready()
+        self._api_token = api_token
+        # For health check, use explicit token or fall back to global setting
+        health_token = api_token if api_token is not None else settings.DAPR_API_TOKEN
+        DaprHealth.wait_until_ready(api_token=health_token)
         self.retry_policy = retry_policy or RetryPolicy()
 
         useragent = f'dapr-sdk-python/{__version__}'
@@ -184,10 +190,12 @@ def __init__(
 
         self._channel = grpc.intercept_channel(self._channel, DaprClientTimeoutInterceptor())  # type: ignore
 
-        if settings.DAPR_API_TOKEN:
+        # Use explicit token if provided, otherwise fall back to global setting
+        token = self._api_token if self._api_token is not None else settings.DAPR_API_TOKEN
+        if token:
             api_token_interceptor = DaprClientInterceptor(
                 [
-                    ('dapr-api-token', settings.DAPR_API_TOKEN),
+                    ('dapr-api-token', token),
                 ]
             )
             self._channel = grpc.intercept_channel(  # type: ignore

dapr/clients/health.py

@@ -12,22 +12,23 @@
 See the License for the specific language governing permissions and
 limitations under the License.
 """
-import urllib.request
-import urllib.error
 import time
+import urllib.error
+import urllib.request
 
-from dapr.clients.http.conf import DAPR_API_TOKEN_HEADER, USER_AGENT_HEADER, DAPR_USER_AGENT
+from dapr.clients.http.conf import DAPR_API_TOKEN_HEADER, DAPR_USER_AGENT, USER_AGENT_HEADER
 from dapr.clients.http.helpers import get_api_url
 from dapr.conf import settings
 
 
 class DaprHealth:
     @staticmethod
-    def wait_until_ready():
+    def wait_until_ready(api_token: str = None):
         health_url = f'{get_api_url()}/healthz/outbound'
         headers = {USER_AGENT_HEADER: DAPR_USER_AGENT}
-        if settings.DAPR_API_TOKEN is not None:
-            headers[DAPR_API_TOKEN_HEADER] = settings.DAPR_API_TOKEN
+        token = api_token if api_token is not None else settings.DAPR_API_TOKEN
+        if token is not None:
+            headers[DAPR_API_TOKEN_HEADER] = token
         timeout = float(settings.DAPR_HEALTH_TIMEOUT)
 
         start = time.time()

dapr/clients/http/client.py

@@ -43,15 +43,21 @@ def __init__(
         timeout: Optional[int] = 60,
         headers_callback: Optional[Callable[[], Dict[str, str]]] = None,
         retry_policy: Optional[RetryPolicy] = None,
+        api_token: Optional[str] = None,
     ):
         """Invokes Dapr over HTTP.
 
         Args:
             message_serializer (Serializer): Dapr serializer.
             timeout (int, optional): Timeout in seconds, defaults to 60.
             headers_callback (lambda: Dict[str, str]], optional): Generates header for each request.
+            api_token (str, optional): Dapr API token for authentication. If not provided,
+                falls back to DAPR_API_TOKEN environment variable.
         """
-        DaprHealth.wait_until_ready()
+        self._api_token = api_token
+        # For health check, use explicit token or fall back to global setting
+        health_token = api_token if api_token is not None else settings.DAPR_API_TOKEN
+        DaprHealth.wait_until_ready(api_token=health_token)
 
         self._timeout = aiohttp.ClientTimeout(total=timeout)
         self._serializer = message_serializer
@@ -71,8 +77,10 @@ async def send_bytes(
         if not headers_map.get(CONTENT_TYPE_HEADER):
             headers_map[CONTENT_TYPE_HEADER] = DEFAULT_JSON_CONTENT_TYPE
 
-        if settings.DAPR_API_TOKEN is not None:
-            headers_map[DAPR_API_TOKEN_HEADER] = settings.DAPR_API_TOKEN
+        # Use explicit token if provided, otherwise fall back to global setting
+        token = self._api_token if self._api_token is not None else settings.DAPR_API_TOKEN
+        if token is not None:
+            headers_map[DAPR_API_TOKEN_HEADER] = token
 
         if self._headers_callback is not None:
             trace_headers = self._headers_callback()

dapr/clients/http/dapr_actor_http_client.py

@@ -36,6 +36,7 @@ def __init__(
         timeout: int = 60,
         headers_callback: Optional[Callable[[], Dict[str, str]]] = None,
         retry_policy: Optional[RetryPolicy] = None,
+        api_token: Optional[str] = None,
     ):
         """Invokes Dapr Actors over HTTP.
 
@@ -44,8 +45,12 @@ def __init__(
             timeout (int, optional): Timeout in seconds, defaults to 60.
             headers_callback (lambda: Dict[str, str]], optional): Generates header for each request.
             retry_policy (RetryPolicy optional): Specifies retry behaviour
+            api_token (str, optional): Dapr API token for authentication. If not provided,
+                falls back to DAPR_API_TOKEN environment variable.
         """
-        self._client = DaprHttpClient(message_serializer, timeout, headers_callback, retry_policy)
+        self._client = DaprHttpClient(
+            message_serializer, timeout, headers_callback, retry_policy, api_token=api_token
+        )
 
     async def invoke_method(
         self, actor_type: str, actor_id: str, method: str, data: Optional[bytes] = None

dapr/clients/http/dapr_invocation_http_client.py

@@ -39,16 +39,23 @@ def __init__(
         timeout: int = 60,
         headers_callback: Optional[Callable[[], Dict[str, str]]] = None,
         retry_policy: Optional[RetryPolicy] = None,
+        api_token: Optional[str] = None,
     ):
         """Invokes Dapr's API for method invocation over HTTP.
 
         Args:
             timeout (int, optional): Timeout in seconds, defaults to 60.
             headers_callback (lambda: Dict[str, str]], optional): Generates header for each request.
             retry_policy (RetryPolicy optional): Specifies retry behaviour
+            api_token (str, optional): Dapr API token for authentication. If not provided,
+                falls back to DAPR_API_TOKEN environment variable.
         """
         self._client = DaprHttpClient(
-            DefaultJSONSerializer(), timeout, headers_callback, retry_policy=retry_policy
+            DefaultJSONSerializer(),
+            timeout,
+            headers_callback,
+            retry_policy=retry_policy,
+            api_token=api_token,
         )
 
     async def invoke_method_async(

ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_client.py

@@ -52,6 +52,7 @@ def __init__(
         host: Optional[str] = None,
         port: Optional[str] = None,
         logger_options: Optional[LoggerOptions] = None,
+        api_token: Optional[str] = None,
     ):
         address = getAddress(host, port)
 
@@ -61,10 +62,12 @@ def __init__(
             raise DaprInternalError(f'{error}') from error
 
         self._logger = Logger('DaprWorkflowClient', logger_options)
+        self._api_token = api_token
 
         metadata = tuple()
-        if settings.DAPR_API_TOKEN:
-            metadata = ((DAPR_API_TOKEN_HEADER, settings.DAPR_API_TOKEN),)
+        token = self._api_token if self._api_token is not None else settings.DAPR_API_TOKEN
+        if token:
+            metadata = ((DAPR_API_TOKEN_HEADER, token),)
         options = self._logger.get_options()
         self.__obj = client.TaskHubGrpcClient(
             host_address=uri.endpoint,

ext/dapr-ext-workflow/dapr/ext/workflow/workflow_runtime.py

@@ -43,11 +43,14 @@ def __init__(
         host: Optional[str] = None,
         port: Optional[str] = None,
         logger_options: Optional[LoggerOptions] = None,
+        api_token: Optional[str] = None,
     ):
         self._logger = Logger('WorkflowRuntime', logger_options)
+        self._api_token = api_token
         metadata = tuple()
-        if settings.DAPR_API_TOKEN:
-            metadata = ((DAPR_API_TOKEN_HEADER, settings.DAPR_API_TOKEN),)
+        token = self._api_token if self._api_token is not None else settings.DAPR_API_TOKEN
+        if token:
+            metadata = ((DAPR_API_TOKEN_HEADER, token),)
         address = getAddress(host, port)
 
         try: