[cross-repo from workflow#130] GitHub #481: server + workflow + sdk-python + cloud: invocable carrier for activity-grade external execution (#30)

Author: durable-workflow-ops

docs/reference/invocable.md

@@ -20,4 +20,32 @@ async def handle_request(request_json: dict) -> dict:
     return await adapter.handle(request_json)
 ```
 
+For HTTP endpoints or API Gateway/Lambda style handlers, use the packaged
+adapters so content type, body parsing, and result serialization stay aligned
+with the carrier contract:
+
+```python
+from durable_workflow import handle_invocable_http_request, lambda_invocable_activity_handler
+
+
+async def handle_http(body: bytes):
+    return await handle_invocable_http_request(body, {"billing.charge-card": charge_card})
+
+
+lambda_handler = lambda_invocable_activity_handler({"billing.charge-card": charge_card})
+```
+
+`handle_invocable_http_request` returns an `InvocableHttpResponse` with
+`status_code`, `headers`, and `body`. On success (HTTP 200) the body is the
+external-task result envelope with `Content-Type:
+application/vnd.durable-workflow.external-task-result+json`. On a bad or
+unparseable request body it returns HTTP 400 with a JSON error object — no
+durable task identity is available in that case so no structured result envelope
+can be built.
+
+`lambda_invocable_activity_handler` wraps the async helper in a synchronous
+AWS Lambda / API Gateway handler. It decodes base64 bodies when
+`isBase64Encoded` is `true` so it works with both REST API and HTTP API
+integrations without additional configuration.
+
 ::: durable_workflow.invocable

src/durable_workflow/__init__.py

@@ -119,7 +119,14 @@
     WorkflowTaskHandler,
     WorkflowTaskInterceptorContext,
 )
-from .invocable import InvocableActivityHandler, handle_invocable_activity
+from .invocable import (
+    InvocableActivityHandler,
+    InvocableHttpResponse,
+    handle_invocable_activity,
+    handle_invocable_http_request,
+    handle_invocable_lambda_event,
+    lambda_invocable_activity_handler,
+)
 from .metrics import (
     InMemoryMetrics,
     MetricsRecorder,
@@ -249,6 +256,7 @@
     "InvalidArgument",
     "InMemoryMetrics",
     "InvocableActivityHandler",
+    "InvocableHttpResponse",
     "GCSExternalStorage",
     "LocalFilesystemExternalStorage",
     "MetricsRecorder",
@@ -286,6 +294,9 @@
     "parse_external_task_input_artifact",
     "parse_external_task_result",
     "handle_invocable_activity",
+    "handle_invocable_http_request",
+    "handle_invocable_lambda_event",
+    "lambda_invocable_activity_handler",
     "store_external_payload",
     "to_avro_payload_value",
     "to_avro_payload_values",

src/durable_workflow/invocable.py

@@ -1,18 +1,25 @@
 from __future__ import annotations
 
 import asyncio
+import base64
 import inspect
+import json
 import time
 import traceback
 from collections.abc import Awaitable, Callable, Mapping
+from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Any
+from typing import Any, cast
 
 from . import serializer
 from .errors import ActivityCancelled, NonRetryableError
 from .external_storage import ExternalPayloadCache, ExternalStorageDriver
 from .external_task_input import ExternalTaskInput, parse_external_task_input
-from .external_task_result import EXTERNAL_TASK_RESULT_SCHEMA, EXTERNAL_TASK_RESULT_VERSION
+from .external_task_result import (
+    EXTERNAL_TASK_RESULT_MEDIA_TYPE,
+    EXTERNAL_TASK_RESULT_SCHEMA,
+    EXTERNAL_TASK_RESULT_VERSION,
+)
 
 InvocableActivityCallable = Callable[..., Any | Awaitable[Any]]
 
@@ -315,3 +322,123 @@ async def handle_invocable_activity(
     """Handle one invocable activity task with a temporary adapter instance."""
 
     return await InvocableActivityHandler(handlers, **options).handle(envelope)
+
+
+@dataclass(frozen=True)
+class InvocableHttpResponse:
+    """Structured HTTP response from an invocable activity carrier endpoint."""
+
+    status_code: int
+    headers: Mapping[str, str]
+    body: str
+
+    def json(self) -> dict[str, Any]:
+        decoded = json.loads(self.body)
+        if not isinstance(decoded, dict):
+            raise ValueError("invocable HTTP response body is not a JSON object")
+        return cast(dict[str, Any], decoded)
+
+
+async def handle_invocable_http_request(
+    body: bytes | str | Mapping[str, Any],
+    handlers: Mapping[str, InvocableActivityCallable],
+    **options: Any,
+) -> InvocableHttpResponse:
+    """Handle one HTTP-addressed invocable activity request.
+
+    The server expects a structured external-task result envelope on HTTP 200.
+    Bad request bodies return HTTP 400 because no durable task identity can be
+    recovered for a valid failure envelope.
+    """
+
+    try:
+        envelope = _coerce_json_object(body)
+    except (TypeError, ValueError) as exc:
+        return InvocableHttpResponse(
+            status_code=400,
+            headers={"Content-Type": "application/json"},
+            body=_json_dump({"error": "invalid_invocable_request", "message": str(exc)}),
+        )
+
+    try:
+        result = await handle_invocable_activity(envelope, handlers, **options)
+    except Exception as exc:
+        return InvocableHttpResponse(
+            status_code=400,
+            headers={"Content-Type": "application/json"},
+            body=_json_dump({"error": "invalid_invocable_request", "message": str(exc)}),
+        )
+
+    return InvocableHttpResponse(
+        status_code=200,
+        headers={"Content-Type": EXTERNAL_TASK_RESULT_MEDIA_TYPE},
+        body=_json_dump(result),
+    )
+
+
+async def handle_invocable_lambda_event(
+    event: Mapping[str, Any],
+    handlers: Mapping[str, InvocableActivityCallable],
+    **options: Any,
+) -> dict[str, Any]:
+    """Handle an AWS Lambda / API Gateway style invocable activity event."""
+
+    try:
+        body = _lambda_event_body(event)
+    except (TypeError, ValueError) as exc:
+        response = InvocableHttpResponse(
+            status_code=400,
+            headers={"Content-Type": "application/json"},
+            body=_json_dump({"error": "invalid_invocable_request", "message": str(exc)}),
+        )
+    else:
+        response = await handle_invocable_http_request(body, handlers, **options)
+
+    return {
+        "statusCode": response.status_code,
+        "headers": dict(response.headers),
+        "body": response.body,
+        "isBase64Encoded": False,
+    }
+
+
+def lambda_invocable_activity_handler(
+    handlers: Mapping[str, InvocableActivityCallable],
+    **options: Any,
+) -> Callable[[Mapping[str, Any], Any], dict[str, Any]]:
+    """Build a synchronous AWS Lambda handler for invocable activities."""
+
+    def _handler(event: Mapping[str, Any], context: Any) -> dict[str, Any]:
+        return asyncio.run(handle_invocable_lambda_event(event, handlers, **options))
+
+    return _handler
+
+
+def _coerce_json_object(body: bytes | str | Mapping[str, Any]) -> dict[str, Any]:
+    if isinstance(body, Mapping):
+        return dict(body)
+
+    if isinstance(body, bytes):
+        body = body.decode("utf-8")
+
+    if not isinstance(body, str):
+        raise TypeError("request body must be bytes, str, or a JSON object")
+
+    decoded = json.loads(body)
+    if not isinstance(decoded, dict):
+        raise ValueError("request body must decode to a JSON object")
+
+    return cast(dict[str, Any], decoded)
+
+
+def _lambda_event_body(event: Mapping[str, Any]) -> bytes | str | Mapping[str, Any]:
+    body = event.get("body")
+    if isinstance(body, str) and event.get("isBase64Encoded") is True:
+        return base64.b64decode(body)
+    if isinstance(body, (bytes, str, Mapping)):
+        return cast(bytes | str | Mapping[str, Any], body)
+    raise ValueError("Lambda event body must contain the invocable request JSON")
+
+
+def _json_dump(value: Mapping[str, Any]) -> str:
+    return json.dumps(value, separators=(",", ":"), sort_keys=True)

tests/test_invocable.py

@@ -1,4 +1,5 @@
 import asyncio
+import base64
 import copy
 import json
 import time
@@ -11,7 +12,12 @@
 from durable_workflow import serializer
 from durable_workflow.errors import NonRetryableError
 from durable_workflow.external_task_result import parse_external_task_result
-from durable_workflow.invocable import InvocableActivityHandler, handle_invocable_activity
+from durable_workflow.invocable import (
+    InvocableActivityHandler,
+    handle_invocable_activity,
+    handle_invocable_http_request,
+    lambda_invocable_activity_handler,
+)
 
 FIXTURES = Path(__file__).parent / "fixtures" / "external-task-input"
 
@@ -187,3 +193,70 @@ async def test_invocable_activity_handler_rejects_workflow_task_inputs() -> None
 async def test_invocable_activity_handler_validates_result_codec() -> None:
     with pytest.raises(ValueError, match="unsupported invocable result codec"):
         InvocableActivityHandler({}, result_codec="protobuf")
+
+
+async def test_invocable_http_request_helper_returns_result_response() -> None:
+    response = await handle_invocable_http_request(
+        json.dumps(activity_input(42)),
+        {"billing.charge-card": lambda amount: {"amount": amount}},
+        result_codec=serializer.JSON_CODEC,
+    )
+
+    assert response.status_code == 200
+    assert response.headers["Content-Type"] == "application/vnd.durable-workflow.external-task-result+json"
+
+    result = parse_external_task_result(response.json())
+    assert result.succeeded is True
+    assert result.result is not None
+    assert serializer.decode_envelope(result.result["payload"]) == {"amount": 42}
+
+
+async def test_invocable_http_request_helper_accepts_bytes_body() -> None:
+    response = await handle_invocable_http_request(
+        json.dumps(activity_input(7)).encode(),
+        {"billing.charge-card": lambda amount: {"amount": amount}},
+        result_codec=serializer.JSON_CODEC,
+    )
+
+    assert response.status_code == 200
+    result = parse_external_task_result(response.json())
+    assert result.succeeded is True
+
+
+async def test_invocable_http_request_helper_rejects_invalid_json_body() -> None:
+    response = await handle_invocable_http_request("{", {})
+
+    assert response.status_code == 400
+    assert response.json()["error"] == "invalid_invocable_request"
+
+
+def test_lambda_invocable_activity_handler_accepts_base64_api_gateway_body() -> None:
+    handler = lambda_invocable_activity_handler(
+        {"billing.charge-card": lambda amount: {"amount": amount}},
+        result_codec=serializer.JSON_CODEC,
+    )
+    body = json.dumps(activity_input(99)).encode()
+
+    response = handler(
+        {
+            "body": base64.b64encode(body).decode(),
+            "isBase64Encoded": True,
+        },
+        None,
+    )
+
+    assert response["statusCode"] == 200
+    assert response["headers"]["Content-Type"] == "application/vnd.durable-workflow.external-task-result+json"
+
+    result = parse_external_task_result(json.loads(response["body"]))
+    assert result.succeeded is True
+    assert result.result is not None
+    assert serializer.decode_envelope(result.result["payload"]) == {"amount": 99}
+
+
+def test_lambda_invocable_activity_handler_rejects_missing_body() -> None:
+    handler = lambda_invocable_activity_handler({})
+    response = handler({"body": None}, None)
+
+    assert response["statusCode"] == 400
+    assert json.loads(response["body"])["error"] == "invalid_invocable_request"