Add plane-scoped auth tokens

Author: durable-workflow-ops

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Plane-scoped SDK bearer tokens: `Client(..., control_token=..., worker_token=...)`
+  and the sync wrapper now support least-privilege server deployments where
+  operator/admin credentials are separate from worker credentials. The existing
+  `token=` argument remains the shared fallback.
 - `Worker.run_until(workflow_id=..., timeout=...)` for examples, smoke tests,
   and single-workflow scripts that need to run a worker until one workflow
   reaches a terminal state.

README.md

@@ -63,6 +63,30 @@ multi-activity order workflow against a local server with Docker Compose.
 - **Codec envelopes**: Avro payloads by default, with JSON decode compatibility for existing history
 - **Metrics hooks**: Pluggable counters and histograms, with an optional Prometheus adapter
 
+## Authentication
+
+For local servers that use one shared bearer token, pass `token=`:
+
+```python
+client = Client("http://server:8080", token="shared-token", namespace="default")
+```
+
+For production servers with role-scoped tokens, pass separate credentials for
+control-plane calls and worker-plane polling:
+
+```python
+client = Client(
+    "https://workflow.example.internal",
+    control_token="operator-token",
+    worker_token="worker-token",
+    namespace="orders",
+)
+```
+
+Create one client per namespace when your deployment issues namespace-scoped
+tokens. The SDK sends the configured token as `Authorization: Bearer ...` and
+the namespace as `X-Namespace` on every request.
+
 ## Metrics
 
 Pass a recorder to `Client(metrics=...)` or `Worker(metrics=...)` to collect request, poll, and task metrics. The SDK ships a no-op default, an `InMemoryMetrics` recorder for tests or custom exporter loops, and `PrometheusMetrics` for deployments that install the optional extra:

src/durable_workflow/client.py

@@ -271,13 +271,17 @@ def __init__(
         base_url: str,
         *,
         token: str | None = None,
+        control_token: str | None = None,
+        worker_token: str | None = None,
         namespace: str = "default",
         timeout: float = 60.0,
         retry_policy: RetryPolicy | None = None,
         metrics: MetricsRecorder | None = None,
     ) -> None:
         self.base_url = base_url.rstrip("/")
         self.token = token
+        self.control_token = control_token
+        self.worker_token = worker_token
         self.namespace = namespace
         self.retry_policy = retry_policy or RetryPolicy()
         self.metrics = metrics or NOOP_METRICS
@@ -294,15 +298,21 @@ async def __aexit__(self, *exc: Any) -> None:
 
     def _headers(self, *, worker: bool = False) -> dict[str, str]:
         h: dict[str, str] = {"Content-Type": "application/json", "Accept": "application/json"}
-        if self.token:
-            h["Authorization"] = f"Bearer {self.token}"
+        token = self._auth_token(worker=worker)
+        if token:
+            h["Authorization"] = f"Bearer {token}"
         h["X-Namespace"] = self.namespace
         if worker:
             h["X-Durable-Workflow-Protocol-Version"] = PROTOCOL_VERSION
         else:
             h["X-Durable-Workflow-Control-Plane-Version"] = CONTROL_PLANE_VERSION
         return h
 
+    def _auth_token(self, *, worker: bool = False) -> str | None:
+        if worker:
+            return self.worker_token or self.token or self.control_token
+        return self.control_token or self.token or self.worker_token
+
     async def _request(
         self,
         method: str,

src/durable_workflow/sync.py

@@ -139,6 +139,8 @@ def __init__(
         base_url: str,
         *,
         token: str | None = None,
+        control_token: str | None = None,
+        worker_token: str | None = None,
         namespace: str = "default",
         timeout: float = 60.0,
         retry_policy: RetryPolicy | None = None,
@@ -147,6 +149,8 @@ def __init__(
         self._async = AsyncClient(
             base_url,
             token=token,
+            control_token=control_token,
+            worker_token=worker_token,
             namespace=namespace,
             timeout=timeout,
             retry_policy=retry_policy,

tests/test_client.py

@@ -51,6 +51,29 @@ def test_no_token(self) -> None:
         h = c._headers()
         assert "Authorization" not in h
 
+    def test_plane_scoped_tokens(self) -> None:
+        c = Client(
+            "http://localhost:8080",
+            token="legacy-token",
+            control_token="operator-token",
+            worker_token="worker-token",
+        )
+
+        control_headers = c._headers(worker=False)
+        worker_headers = c._headers(worker=True)
+
+        assert control_headers["Authorization"] == "Bearer operator-token"
+        assert worker_headers["Authorization"] == "Bearer worker-token"
+
+    def test_single_plane_token_can_fetch_cluster_info(self) -> None:
+        c = Client("http://localhost:8080", worker_token="worker-token")
+
+        control_headers = c._headers(worker=False)
+        worker_headers = c._headers(worker=True)
+
+        assert control_headers["Authorization"] == "Bearer worker-token"
+        assert worker_headers["Authorization"] == "Bearer worker-token"
+
 
 class TestStartWorkflow:
     @pytest.mark.asyncio