Add Python bridge webhook client

Author: durable-workflow-ops

README.md

@@ -420,6 +420,27 @@ decisions for success, retryability, malformed output, cancellation, deadline
 exceeded, handler crash, decode failure, and unsupported payload
 codec/reference states without treating stderr as a machine signal.
 
+Bridge adapters can hand bounded webhook ingress into the server through
+`Client.send_webhook_bridge_event()`. The method returns the server's typed
+bridge outcome for accepted, duplicate, and rejected events, including
+machine-readable HTTP 422 rejection outcomes:
+
+```python
+outcome = await client.send_webhook_bridge_event(
+    "pagerduty",
+    action="signal_workflow",
+    idempotency_key="pagerduty-event-3003",
+    target={"workflow_id": "wf-remediation-42", "signal_name": "incident_escalated"},
+    input={"severity": "critical", "service": "checkout"},
+    correlation={"provider": "pagerduty", "event_type": "incident.triggered"},
+)
+
+if outcome.accepted:
+    print(outcome.workflow_id, outcome.control_plane_outcome)
+else:
+    print(outcome.outcome, outcome.reason)
+```
+
 ## Development
 
 ```bash

src/durable_workflow/__init__.py

@@ -17,6 +17,7 @@
     parse_auth_composition_contract,
 )
 from .client import (
+    BridgeAdapterOutcome,
     Client,
     ScheduleAction,
     ScheduleBackfillResult,
@@ -130,6 +131,7 @@
     "ActivityInfo",
     "ActivityInterceptorContext",
     "ActivityRetryPolicy",
+    "BridgeAdapterOutcome",
     "ChildWorkflowRetryPolicy",
     "ChildWorkflowFailed",
     "Client",

src/durable_workflow/client.py

@@ -70,6 +70,8 @@ def _route_for_metrics(path: str) -> str:
             parts[3] = "{run_id}"
     elif parts[0] == "schedules" and len(parts) >= 2:
         parts[1] = "{schedule_id}"
+    elif parts[:2] == ["bridge-adapters", "webhook"] and len(parts) >= 3:
+        parts[2] = "{adapter}"
     elif (
         parts[:2] == ["worker", "workflow-tasks"]
         or parts[:2] == ["worker", "activity-tasks"]
@@ -362,6 +364,47 @@ class ScheduleBackfillResult:
     results: list[dict[str, Any]] | None = None
 
 
+@dataclass
+class BridgeAdapterOutcome:
+    """Machine-readable result returned by a bridge adapter event."""
+
+    schema: str
+    version: int
+    adapter: str
+    action: str | None
+    accepted: bool
+    outcome: str
+    idempotency_key: str | None = None
+    reason: str | None = None
+    target: dict[str, Any] | None = None
+    correlation: dict[str, Any] | None = None
+    workflow_id: str | None = None
+    run_id: str | None = None
+    workflow_type: str | None = None
+    control_plane_outcome: str | None = None
+    raw: dict[str, Any] | None = None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> BridgeAdapterOutcome:
+        return cls(
+            schema=str(data.get("schema", "")),
+            version=int(data.get("version", 0)),
+            adapter=str(data.get("adapter", "")),
+            action=data.get("action"),
+            accepted=bool(data.get("accepted", False)),
+            outcome=str(data.get("outcome", "")),
+            idempotency_key=data.get("idempotency_key"),
+            reason=data.get("reason"),
+            target=data.get("target") if isinstance(data.get("target"), dict) else None,
+            correlation=data.get("correlation") if isinstance(data.get("correlation"), dict) else None,
+            workflow_id=data.get("workflow_id"),
+            run_id=data.get("run_id"),
+            workflow_type=data.get("workflow_type"),
+            control_plane_outcome=data.get("control_plane_outcome"),
+            raw=data,
+        )
+
+
 class WorkflowHandle:
     """Convenience wrapper for operating on one workflow ID."""
 
@@ -696,6 +739,68 @@ async def _do_request() -> httpx.Response:
             self.metrics.increment(CLIENT_REQUESTS, tags=tags)
             self.metrics.record(CLIENT_REQUEST_DURATION_SECONDS, time.perf_counter() - start, tags=tags)
 
+    async def _request_bridge_outcome(self, path: str, *, json: Any = None, context: str = "") -> dict[str, Any]:
+        start = time.perf_counter()
+        route = _route_for_metrics(path)
+        status_code = "none"
+        outcome = "pending"
+
+        async def _do_request() -> httpx.Response:
+            resp = await self._http.request(
+                "POST",
+                f"/api{path}",
+                headers=self._headers(worker=False),
+                json=json,
+            )
+            if resp.status_code != 422:
+                resp.raise_for_status()
+            return resp
+
+        try:
+            try:
+                resp = await self.retry_policy.execute(_do_request)
+            except httpx.HTTPStatusError as exc:
+                status_code = str(exc.response.status_code)
+                outcome = "http_error"
+                try:
+                    body = exc.response.json()
+                except ValueError:
+                    body = exc.response.text
+                _raise_for_status(exc.response.status_code, body, context=context)
+                raise
+
+            status_code = str(resp.status_code)
+            if not resp.content:
+                raise ServerError(
+                    resp.status_code,
+                    {"reason": "invalid_bridge_outcome", "message": "expected JSON object, got empty response"},
+                )
+            data = resp.json()
+            if not isinstance(data, dict):
+                raise ServerError(
+                    resp.status_code,
+                    {
+                        "reason": "invalid_bridge_outcome",
+                        "message": f"expected JSON object, got {type(data).__name__}",
+                    },
+                )
+            outcome = "bridge_rejected" if resp.status_code == 422 else "ok"
+            return data
+        except Exception as exc:
+            if outcome == "pending":
+                outcome = type(exc).__name__
+            raise
+        finally:
+            tags = {
+                "method": "POST",
+                "route": route,
+                "plane": "control",
+                "status_code": status_code,
+                "outcome": outcome,
+            }
+            self.metrics.increment(CLIENT_REQUESTS, tags=tags)
+            self.metrics.record(CLIENT_REQUEST_DURATION_SECONDS, time.perf_counter() - start, tags=tags)
+
     async def get_cluster_info(self) -> dict[str, Any]:
         """Fetch server build identity, capabilities, and protocol manifests."""
         result = await self._request("GET", "/cluster/info", worker=False, context="get_cluster_info")
@@ -911,6 +1016,40 @@ async def get_history(self, workflow_id: str, run_id: str) -> Any:
             "GET", f"/workflows/{workflow_id}/runs/{run_id}/history", context=workflow_id
         )
 
+    async def send_webhook_bridge_event(
+        self,
+        adapter: str,
+        *,
+        action: str,
+        idempotency_key: str,
+        target: dict[str, Any],
+        input: dict[str, Any] | None = None,
+        correlation: dict[str, Any] | None = None,
+    ) -> BridgeAdapterOutcome:
+        """Send one bounded webhook bridge event and return its contract outcome.
+
+        The bridge endpoint intentionally returns machine-readable rejected
+        outcomes as HTTP 422. This method returns those outcomes instead of
+        raising :class:`InvalidArgument`, while auth and unexpected server
+        failures still use the normal SDK exception mapping.
+        """
+        body: dict[str, Any] = {
+            "action": action,
+            "idempotency_key": idempotency_key,
+            "target": target,
+        }
+        if input is not None:
+            body["input"] = input
+        if correlation is not None:
+            body["correlation"] = correlation
+
+        data = await self._request_bridge_outcome(
+            f"/bridge-adapters/webhook/{quote(adapter, safe='._:-')}",
+            json=body,
+            context=f"bridge adapter {adapter}",
+        )
+        return BridgeAdapterOutcome.from_dict(data)
+
     async def signal_workflow(
         self, workflow_id: str, signal_name: str, *, args: list[Any] | None = None
     ) -> None:

tests/test_client.py

@@ -335,6 +335,145 @@ async def test_signal_request_matches_polyglot_fixture(self, client: Client) ->
         assert sdk["args"]["signal_name"] == semantic["signal_name"]
 
 
+class TestWebhookBridgeAdapters:
+    @pytest.mark.asyncio
+    async def test_start_workflow_bridge_event_returns_accepted_outcome(self, client: Client) -> None:
+        resp = _mock_response(202, {
+            "schema": "durable-workflow.v2.bridge-adapter-outcome.contract",
+            "version": 1,
+            "adapter": "stripe",
+            "action": "start_workflow",
+            "accepted": True,
+            "outcome": "accepted",
+            "idempotency_key": "stripe-event-1001",
+            "target": {
+                "workflow_id": "bridge-stripe-derived",
+                "workflow_type": "orders.fulfillment",
+                "task_queue": "external-workflows",
+                "business_key": "order-1001",
+            },
+            "correlation": {
+                "provider": "stripe",
+                "event_type": "checkout.session.completed",
+            },
+            "workflow_id": "bridge-stripe-derived",
+            "run_id": "run-bridge-1",
+            "workflow_type": "orders.fulfillment",
+            "control_plane_outcome": "started_new",
+        })
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            outcome = await client.send_webhook_bridge_event(
+                "stripe",
+                action="start_workflow",
+                idempotency_key="stripe-event-1001",
+                target={
+                    "workflow_type": "orders.fulfillment",
+                    "task_queue": "external-workflows",
+                    "business_key": "order-1001",
+                    "duplicate_policy": "use_existing",
+                },
+                input={"order_id": "order-1001"},
+                correlation={
+                    "provider": "stripe",
+                    "event_type": "checkout.session.completed",
+                },
+            )
+
+        assert outcome.accepted is True
+        assert outcome.outcome == "accepted"
+        assert outcome.workflow_id == "bridge-stripe-derived"
+        assert outcome.control_plane_outcome == "started_new"
+        assert outcome.target is not None
+        assert outcome.target["business_key"] == "order-1001"
+
+        call_args = mock.call_args
+        assert call_args.args[0] == "POST"
+        assert call_args.args[1] == "/api/bridge-adapters/webhook/stripe"
+        body = call_args.kwargs.get("json") or call_args[1].get("json")
+        assert body == {
+            "action": "start_workflow",
+            "idempotency_key": "stripe-event-1001",
+            "target": {
+                "workflow_type": "orders.fulfillment",
+                "task_queue": "external-workflows",
+                "business_key": "order-1001",
+                "duplicate_policy": "use_existing",
+            },
+            "input": {"order_id": "order-1001"},
+            "correlation": {
+                "provider": "stripe",
+                "event_type": "checkout.session.completed",
+            },
+        }
+
+    @pytest.mark.asyncio
+    async def test_signal_bridge_event_returns_rejected_outcome_for_422(self, client: Client) -> None:
+        resp = _mock_response(422, {
+            "schema": "durable-workflow.v2.bridge-adapter-outcome.contract",
+            "version": 1,
+            "adapter": "pagerduty",
+            "action": "signal_workflow",
+            "accepted": False,
+            "outcome": "rejected",
+            "reason": "unknown_target",
+            "idempotency_key": "pagerduty-event-3003",
+            "target": {
+                "workflow_id": "wf-remediation-42",
+                "signal_name": "incident_escalated",
+            },
+            "correlation": {
+                "provider": "pagerduty",
+                "event_type": "incident.triggered",
+            },
+        })
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp):
+            outcome = await client.send_webhook_bridge_event(
+                "pagerduty",
+                action="signal_workflow",
+                idempotency_key="pagerduty-event-3003",
+                target={
+                    "workflow_id": "wf-remediation-42",
+                    "signal_name": "incident_escalated",
+                },
+                input={
+                    "severity": "critical",
+                    "service": "checkout",
+                },
+                correlation={
+                    "provider": "pagerduty",
+                    "event_type": "incident.triggered",
+                },
+            )
+
+        assert outcome.accepted is False
+        assert outcome.outcome == "rejected"
+        assert outcome.reason == "unknown_target"
+        assert outcome.idempotency_key == "pagerduty-event-3003"
+
+    @pytest.mark.asyncio
+    async def test_bridge_adapter_path_escapes_adapter_segment(self, client: Client) -> None:
+        resp = _mock_response(202, {
+            "schema": "durable-workflow.v2.bridge-adapter-outcome.contract",
+            "version": 1,
+            "adapter": "ops/foo",
+            "action": "update_workflow",
+            "accepted": True,
+            "outcome": "accepted",
+        })
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            await client.send_webhook_bridge_event(
+                "ops/foo",
+                action="update_workflow",
+                idempotency_key="evt-1",
+                target={"workflow_id": "wf-1", "update_name": "acknowledge"},
+            )
+
+        assert mock.call_args.args[1] == "/api/bridge-adapters/webhook/ops%2Ffoo"
+
+
 class TestCancelWorkflow:
     @pytest.mark.asyncio
     async def test_cancel(self, client: Client) -> None: