Nexus

Author: dandavison

README.md

@@ -96,6 +96,8 @@ informal introduction to the features and their implementation.
         - [Heartbeating and Cancellation](#heartbeating-and-cancellation)
         - [Worker Shutdown](#worker-shutdown)
       - [Testing](#testing-1)
+    - [Nexus](#nexus)
+      - [hello](#hello)
     - [Workflow Replay](#workflow-replay)
     - [Observability](#observability)
       - [Metrics](#metrics)
@@ -1314,6 +1316,71 @@ affect calls activity code might make to functions on the `temporalio.activity`
 * `cancel()` can be invoked to simulate a cancellation of the activity
 * `worker_shutdown()` can be invoked to simulate a worker shutdown during execution of the activity
 
+
+### Nexus
+
+See [docs.temporal.io/nexus](https://docs.temporal.io/nexus).
+
+#### Service Interface Definition
+
+A Nexus Service interface definition is a set of named operations, where each operation is an
+`(input_type, output_type)` pair:
+
+```python
+@nexusrpc.service
+class MyNexusService:
+    my_operation: nexusrpc.Operation[MyOpInput, MyOpOutput]
+```
+
+### Operation implementation
+
+```python
+@nexusrpc.service(interface.MyNexusService)
+class MyNexusService:
+
+    @nexusrpc.sync_operation
+    def echo(self, input: EchoInput) -> EchoOutput:
+        return EchoOutput(message=input.message)
+```
+
+```python
+@nexusrpc.service(interface.MyNexusService)
+class MyNexusService:
+
+    @temporalio.nexus.workflow_operation
+    async def hello(
+        self, input: HelloInput
+    ) -> AsyncWorkflowOperationResult[HelloOutput]:
+        return await temporalio.nexus.handler.start_workflow(HelloWorkflow.run, input)
+```
+
+
+### Request options
+
+```python
+@dataclass
+class OperationOptions:
+    """Options passed by the Nexus caller when starting an operation."""
+
+    # A callback URL is required to deliver the completion of an async operation. This URL should be
+    # called by a handler upon completion if the started operation is async.
+    callback_url: Optional[str] = None
+
+    # Optional header fields set by the caller to be attached to the callback request when an
+    # asynchronous operation completes.
+    callback_header: dict[str, str] = field(default_factory=dict)
+
+    # Request ID that may be used by the server handler to dedupe a start request.
+    # By default a v4 UUID will be generated by the client.
+    request_id: Optional[str] = None
+
+    # Links contain arbitrary caller information. Handlers may use these links as
+    # metadata on resources associated with an operation.
+    links: list[Link] = field(default_factory=list)
+```
+
+
+
 ### Workflow Replay
 
 Given a workflow's history, it can be replayed locally to check for things like non-determinism errors. For example,

pyproject.toml

@@ -11,8 +11,12 @@ keywords = [
     "workflow",
 ]
 dependencies = [
+    "hyperlinked",
+    "nexus-rpc",
+    "pdbpp>=0.11.6",
     "protobuf>=3.20",
     "python-dateutil>=2.8.2,<3 ; python_version < '3.11'",
+    "temporalio-xray",
     "types-protobuf>=3.20",
     "typing-extensions>=4.2.0,<5",
 ]
@@ -40,7 +44,7 @@ dev = [
     "psutil>=5.9.3,<6",
     "pydocstyle>=6.3.0,<7",
     "pydoctor>=24.11.1,<25",
-    "pyright==1.1.377",
+    "pyright==1.1.400",
     "pytest~=7.4",
     "pytest-asyncio>=0.21,<0.22",
     "pytest-timeout~=2.2",
@@ -49,6 +53,7 @@ dev = [
     "twine>=4.0.1,<5",
     "ruff>=0.5.0,<0.6",
     "maturin>=1.8.2",
+    "pytest-cov>=6.1.1",
 ]
 
 [tool.poe.tasks]
@@ -60,8 +65,8 @@ gen-protos = "uv run python scripts/gen_protos.py"
 lint = [
   {cmd = "uv run ruff check --select I"},
   {cmd = "uv run ruff format --check"},
-  {ref = "lint-types"},
   {cmd = "uv run pyright"},
+  {ref = "lint-types"},
   {ref = "lint-docs"},
 ]
 bridge-lint = { cmd = "cargo clippy -- -D warnings", cwd = "temporalio/bridge" }
@@ -70,7 +75,7 @@ bridge-lint = { cmd = "cargo clippy -- -D warnings", cwd = "temporalio/bridge" }
 lint-docs = "uv run pydocstyle --ignore-decorators=overload"
 lint-types = "uv run mypy --namespace-packages --check-untyped-defs ."
 run-bench = "uv run python scripts/run_bench.py"
-test = "uv run pytest"
+test = "uv run pytest --cov temporalio --cov-report xml"
 
 
 [tool.pytest.ini_options]
@@ -83,8 +88,6 @@ testpaths = ["tests"]
 timeout = 600
 timeout_func_only = true
 filterwarnings = [
-  "error::temporalio.workflow.UnfinishedUpdateHandlersWarning",
-  "error::temporalio.workflow.UnfinishedSignalHandlersWarning",
   "ignore::pytest.PytestDeprecationWarning",
   "ignore::DeprecationWarning",
 ]
@@ -157,6 +160,7 @@ exclude = [
   "tests/worker/workflow_sandbox/testmodules/proto",
   "temporalio/bridge/worker.py",
   "temporalio/contrib/opentelemetry.py",
+  "temporalio/contrib/pydantic.py",
   "temporalio/converter.py",
   "temporalio/testing/_workflow.py",
   "temporalio/worker/_activity.py",
@@ -168,6 +172,10 @@ exclude = [
   "tests/api/test_grpc_stub.py",
   "tests/conftest.py",
   "tests/contrib/test_opentelemetry.py",
+  "tests/contrib/pydantic/models.py",
+  "tests/contrib/pydantic/models_2.py",
+  "tests/contrib/pydantic/test_pydantic.py",
+  "tests/contrib/pydantic/workflows.py",
   "tests/test_converter.py",
   "tests/test_service.py",
   "tests/test_workflow.py",
@@ -186,6 +194,9 @@ exclude = [
 [tool.ruff]
 target-version = "py39"
 
+[tool.ruff.lint]
+extend-ignore = ["E741"]  # Allow single-letter variable names like I, O
+
 [build-system]
 requires = ["maturin>=1.0,<2.0"]
 build-backend = "maturin"
@@ -202,3 +213,8 @@ exclude = [
 [tool.uv]
 # Prevent uv commands from building the package by default
 package = false
+
+[tool.uv.sources]
+nexus-rpc = { path = "../nexus-sdk-python", editable = true }
+temporalio-xray = { path = "../xray/sdks/python", editable = true }
+hyperlinked = { path = "../../hyperlinked/python", editable = true }

temporalio/bridge/src/worker.rs

@@ -20,7 +20,7 @@ use temporal_sdk_core_api::worker::{
 };
 use temporal_sdk_core_api::Worker;
 use temporal_sdk_core_protos::coresdk::workflow_completion::WorkflowActivationCompletion;
-use temporal_sdk_core_protos::coresdk::{ActivityHeartbeat, ActivityTaskCompletion};
+use temporal_sdk_core_protos::coresdk::{ActivityHeartbeat, ActivityTaskCompletion, nexus::NexusTaskCompletion};
 use temporal_sdk_core_protos::temporal::api::history::v1::History;
 use tokio::sync::mpsc::{channel, Sender};
 use tokio_stream::wrappers::ReceiverStream;
@@ -570,6 +570,19 @@ impl WorkerRef {
         })
     }
 
+    fn poll_nexus_task<'p>(&self, py: Python<'p>) -> PyResult<&'p PyAny> {
+        let worker = self.worker.as_ref().unwrap().clone();
+        self.runtime.future_into_py(py, async move {
+            let bytes = match worker.poll_nexus_task().await {
+                Ok(task) => task.encode_to_vec(),
+                Err(PollError::ShutDown) => return Err(PollShutdownError::new_err(())),
+                Err(err) => return Err(PyRuntimeError::new_err(format!("Poll failure: {}", err))),
+            };
+            let bytes: &[u8] = &bytes;
+            Ok(Python::with_gil(|py| bytes.into_py(py)))
+        })
+    }
+
     fn complete_workflow_activation<'p>(
         &self,
         py: Python<'p>,
@@ -600,6 +613,19 @@ impl WorkerRef {
         })
     }
 
+    fn complete_nexus_task<'p>(&self, py: Python<'p>, proto: &PyBytes) -> PyResult<&'p PyAny> {
+        let worker = self.worker.as_ref().unwrap().clone();
+        let completion = NexusTaskCompletion::decode(proto.as_bytes())
+            .map_err(|err| PyValueError::new_err(format!("Invalid proto: {}", err)))?;
+        self.runtime.future_into_py(py, async move {
+            worker
+                .complete_nexus_task(completion)
+                .await
+                .context("Completion failure")
+                .map_err(Into::into)
+        })
+    }
+
     fn record_activity_heartbeat(&self, proto: &PyBytes) -> PyResult<()> {
         enter_sync!(self.runtime);
         let heartbeat = ActivityHeartbeat::decode(proto.as_bytes())

temporalio/bridge/worker.py

@@ -26,6 +26,7 @@
 import temporalio.bridge.client
 import temporalio.bridge.proto
 import temporalio.bridge.proto.activity_task
+import temporalio.bridge.proto.nexus
 import temporalio.bridge.proto.workflow_activation
 import temporalio.bridge.proto.workflow_completion
 import temporalio.bridge.runtime
@@ -35,7 +36,7 @@
 from temporalio.bridge.temporal_sdk_bridge import (
     CustomSlotSupplier as BridgeCustomSlotSupplier,
 )
-from temporalio.bridge.temporal_sdk_bridge import PollShutdownError
+from temporalio.bridge.temporal_sdk_bridge import PollShutdownError  # type: ignore
 
 
 @dataclass
@@ -216,6 +217,14 @@ async def poll_activity_task(
             await self._ref.poll_activity_task()
         )
 
+    async def poll_nexus_task(
+        self,
+    ) -> temporalio.bridge.proto.nexus.NexusTask:
+        """Poll for a nexus task."""
+        return temporalio.bridge.proto.nexus.NexusTask.FromString(
+            await self._ref.poll_nexus_task()
+        )
+
     async def complete_workflow_activation(
         self,
         comp: temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion,
@@ -229,6 +238,12 @@ async def complete_activity_task(
         """Complete an activity task."""
         await self._ref.complete_activity_task(comp.SerializeToString())
 
+    async def complete_nexus_task(
+        self, comp: temporalio.bridge.proto.nexus.NexusTaskCompletion
+    ) -> None:
+        """Complete a nexus task."""
+        await self._ref.complete_nexus_task(comp.SerializeToString())
+
     def record_activity_heartbeat(
         self, comp: temporalio.bridge.proto.ActivityHeartbeat
     ) -> None:

temporalio/client.py

@@ -459,8 +459,15 @@ async def start_workflow(
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
-        stack_level: int = 2,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
+        # The following options are deliberately not exposed in overloads
+        stack_level: int = 2,
+        nexus_completion_callbacks: Sequence[
+            temporalio.common.NexusCompletionCallback
+        ] = [],
+        workflow_event_links: Sequence[
+            temporalio.api.common.v1.Link.WorkflowEvent
+        ] = [],
     ) -> WorkflowHandle[Any, Any]:
         """Start a workflow and return its handle.
 
@@ -523,6 +530,11 @@ async def start_workflow(
             temporalio.workflow._Definition.get_name_and_result_type(workflow)
         )
 
+        for l in workflow_event_links:
+            print(
+                f"🌈@@ worker starting workflow with link: {google.protobuf.json_format.MessageToJson(l)}"
+            )
+
         return await self._impl.start_workflow(
             StartWorkflowInput(
                 workflow=name,
@@ -549,6 +561,8 @@ async def start_workflow(
                 rpc_timeout=rpc_timeout,
                 request_eager_start=request_eager_start,
                 priority=priority,
+                nexus_completion_callbacks=nexus_completion_callbacks,
+                workflow_event_links=workflow_event_links,
             )
         )
 
@@ -5156,6 +5170,8 @@ class StartWorkflowInput:
     rpc_timeout: Optional[timedelta]
     request_eager_start: bool
     priority: temporalio.common.Priority
+    nexus_completion_callbacks: Sequence[temporalio.common.NexusCompletionCallback]
+    workflow_event_links: Sequence[temporalio.api.common.v1.Link.WorkflowEvent]
 
 
 @dataclass
@@ -5770,6 +5786,16 @@ async def _build_start_workflow_execution_request(
         req = temporalio.api.workflowservice.v1.StartWorkflowExecutionRequest()
         req.request_eager_execution = input.request_eager_start
         await self._populate_start_workflow_execution_request(req, input)
+        for callback in input.nexus_completion_callbacks:
+            c = temporalio.api.common.v1.Callback()
+            c.nexus.url = callback.url
+            c.nexus.header.update(callback.header)
+            req.completion_callbacks.append(c)
+
+        req.links.extend(
+            temporalio.api.common.v1.Link(workflow_event=link)
+            for link in input.workflow_event_links
+        )
         return req
 
     async def _build_signal_with_start_workflow_execution_request(

temporalio/common.py

@@ -8,7 +8,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from datetime import datetime, timedelta
-from enum import Enum, IntEnum
+from enum import IntEnum
 from typing import (
     Any,
     Callable,
@@ -195,6 +195,37 @@ def __setstate__(self, state: object) -> None:
         )
 
 
+@dataclass(frozen=True)
+class NexusCompletionCallback:
+    """Nexus callback to attach to events such as workflow completion."""
+
+    url: str
+    """Callback URL."""
+
+    header: Mapping[str, str]
+    """Header to attach to callback request."""
+
+
+@dataclass(frozen=True)
+class WorkflowEventLink:
+    """A link to a history event that can be attached to a different history event."""
+
+    namespace: str
+    """Namespace of the workflow to link to."""
+
+    workflow_id: str
+    """ID of the workflow to link to."""
+
+    run_id: str
+    """Run ID of the workflow to link to."""
+
+    event_type: temporalio.api.enums.v1.EventType
+    """Type of the event to link to."""
+
+    event_id: int
+    """ID of the event to link to."""
+
+
 # We choose to make this a list instead of an sequence so we can catch if people
 # are not sending lists each time but maybe accidentally sending a string (which
 # is a sequence)

temporalio/converter.py

@@ -901,13 +901,57 @@ def _error_to_failure(
             failure.child_workflow_execution_failure_info.retry_state = (
                 temporalio.api.enums.v1.RetryState.ValueType(error.retry_state or 0)
             )
+        # TODO(dan): test coverage for this
+        elif isinstance(error, temporalio.exceptions.NexusOperationError):
+            failure.nexus_operation_execution_failure_info.SetInParent()
+            failure.nexus_operation_execution_failure_info.operation_token = (
+                error.operation_token
+            )
 
     def from_failure(
         self,
         failure: temporalio.api.failure.v1.Failure,
         payload_converter: PayloadConverter,
     ) -> BaseException:
         """See base class."""
+
+        # message Failure {
+        #     string message = 1;
+        #     // The source this Failure originated in, e.g. TypeScriptSDK / JavaSDK
+        #     // In some SDKs this is used to rehydrate the stack trace into an exception object.
+        #     string source = 2;
+        #     string stack_trace = 3;
+        #     // Alternative way to supply `message` and `stack_trace` and possibly other attributes, used for encryption of
+        #     // errors originating in user code which might contain sensitive information.
+        #     // The `encoded_attributes` Payload could represent any serializable object, e.g. JSON object or a `Failure` proto
+        #     // message.
+        #     //
+        #     // SDK authors:
+        #     // - The SDK should provide a default `encodeFailureAttributes` and `decodeFailureAttributes` implementation that:
+        #     //   - Uses a JSON object to represent `{ message, stack_trace }`.
+        #     //   - Overwrites the original message with "Encoded failure" to indicate that more information could be extracted.
+        #     //   - Overwrites the original stack_trace with an empty string.
+        #     //   - The resulting JSON object is converted to Payload using the default PayloadConverter and should be processed
+        #     //     by the user-provided PayloadCodec
+        #     //
+        #     // - If there's demand, we could allow overriding the default SDK implementation to encode other opaque Failure attributes.
+        #     // (-- api-linter: core::0203::optional=disabled --)
+        #     temporal.api.common.v1.Payload encoded_attributes = 20;
+        #     Failure cause = 4;
+        #     oneof failure_info {
+        #         ApplicationFailureInfo application_failure_info = 5;
+        #         TimeoutFailureInfo timeout_failure_info = 6;
+        #         CanceledFailureInfo canceled_failure_info = 7;
+        #         TerminatedFailureInfo terminated_failure_info = 8;
+        #         ServerFailureInfo server_failure_info = 9;
+        #         ResetWorkflowFailureInfo reset_workflow_failure_info = 10;
+        #         ActivityFailureInfo activity_failure_info = 11;
+        #         ChildWorkflowExecutionFailureInfo child_workflow_execution_failure_info = 12;
+        #         NexusOperationFailureInfo nexus_operation_execution_failure_info = 13;
+        #         NexusHandlerFailureInfo nexus_handler_failure_info = 14;
+        #     }
+        # }
+
         # If encoded attributes are present and have the fields we expect,
         # extract them
         if failure.HasField("encoded_attributes"):
@@ -978,6 +1022,15 @@ def from_failure(
                 else None,
             )
         elif failure.HasField("child_workflow_execution_failure_info"):
+            # message ChildWorkflowExecutionFailureInfo {
+            #     string namespace = 1;
+            #     temporal.api.common.v1.WorkflowExecution workflow_execution = 2;
+            #     temporal.api.common.v1.WorkflowType workflow_type = 3;
+            #     int64 initiated_event_id = 4;
+            #     int64 started_event_id = 5;
+            #     temporal.api.enums.v1.RetryState retry_state = 6;
+            # }
+
             child_info = failure.child_workflow_execution_failure_info
             err = temporalio.exceptions.ChildWorkflowError(
                 failure.message or "Child workflow error",
@@ -993,6 +1046,53 @@ def from_failure(
                 if child_info.retry_state
                 else None,
             )
+        elif failure.HasField("nexus_handler_failure_info"):
+            # message NexusHandlerFailureInfo {
+            #     // The Nexus error type as defined in the spec:
+            #     // https://github.com/nexus-rpc/api/blob/main/SPEC.md#predefined-handler-errors.
+            #     string type = 1;
+            #     // Retry behavior, defaults to the retry behavior of the error type as defined in the spec.
+            #     temporal.api.enums.v1.NexusHandlerErrorRetryBehavior retry_behavior = 2;
+            # }
+            # TODO(dan): check that handler-side phenomena are correctly turning into this vs nexus_operation_execution_failure_info
+            # TODO(dan): What exception should be raised for this vs nexus_operation_execution_failure_info?
+            nexus_handler_failure_info = failure.nexus_handler_failure_info
+            err = temporalio.exceptions.NexusHandlerError(
+                failure.message or "Nexus handler error",
+                type=nexus_handler_failure_info.type,
+                retryable={
+                    temporalio.api.enums.v1.NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE: True,
+                    temporalio.api.enums.v1.NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE: False,
+                }.get(nexus_handler_failure_info.retry_behavior),
+            )
+        elif failure.HasField("nexus_operation_execution_failure_info"):
+            # message NexusOperationFailureInfo {
+            #     // The NexusOperationScheduled event ID.
+            #     int64 scheduled_event_id = 1;
+            #     // Endpoint name.
+            #     string endpoint = 2;
+            #     // Service name.
+            #     string service = 3;
+            #     // Operation name.
+            #     string operation = 4;
+            #     // Operation ID - may be empty if the operation completed synchronously.
+            #     //
+            #     // Deprecated: Renamed to operation_token.
+            #     string operation_id = 5;
+            #     // Operation token - may be empty if the operation completed synchronously.
+            #     string operation_token = 6;
+            # }
+            # TODO(dan)
+            # This is covered by cancellation tests
+            nexus_op_failure_info = failure.nexus_operation_execution_failure_info
+            err = temporalio.exceptions.NexusOperationError(
+                failure.message or "Nexus operation error",
+                scheduled_event_id=nexus_op_failure_info.scheduled_event_id,
+                endpoint=nexus_op_failure_info.endpoint,
+                service=nexus_op_failure_info.service,
+                operation=nexus_op_failure_info.operation,
+                operation_token=nexus_op_failure_info.operation_token,
+            )
         else:
             err = temporalio.exceptions.FailureError(failure.message or "Failure error")
         err._failure = failure

temporalio/exceptions.py

@@ -284,6 +284,15 @@ def retry_state(self) -> Optional[RetryState]:
 class ChildWorkflowError(FailureError):
     """Error raised on child workflow failure."""
 
+    # message ChildWorkflowExecutionFailureInfo {
+    #     string namespace = 1;
+    #     temporal.api.common.v1.WorkflowExecution workflow_execution = 2;
+    #     temporal.api.common.v1.WorkflowType workflow_type = 3;
+    #     int64 initiated_event_id = 4;
+    #     int64 started_event_id = 5;
+    #     temporal.api.enums.v1.RetryState retry_state = 6;
+    # }
+
     def __init__(
         self,
         message: str,
@@ -342,6 +351,94 @@ def retry_state(self) -> Optional[RetryState]:
         return self._retry_state
 
 
+class NexusHandlerError(FailureError):
+    """Error raised on Nexus handler failure."""
+
+    # message NexusHandlerFailureInfo {
+    #     // The Nexus error type as defined in the spec:
+    #     // https://github.com/nexus-rpc/api/blob/main/SPEC.md#predefined-handler-errors.
+    #     string type = 1;
+    #     // Retry behavior, defaults to the retry behavior of the error type as defined in the spec.
+    #     temporal.api.enums.v1.NexusHandlerErrorRetryBehavior retry_behavior = 2;
+    # }
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        type: str,
+        retryable: Optional[bool] = None,
+    ):
+        """Initialize a Nexus handler error."""
+        super().__init__(message)
+        self._type = type
+        self._retryable = retryable
+
+
+class NexusOperationError(FailureError):
+    """Error raised on Nexus operation failure."""
+
+    # message NexusOperationFailureInfo {
+    #     // The NexusOperationScheduled event ID.
+    #     int64 scheduled_event_id = 1;
+    #     // Endpoint name.
+    #     string endpoint = 2;
+    #     // Service name.
+    #     string service = 3;
+    #     // Operation name.
+    #     string operation = 4;
+    #     // Operation ID - may be empty if the operation completed synchronously.
+    #     //
+    #     // Deprecated: Renamed to operation_token.
+    #     string operation_id = 5;
+    #     // Operation token - may be empty if the operation completed synchronously.
+    #     string operation_token = 6;
+    # }
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        scheduled_event_id: int,
+        endpoint: str,
+        service: str,
+        operation: str,
+        operation_token: str,
+    ):
+        """Initialize a Nexus operation error."""
+        super().__init__(message)
+        self._scheduled_event_id = scheduled_event_id
+        self._endpoint = endpoint
+        self._service = service
+        self._operation = operation
+        self._operation_token = operation_token
+
+    @property
+    def scheduled_event_id(self) -> int:
+        """The NexusOperationScheduled event ID for the failed operation."""
+        return self._scheduled_event_id
+
+    @property
+    def endpoint(self) -> str:
+        """The endpoint name for the failed operation."""
+        return self._endpoint
+
+    @property
+    def service(self) -> str:
+        """The service name for the failed operation."""
+        return self._service
+
+    @property
+    def operation(self) -> str:
+        """The name of the failed operation."""
+        return self._operation
+
+    @property
+    def operation_token(self) -> str:
+        """The operation token returned by the failed operation."""
+        return self._operation_token
+
+
 def is_cancelled_exception(exception: BaseException) -> bool:
     """Check whether the given exception is considered a cancellation exception
     according to Temporal.

temporalio/nexus/__init__.py

@@ -0,0 +1,12 @@
+import logging
+from collections.abc import Mapping
+from typing import Any, Optional
+
+
+class LoggerAdapter(logging.LoggerAdapter):
+    def __init__(self, logger: logging.Logger, extra: Optional[Mapping[str, Any]]):
+        super().__init__(logger, extra or {})
+
+
+logger = LoggerAdapter(logging.getLogger(__name__), None)
+"""Logger that has additional details regarding the current Nexus operation."""

temporalio/nexus/handler.py

@@ -3,22 +3,26 @@
 from __future__ import annotations
 
 import concurrent.futures
-from dataclasses import dataclass
+from collections.abc import Callable, Mapping, MutableMapping
+from dataclasses import dataclass, field
 from datetime import timedelta
 from typing import (
     Any,
     Awaitable,
-    Callable,
+    Generic,
     List,
-    Mapping,
-    MutableMapping,
     NoReturn,
     Optional,
     Sequence,
     Type,
+    TypeVar,
     Union,
 )
 
+import nexusrpc
+import nexusrpc.handler
+import nexusrpc.interface
+
 import temporalio.activity
 import temporalio.api.common.v1
 import temporalio.common
@@ -285,6 +289,60 @@ class StartChildWorkflowInput:
     ret_type: Optional[Type]
 
 
+# TODO(dan): Put these in a better location. Type variance?
+I = TypeVar("I")
+O = TypeVar("O")
+
+
+@dataclass
+class StartNexusOperationInput(Generic[I, O]):
+    """Input for :py:meth:`WorkflowOutboundInterceptor.start_nexus_operation`."""
+
+    endpoint: str
+    service: str
+    operation: Union[
+        nexusrpc.interface.Operation[I, O],
+        Callable[[Any], nexusrpc.handler.Operation[I, O]],
+        str,
+    ]
+    input: I
+    schedule_to_close_timeout: Optional[timedelta]
+    headers: Optional[Mapping[str, str]]
+    output_type: Optional[Type[O]] = None
+
+    _operation_name: str = field(init=False, repr=False)
+    _input_type: Optional[Type[I]] = field(init=False, repr=False)
+
+    def __post_init__(self) -> None:
+        if isinstance(self.operation, str):
+            self._operation_name = self.operation
+            self._input_type = None
+        elif isinstance(self.operation, nexusrpc.interface.Operation):
+            self._operation_name = self.operation.name
+            self._input_type = self.operation.input_type
+            self.output_type = self.operation.output_type
+        elif isinstance(self.operation, Callable):
+            defn = getattr(self.operation, "__nexus_operation__", None)
+            if isinstance(defn, nexusrpc.handler.NexusOperationDefinition):
+                self._operation_name = defn.name
+                self._input_type = defn.input_type
+                self.output_type = defn.output_type
+            else:
+                raise ValueError(
+                    f"Operation callable is not a Nexus operation: {self.operation}"
+                )
+        else:
+            raise ValueError(f"Operation is not a Nexus operation: {self.operation}")
+
+    @property
+    def operation_name(self) -> str:
+        return self._operation_name
+
+    @property
+    def input_type(self) -> Optional[Type[I]]:
+        return self._input_type
+
+
 @dataclass
 class StartLocalActivityInput:
     """Input for :py:meth:`WorkflowOutboundInterceptor.start_local_activity`."""
@@ -409,3 +467,9 @@ def start_local_activity(
         and :py:func:`temporalio.workflow.execute_local_activity` call.
         """
         return self.next.start_local_activity(input)
+
+    async def start_nexus_operation(
+        self, input: StartNexusOperationInput
+    ) -> temporalio.workflow.NexusOperationHandle[Any]:
+        """Called for every :py:func:`temporalio.workflow.start_nexus_operation` call."""
+        return await self.next.start_nexus_operation(input)

temporalio/worker/_activity.py

@@ -41,6 +41,7 @@
 
 from ._activity import SharedStateManager, _ActivityWorker
 from ._interceptor import Interceptor
+from ._nexus import _NexusWorker
 from ._tuning import WorkerTuner
 from ._workflow import _WorkflowWorker
 from ._workflow_instance import UnsandboxedWorkflowRunner, WorkflowRunner
@@ -106,6 +107,7 @@ def __init__(
         *,
         task_queue: str,
         activities: Sequence[Callable] = [],
+        nexus_services: Sequence[Any] = [],
         workflows: Sequence[Type] = [],
         activity_executor: Optional[concurrent.futures.Executor] = None,
         workflow_task_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None,
@@ -144,6 +146,8 @@ def __init__(
             maximum=5
         ),
     ) -> None:
+        # TODO(dan): consider not allowing max_workers < max_concurrent_nexus_operations?
+        # TODO(dan): Nexus tuner support?
         """Create a worker to process workflows and/or activities.
 
         Args:
@@ -153,10 +157,12 @@ def __init__(
                 client's underlying service client. This client cannot be
                 "lazy".
             task_queue: Required task queue for this worker.
-            activities: Set of activity callables decorated with
+            activities: Activity callables decorated with
                 :py:func:`@activity.defn<temporalio.activity.defn>`. Activities
                 may be async functions or non-async functions.
-            workflows: Set of workflow classes decorated with
+            nexus_services: Nexus service instances decorated with
+                :py:func:`@nexusrpc.handler.service<nexusrpc.handler.service>`.
+            workflows: Workflow classes decorated with
                 :py:func:`@workflow.defn<temporalio.workflow.defn>`.
             activity_executor: Concurrent executor to use for non-async
                 activities. This is required if any activities are non-async.
@@ -167,6 +173,14 @@ def __init__(
                 executor should at least be ``max_concurrent_activities`` or a
                 warning is issued. Note, a broken-executor failure from this
                 executor will cause the worker to fail and shutdown.
+            nexus_operation_executor: Concurrent executor to use for non-async
+                Nexus operations. This is required if any operation start methods
+                are non-async. :py:class:`concurrent.futures.ThreadPoolExecutor`
+                is recommended. If this is a
+                :py:class:`concurrent.futures.ProcessPoolExecutor`, all non-async
+                start methods must be picklable. ``max_workers`` on the executor
+                should at least be ``max_concurrent_nexus_operations`` or a warning
+                is issued.
             workflow_task_executor: Thread pool executor for workflow tasks. If
                 this is not present, a new
                 :py:class:`concurrent.futures.ThreadPoolExecutor` will be
@@ -191,12 +205,14 @@ def __init__(
                 identity is used.
             max_cached_workflows: If nonzero, workflows will be cached and
                 sticky task queues will be used.
-            max_concurrent_workflow_tasks: Maximum allowed number of workflow
-                tasks that will ever be given to this worker at one time. Mutually exclusive with ``tuner``.
             max_concurrent_activities: Maximum number of activity tasks that
-                will ever be given to this worker concurrently. Mutually exclusive with ``tuner``.
+                will ever be given to the activity worker concurrently. Mutually exclusive with ``tuner``.
             max_concurrent_local_activities: Maximum number of local activity
-                tasks that will ever be given to this worker concurrently. Mutually exclusive with ``tuner``.
+                tasks that will ever be given to the activity worker concurrently. Mutually exclusive with ``tuner``.
+            max_concurrent_nexus_operations: Maximum number of Nexus operations that
+                will ever be given to the Nexus worker concurrently. Mutually exclusive with ``tuner``.
+            max_concurrent_workflow_tasks: Maximum allowed number of
+                tasks that will ever be given to the workflow worker at one time. Mutually exclusive with ``tuner``.
             tuner:  Provide a custom :py:class:`WorkerTuner`. Mutually exclusive with the
                 ``max_concurrent_workflow_tasks``, ``max_concurrent_activities``, and
                 ``max_concurrent_local_activities`` arguments.
@@ -392,6 +408,17 @@ def __init__(
                 interceptors=interceptors,
                 metric_meter=self._runtime.metric_meter,
             )
+        self._nexus_worker: Optional[_NexusWorker] = None
+        if nexus_services:
+            self._nexus_worker = _NexusWorker(
+                bridge_worker=lambda: self._bridge_worker,
+                client=client,
+                task_queue=task_queue,
+                nexus_services=nexus_services,
+                data_converter=client_config["data_converter"],
+                interceptors=interceptors,
+                metric_meter=self._runtime.metric_meter,
+            )
         self._workflow_worker: Optional[_WorkflowWorker] = None
         if workflows:
             should_enforce_versioning_behavior = (
@@ -606,21 +633,30 @@ async def raise_on_shutdown():
             except asyncio.CancelledError:
                 pass
 
-        tasks: List[asyncio.Task] = [asyncio.create_task(raise_on_shutdown())]
+        tasks: dict[
+            Union[None, _ActivityWorker, _WorkflowWorker, _NexusWorker], asyncio.Task
+        ] = {None: asyncio.create_task(raise_on_shutdown())}
         # Create tasks for workers
         if self._activity_worker:
-            tasks.append(asyncio.create_task(self._activity_worker.run()))
+            tasks[self._activity_worker] = asyncio.create_task(
+                self._activity_worker.run()
+            )
         if self._workflow_worker:
-            tasks.append(asyncio.create_task(self._workflow_worker.run()))
+            tasks[self._workflow_worker] = asyncio.create_task(
+                self._workflow_worker.run()
+            )
+        if self._nexus_worker:
+            tasks[self._nexus_worker] = asyncio.create_task(self._nexus_worker.run())
 
         # Wait for either worker or shutdown requested
-        wait_task = asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
+        wait_task = asyncio.wait(tasks.values(), return_when=asyncio.FIRST_EXCEPTION)
         try:
             await asyncio.shield(wait_task)
 
-            # If any of the last two tasks failed, we want to re-raise that as
-            # the exception
-            exception = next((t.exception() for t in tasks[1:] if t.done()), None)
+            # If any of the worker tasks failed, re-raise that as the exception
+            exception = next(
+                (t.exception() for w, t in tasks.items() if w and t.done()), None
+            )
             if exception:
                 logger.error("Worker failed, shutting down", exc_info=exception)
                 if self._config["on_fatal_error"]:
@@ -635,7 +671,7 @@ async def raise_on_shutdown():
             exception = user_cancel_err
 
         # Cancel the shutdown task (safe if already done)
-        tasks[0].cancel()
+        tasks[None].cancel()
         graceful_timeout = self._config["graceful_shutdown_timeout"]
         logger.info(
             f"Beginning worker shutdown, will wait {graceful_timeout} before cancelling activities"
@@ -644,18 +680,10 @@ async def raise_on_shutdown():
         # Initiate core worker shutdown
         self._bridge_worker.initiate_shutdown()
 
-        # If any worker task had an exception, replace that task with a queue
-        # drain (task at index 1 can be activity or workflow worker, task at
-        # index 2 must be workflow worker if present)
-        if tasks[1].done() and tasks[1].exception():
-            if self._activity_worker:
-                tasks[1] = asyncio.create_task(self._activity_worker.drain_poll_queue())
-            else:
-                assert self._workflow_worker
-                tasks[1] = asyncio.create_task(self._workflow_worker.drain_poll_queue())
-        if len(tasks) > 2 and tasks[2].done() and tasks[2].exception():
-            assert self._workflow_worker
-            tasks[2] = asyncio.create_task(self._workflow_worker.drain_poll_queue())
+        # If any worker task had an exception, replace that task with a queue drain
+        for worker, task in tasks.items():
+            if worker and task.done() and task.exception():
+                tasks[worker] = asyncio.create_task(worker.drain_poll_queue())
 
         # Notify shutdown occurring
         if self._activity_worker:
@@ -664,20 +692,23 @@ async def raise_on_shutdown():
             self._workflow_worker.notify_shutdown()
 
         # Wait for all tasks to complete (i.e. for poller loops to stop)
-        await asyncio.wait(tasks)
+        await asyncio.wait(tasks.values())
         # Sometimes both workers throw an exception and since we only take the
         # first, Python may complain with "Task exception was never retrieved"
         # if we don't get the others. Therefore we call cancel on each task
         # which suppresses this.
-        for task in tasks:
+        for task in tasks.values():
             task.cancel()
 
-        # If there's an activity worker, we have to let all activity completions
-        # finish. We cannot guarantee that because poll shutdown completed
-        # (which means activities completed) that they got flushed to the
-        # server.
+        # Let all activity / nexus operations completions finish. We cannot guarantee that
+        # because poll shutdown completed (which means activities/operations completed)
+        # that they got flushed to the server.
         if self._activity_worker:
             await self._activity_worker.wait_all_completed()
+        if self._nexus_worker:
+            await self._nexus_worker.wait_all_completed()
+
+        # TODO(dan): check that we do all appropriate things for nexus worker that we do for activity worker
 
         # Do final shutdown
         try:

temporalio/worker/_interceptor.py

@@ -104,7 +104,7 @@ def __init__(
             if interceptor_class:
                 self._interceptor_classes.append(interceptor_class)
         self._extern_functions.update(
-            **_WorkflowExternFunctions(__temporal_get_metric_meter=lambda: metric_meter)
+            **_WorkflowExternFunctions(__temporal_get_metric_meter=lambda: metric_meter)  # pyright: ignore
         )
         self._workflow_failure_exception_types = workflow_failure_exception_types
         self._running_workflows: Dict[str, _RunningWorkflow] = {}

temporalio/worker/_nexus.py

@@ -4,7 +4,6 @@
 
 import asyncio
 import contextvars
-import dataclasses
 import inspect
 import logging
 import threading
@@ -23,6 +22,7 @@
     Awaitable,
     Callable,
     Dict,
+    Generator,
     Generic,
     Iterable,
     Iterator,
@@ -40,6 +40,8 @@
     overload,
 )
 
+import nexusrpc
+import nexusrpc.handler
 from typing_extensions import (
     Concatenate,
     Literal,
@@ -832,6 +834,22 @@ def workflow_start_local_activity(
         activity_id: Optional[str],
     ) -> ActivityHandle[Any]: ...
 
+    @abstractmethod
+    async def workflow_start_nexus_operation(
+        self,
+        endpoint: str,
+        service: str,
+        operation: Union[
+            nexusrpc.interface.Operation[I, O],
+            Callable[[Any], nexusrpc.handler.Operation[I, O]],
+            str,
+        ],
+        input: Any,
+        output_type: Optional[Type[O]] = None,
+        schedule_to_close_timeout: Optional[timedelta] = None,
+        headers: Optional[Mapping[str, str]] = None,
+    ) -> NexusOperationHandle[Any]: ...
+
     @abstractmethod
     def workflow_time_ns(self) -> int: ...
 
@@ -4340,6 +4358,77 @@ async def execute_child_workflow(
     return await handle
 
 
+I = TypeVar("I")
+O = TypeVar("O")
+S = TypeVar("S")
+
+
+# TODO(dan): ABC?
+class NexusOperationHandle(Generic[O]):
+    def cancel(self) -> bool:
+        # TODO(dan): docstring
+        """
+        Call task.cancel() on the asyncio task that is backing this handle.
+
+        From asyncio docs:
+
+        Cancel the future and schedule callbacks.
+
+        If the future is already done or cancelled, return False. Otherwise, change the future's state to cancelled, schedule the callbacks and return True.
+        """
+        raise NotImplementedError
+
+    def __await__(self) -> Generator[Any, Any, O]:
+        raise NotImplementedError
+
+    # TODO(dan): check SDK-wide philosophy on @property vs nullary accessor methods.
+    @property
+    def operation_token(self) -> Optional[str]:
+        raise NotImplementedError
+
+
+async def start_nexus_operation(
+    endpoint: str,
+    service: str,
+    operation: Union[
+        nexusrpc.interface.Operation[I, O],
+        Callable[[Any], nexusrpc.handler.Operation[I, O]],
+        str,
+    ],
+    input: Any,
+    *,
+    output_type: Optional[Type[O]] = None,
+    schedule_to_close_timeout: Optional[timedelta] = None,
+    headers: Optional[Mapping[str, str]] = None,
+) -> NexusOperationHandle[Any]:
+    """Start a Nexus operation and return its handle.
+
+    Args:
+        endpoint: The Nexus endpoint.
+        service: The Nexus service.
+        operation: The Nexus operation.
+        input: The Nexus operation input.
+        output_type: The Nexus operation output type.
+        schedule_to_close_timeout: Timeout for the entire operation attempt.
+        headers: Headers to send with the Nexus HTTP request.
+
+    Returns:
+        A handle to the Nexus operation. The result can be obtained as
+        ```python
+        await handle.result()
+        ```
+    """
+    return await _Runtime.current().workflow_start_nexus_operation(
+        endpoint=endpoint,
+        service=service,
+        operation=operation,
+        input=input,
+        output_type=output_type,
+        schedule_to_close_timeout=schedule_to_close_timeout,
+        headers=headers,
+    )
+
+
 class ExternalWorkflowHandle(Generic[SelfType]):
     """Handle for interacting with an external workflow.
 
@@ -5042,3 +5131,88 @@ def _to_proto(self) -> temporalio.bridge.proto.common.VersioningIntent.ValueType
         elif self == VersioningIntent.DEFAULT:
             return temporalio.bridge.proto.common.VersioningIntent.DEFAULT
         return temporalio.bridge.proto.common.VersioningIntent.UNSPECIFIED
+
+
+# Nexus
+
+
+class NexusClient(Generic[S]):
+    def __init__(
+        self,
+        service: Union[
+            # TODO(dan): Type[S] is modeling the interface case as well the impl case, but
+            # the typevar S is used below only in the impl case. I think this is OK, but
+            # think about it again before deleting this TODO.
+            Type[S],
+            str,
+        ],
+        endpoint: str,
+        schedule_to_close_timeout: Optional[timedelta] = None,
+    ) -> None:
+        # If service is not a str, then it must be a service interface or implementation
+        # class.
+        if isinstance(service, str):
+            self._service_name = service
+        elif service_defn := getattr(service, "__nexus_service__", None):
+            self._service_name = service_defn.name
+        elif service_metadata := getattr(service, "__nexus_service_metadata__", None):
+            self._service_name = service_metadata.name
+        else:
+            raise ValueError(
+                f"`service` may be a name (str), or a class decorated with either "
+                f"@nexusrpc.handler.service or @nexusrpc.interface.service. "
+                f"Invalid service type: {type(service)}"
+            )
+        print(f"🌈 NexusClient using service name: {self._service_name}")
+        self._endpoint = endpoint
+        self._schedule_to_close_timeout = schedule_to_close_timeout
+
+    # TODO(dan): overloads: no-input, operation name, ret type
+    # TODO(dan): should it be an error to use a reference to a method on a class other than that supplied?
+    async def start_operation(
+        self,
+        operation: Union[
+            nexusrpc.interface.Operation[I, O],
+            Callable[[S], nexusrpc.handler.Operation[I, O]],
+            str,
+        ],
+        input: I,
+        *,
+        output_type: Optional[Type[O]] = None,
+        schedule_to_close_timeout: Optional[timedelta] = None,
+        headers: Optional[Mapping[str, str]] = None,
+    ) -> NexusOperationHandle[O]:
+        return await temporalio.workflow.start_nexus_operation(
+            endpoint=self._endpoint,
+            service=self._service_name,
+            operation=operation,
+            input=input,
+            output_type=output_type,
+            schedule_to_close_timeout=(
+                schedule_to_close_timeout or self._schedule_to_close_timeout
+            ),
+            headers=headers,
+        )
+
+    # TODO(dan): overloads: no-input, operation name, ret type
+    async def execute_operation(
+        self,
+        operation: Union[
+            nexusrpc.interface.Operation[I, O],
+            Callable[[S], nexusrpc.handler.Operation[I, O]],
+            str,
+        ],
+        input: I,
+        *,
+        output_type: Optional[Type[O]] = None,
+        schedule_to_close_timeout: Optional[timedelta] = None,
+        headers: Optional[Mapping[str, str]] = None,
+    ) -> O:
+        handle: NexusOperationHandle[O] = await self.start_operation(
+            operation,
+            input,
+            output_type=output_type,
+            schedule_to_close_timeout=schedule_to_close_timeout,
+            headers=headers,
+        )
+        return await handle