[Brook] feat(python): generate a caller per service over a replaceable transport

GUIDE.md

@@ -29,6 +29,7 @@ of `@nexus` directives.
   - [Omitted Fields](#omitted-fields)
   - [Flattened Records](#flattened-records)
   - [Output Transforms](#output-transforms)
+- [Callers Outside a Workflow](#callers-outside-a-workflow)
 - [Function References](#function-references)
 - [Directive Glossary](#directive-glossary)
 
@@ -1056,6 +1057,46 @@ return handles.FromSignalWithStart(request, &result)
 
 ---
 
+## Callers Outside a Workflow
+
+With `--client`, the Python target also writes `client.py` beside `models.py`.
+It declares one caller per service, `<Service>Client`, for code that is not in a
+Workflow, such as an Activity or a client process. The caller does not choose
+how a call travels. It takes a `Transport`, an object with one method:
+
+```python
+class Transport(typing.Protocol):
+    async def call(
+        self,
+        *,
+        service: str,
+        operation: str,
+        input: temporalio.api.common.v1.Payload,
+    ) -> temporalio.api.common.v1.Payload: ...
+```
+
+Each caller method converts its request to a payload, gives it to the transport,
+and converts the answer. A proto-backed model travels as a `binary/protobuf`
+payload of its proto message, so a transport that moves proto bytes needs
+nothing else. An answer for it can be `binary/protobuf`, `json/protobuf` or
+`json/plain`. The caller takes a `data_converter`, the SDK default when none is
+given. Each payload and failure inside a proto-backed model goes through its
+converters and codec, as in Workflow code. The model's own payload doesn't,
+since the service reads it. A model that is not proto-backed is one user value,
+and the data converter converts and encodes it whole. A transport raises
+`TransportError` for a call that failed, with `retryable` set when the same call
+can succeed later.
+
+```python
+client = StartWorkflowServiceClient(my_transport)
+response = await client.cancel_workflow(request)
+```
+
+The caller supports operations whose input and output are models of the same
+module. An operation whose models take type parameters, or that returns a
+resource without a wire type, is a generation error. TypeScript and .NET do not
+generate a caller yet.
+
 ## Function References
 
 Nexus and Temporal often need to identify a function by name on the wire, while

advanced/README.md

@@ -182,8 +182,11 @@ cargo run --features advanced -- dotnet \
 
 Inputs are positional; pass more than one path (a file or a directory) to link
 additional WIT into the parser workspace. Generation produces definitions by
-default. Add `--native-api` to also generate native API bindings. Add `--format`
-to run a formatter after generation:
+default. Add `--native-api` to also generate native API bindings. Add `--client`
+to also generate, in Python, a `client.py` with one caller per service for code
+outside a Workflow. The caller sends each call through a `Transport` that you
+give, so the same caller works over HTTP, a bridge, or a test double. Add
+`--format` to run a formatter after generation:
 
 - Python: `ruff format`
 - TypeScript: `prettier --write`

advanced/samples/inputs/examples.toml

@@ -12,3 +12,6 @@ languages = ["python", "dotnet", "typescript"]
 
 [notification-service]
 languages = ["python", "dotnet", "typescript"]
+
+[start-workflow]
+client = true

advanced/samples/python/tests/test_start_workflow.py

@@ -206,3 +206,38 @@ async def test_start_workflow_returns_wrapper_handle(
         result_error.value.args[0]
         == "started-workflow.get_result is not yet implemented"
     )
+
+
+async def test_client_calls_an_operation_over_a_transport() -> None:
+    from temporalio.api.common.v1 import Payload
+    from temporalio.api.workflowservice.v1 import (
+        RequestCancelWorkflowExecutionRequest,
+        RequestCancelWorkflowExecutionResponse,
+    )
+    import wit.start_workflow.client as start_workflow_client
+
+    sent: list[RequestCancelWorkflowExecutionRequest] = []
+
+    class Transport:
+        async def call(
+            self, *, service: str, operation: str, input: Payload
+        ) -> Payload:
+            assert (service, operation) == ("StartWorkflowService", "CancelWorkflow")
+            sent.append(RequestCancelWorkflowExecutionRequest.FromString(input.data))
+            answer = RequestCancelWorkflowExecutionResponse()
+            return Payload(
+                metadata={"encoding": b"binary/protobuf"},
+                data=answer.SerializeToString(),
+            )
+
+    client = start_workflow_client.StartWorkflowServiceClient(Transport())
+    request = start_workflow_models.CancelWorkflowRequest(
+        namespace="default",
+        workflow_execution=start_workflow_models.WorkflowExecution(workflow_id="wf"),
+        reason="done",
+    )
+    response = await client.cancel_workflow(request)
+
+    assert isinstance(response, start_workflow_models.CancelWorkflowResponse)
+    assert sent[0].workflow_execution.workflow_id == "wf"
+    assert sent[0].reason == "done"

advanced/samples/python/wit/start_workflow/client.py

@@ -0,0 +1,230 @@
+# Generated by nexgen latest. DO NOT EDIT!
+
+from __future__ import annotations
+
+import collections.abc
+import contextlib
+import typing
+
+import google.protobuf.json_format
+import google.protobuf.message
+import temporalio.api.common.v1
+import temporalio.converter
+import temporalio.nexus.system
+
+from .models import (
+    CancelWorkflowRequest,
+    CancelWorkflowResponse,
+    StartWorkflowRequest,
+    StartWorkflowResult,
+    _CancelWorkflowRequestTransferTypeConverter,
+    _CancelWorkflowResponseTransferTypeConverter,
+    _StartWorkflowRequestTransferTypeConverter,
+    _StartWorkflowResultTransferTypeConverter,
+)
+
+
+class Transport(typing.Protocol):
+    """Carries one operation call to a service and its answer back.
+
+    A transport moves payloads only. The caller converts its models to and
+    from payloads. A proto-backed model travels as a ``binary/protobuf``
+    payload, and an answer for one can be ``binary/protobuf``,
+    ``json/protobuf`` or ``json/plain``.
+    """
+
+    async def call(
+        self,
+        *,
+        service: str,
+        operation: str,
+        input: temporalio.api.common.v1.Payload,
+    ) -> temporalio.api.common.v1.Payload:
+        """Send one call and return its answer.
+
+        Raise :py:class:`TransportError` for a call that failed.
+        """
+        ...
+
+
+class TransportError(Exception):
+    """A call that the transport could not complete.
+
+    ``retryable`` says whether the same call can succeed later.
+    """
+
+    def __init__(self, message: str, *, retryable: bool = False) -> None:
+        """Create a transport error."""
+        super().__init__(message)
+        self.retryable: bool = retryable
+
+
+_MessageT = typing.TypeVar("_MessageT", bound=google.protobuf.message.Message)
+
+
+def _proto_payload(
+    message: google.protobuf.message.Message,
+) -> temporalio.api.common.v1.Payload:
+    return temporalio.api.common.v1.Payload(
+        metadata={
+            "encoding": b"binary/protobuf",
+            "messageType": message.DESCRIPTOR.full_name.encode(),
+        },
+        data=message.SerializeToString(),
+    )
+
+
+def _proto_message(
+    payload: temporalio.api.common.v1.Payload,
+    message_type: type[_MessageT] | None,
+) -> _MessageT:
+    if message_type is None:
+        raise TypeError("the answer model has no proto message type")
+    message = message_type()
+    encoding = payload.metadata.get("encoding", b"")
+    if encoding == b"binary/protobuf":
+        _ = message.ParseFromString(payload.data)
+    elif encoding in (b"json/protobuf", b"json/plain"):
+        _ = google.protobuf.json_format.Parse(payload.data, message)
+    elif encoding != b"binary/null":
+        raise ValueError(
+            f"cannot read {message.DESCRIPTOR.full_name} from a {encoding.decode()!r} payload"
+        )
+    return message
+
+
+# The codec runs on the user payloads inside a proto-backed model, failure details
+# included, and never on the model's own payload, which the service reads.
+def _payloads_in(
+    message: google.protobuf.message.Message,
+) -> list[temporalio.api.common.v1.Payload]:
+    if isinstance(message, temporalio.api.common.v1.Payload):
+        return [message]
+    found: list[temporalio.api.common.v1.Payload] = []
+    for _, value in message.ListFields():
+        if isinstance(value, google.protobuf.message.Message):
+            found.extend(_payloads_in(value))
+        elif not isinstance(value, (str, bytes, int, float, bool)):
+            items = typing.cast(
+                "collections.abc.Iterable[object]",
+                value.values() if hasattr(value, "values") else value,
+            )
+            for item in items:
+                if isinstance(item, google.protobuf.message.Message):
+                    found.extend(_payloads_in(item))
+    return found
+
+
+async def _encode_payloads(
+    message: google.protobuf.message.Message,
+    codec: temporalio.converter.PayloadCodec | None,
+) -> None:
+    if codec is None:
+        return
+    for payload in _payloads_in(message):
+        [encoded] = await codec.encode([payload])
+        payload.CopyFrom(encoded)
+
+
+async def _decode_payloads(
+    message: google.protobuf.message.Message,
+    codec: temporalio.converter.PayloadCodec | None,
+) -> None:
+    if codec is None:
+        return
+    for payload in _payloads_in(message):
+        [decoded] = await codec.decode([payload])
+        payload.CopyFrom(decoded)
+
+
+class StartWorkflowServiceClient:
+    """Calls the `StartWorkflowService` operations over a transport.
+
+    For code outside a Workflow. Workflow code uses the operation stubs.
+    """
+
+    def __init__(
+        self,
+        transport: Transport,
+        *,
+        data_converter: temporalio.converter.DataConverter | None = None,
+    ) -> None:
+        """Create a caller.
+
+        Args:
+            transport: Carries each call to the service.
+            data_converter: Converts and encodes the user values. A proto-backed
+                model travels as is, and each payload and failure inside it goes
+                through the converter and the codec, as in Workflow code. A model
+                that is not proto-backed is one user value. Defaults to the SDK's
+                default data converter.
+        """
+        self._transport: Transport = transport
+        self._data_converter: temporalio.converter.DataConverter = (
+            data_converter or temporalio.converter.DataConverter.default
+        )
+
+    def _converters(self) -> contextlib.AbstractContextManager[None]:
+        return temporalio.nexus.system._user_converter_context(
+            temporalio.nexus.system._SystemNexusUserConverters(
+                payload_converter=self._data_converter.payload_converter,
+                failure_converter=self._data_converter.failure_converter,
+            )
+        )
+
+    async def _call(
+        self, operation: str, payload: temporalio.api.common.v1.Payload
+    ) -> temporalio.api.common.v1.Payload:
+        return await self._transport.call(
+            service="StartWorkflowService",
+            operation=operation,
+            input=payload,
+        )
+
+    async def start_workflow(
+        self, request: StartWorkflowRequest
+    ) -> StartWorkflowResult:
+        with self._converters():
+            message = _StartWorkflowRequestTransferTypeConverter().to_transfer_type(
+                request
+            )
+        await _encode_payloads(message, self._data_converter.payload_codec)
+        payload = _proto_payload(message)
+        answer = await self._call("StartWorkflow", payload)
+        converter = _StartWorkflowResultTransferTypeConverter()
+        message = _proto_message(answer, converter.transfer_type)
+        await _decode_payloads(message, self._data_converter.payload_codec)
+        with self._converters():
+            return converter.from_transfer_type(message, StartWorkflowResult)
+
+    async def restart_workflow(
+        self, request: StartWorkflowRequest
+    ) -> StartWorkflowResult:
+        with self._converters():
+            message = _StartWorkflowRequestTransferTypeConverter().to_transfer_type(
+                request
+            )
+        await _encode_payloads(message, self._data_converter.payload_codec)
+        payload = _proto_payload(message)
+        answer = await self._call("RestartWorkflow", payload)
+        converter = _StartWorkflowResultTransferTypeConverter()
+        message = _proto_message(answer, converter.transfer_type)
+        await _decode_payloads(message, self._data_converter.payload_codec)
+        with self._converters():
+            return converter.from_transfer_type(message, StartWorkflowResult)
+
+    async def cancel_workflow(
+        self, request: CancelWorkflowRequest
+    ) -> CancelWorkflowResponse:
+        with self._converters():
+            message = _CancelWorkflowRequestTransferTypeConverter().to_transfer_type(
+                request
+            )
+        await _encode_payloads(message, self._data_converter.payload_codec)
+        payload = _proto_payload(message)
+        answer = await self._call("CancelWorkflow", payload)
+        converter = _CancelWorkflowResponseTransferTypeConverter()
+        message = _proto_message(answer, converter.transfer_type)
+        await _decode_payloads(message, self._data_converter.payload_codec)
+        with self._converters():
+            return converter.from_transfer_type(message, CancelWorkflowResponse)

changelog/added/relay-raccoon.md

@@ -0,0 +1 @@
+Python generation accepts `--client`, which writes `client.py` with one caller per service over a `Transport` that the caller gives.

src/error.rs

@@ -215,6 +215,14 @@ pub enum Error {
     )]
     IncompleteOperationOutputTransform { service: String, operation: String },
 
+    #[error("{language} cannot generate a caller for operation `{service}.{operation}`: {reason}")]
+    UnsupportedClientOperation {
+        language: Language,
+        service: String,
+        operation: String,
+        reason: String,
+    },
+
     #[error(
         "service `{service}` operation `{operation}` references unknown input proto type `{type_name}`"
     )]

src/generator/go.rs

@@ -5838,6 +5838,7 @@ mod tests {
             config: crate::nexgen_config::NexgenConfig {
                 mode: crate::generator::GenerationMode::NativeApi,
                 system_nexus,
+                client: false,
             },
             language: crate::language::Language::Go,
             input_paths: vec![input],