Merge branch 'main' into openai/rebuild

Author: tconley1428

README.md

@@ -94,6 +94,7 @@ informal introduction to the features and their implementation.
         - [Heartbeating and Cancellation](#heartbeating-and-cancellation)
         - [Worker Shutdown](#worker-shutdown)
       - [Testing](#testing-1)
+    - [Interceptors](#interceptors)
     - [Nexus](#nexus)
     - [Workflow Replay](#workflow-replay)
     - [Observability](#observability)
@@ -1256,6 +1257,7 @@ calls in the `temporalio.activity` package make use of it. Specifically:
 
 * `in_activity()` - Whether an activity context is present
 * `info()` - Returns the immutable info of the currently running activity
+* `client()` - Returns the Temporal client used by this worker. Only available in `async def` activities.
 * `heartbeat(*details)` - Record a heartbeat
 * `is_cancelled()` - Whether a cancellation has been requested on this activity
 * `wait_for_cancelled()` - `async` call to wait for cancellation request
@@ -1310,6 +1312,70 @@ affect calls activity code might make to functions on the `temporalio.activity`
 * `worker_shutdown()` can be invoked to simulate a worker shutdown during execution of the activity
 
 
+### Interceptors
+
+The behavior of the SDK can be customized in many useful ways by modifying inbound and outbound calls using
+interceptors. This is similar to the use of middleware in other frameworks.
+
+There are five categories of inbound and outbound calls that you can modify in this way:
+
+1. Outbound client calls, such as `start_workflow()`, `signal_workflow()`, `list_workflows()`, `update_schedule()`, etc.
+
+2. Inbound workflow calls: `execute_workflow()`, `handle_signal()`, `handle_update_handler()`, etc
+
+3. Outbound workflow calls: `start_activity()`, `start_child_workflow()`, `start_nexus_operation()`, etc
+
+4. Inbound call to execute an activity: `execute_activity()`
+
+5. Outbound activity calls: `info()` and `heartbeat()`
+
+
+To modify outbound client calls, define a class inheriting from
+[`client.Interceptor`](https://python.temporal.io/temporalio.client.Interceptor.html), and implement the method
+`intercept_client()` to return an instance of
+[`OutboundInterceptor`](https://python.temporal.io/temporalio.client.OutboundInterceptor.html) that implements the
+subset of outbound client calls that you wish to modify.
+
+Then, pass a list containing an instance of your `client.Interceptor` class as the
+`interceptors` argument of [`Client.connect()`](https://python.temporal.io/temporalio.client.Client.html#connect).
+
+The purpose of the interceptor framework is that the methods you implement on your interceptor classes can perform
+arbitrary side effects and/or arbitrary modifications to the data, before it is received by the SDK's "real"
+implementation. The `interceptors` list can contain multiple interceptors. In this case they form a chain: a method
+implemented on an interceptor instance in the list can perform side effects, and modify the data, before passing it on
+to the corresponding method on the next interceptor in the list. Your interceptor classes need not implement every
+method; the default implementation is always to pass the data on to the next method in the interceptor chain.
+
+The remaining four categories are worker calls. To modify these, define a class inheriting from
+[`worker.Interceptor`](https://python.temporal.io/temporalio.worker.Interceptor.html) and implement methods on that
+class to define the
+[`ActivityInboundInterceptor`](https://python.temporal.io/temporalio.worker.ActivityInboundInterceptor.html),
+[`ActivityOutboundInterceptor`](https://python.temporal.io/temporalio.worker.ActivityOutboundInterceptor.html),
+[`WorkflowInboundInterceptor`](https://python.temporal.io/temporalio.worker.WorkflowInboundInterceptor.html), and
+[`WorkflowOutboundInterceptor`](https://python.temporal.io/temporalio.worker.WorkflowOutboundInterceptor.html) classes
+that you wish to use to effect your modifications. Then, pass a list containing an instance of your `worker.Interceptor`
+class as the `interceptors` argument of the [`Worker()`](https://python.temporal.io/temporalio.worker.Worker.html)
+constructor.
+
+It often happens that your worker and client interceptors will share code because they implement closely related logic.
+For convenience, you can create an interceptor class that inherits from _both_ `client.Interceptor` and
+`worker.Interceptor` (their method sets do not overlap). You can then pass this in the `interceptors` argument of
+`Client.connect()` when starting your worker _as well as_ in your client/starter code. If you do this, your worker will
+automatically pick up the interceptors from its underlying client (and you should not pass them directly to the
+`Worker()` constructor).
+
+This is best explained by example. The [Context Propagation Interceptor
+Sample](https://github.com/temporalio/samples-python/tree/main/context_propagation) is a good starting point. In
+[context_propagation/interceptor.py](https://github.com/temporalio/samples-python/blob/main/context_propagation/interceptor.py)
+a class is defined that inherits from both `client.Interceptor` and `worker.Interceptor`. It implements the various
+methods such that the outbound client and workflow calls set a certain key in the outbound `headers` field, and the
+inbound workflow and activity calls retrieve the header value from the inbound workflow/activity input data. An instance
+of this interceptor class is passed to `Client.connect()` when [starting the
+worker](https://github.com/temporalio/samples-python/blob/main/context_propagation/worker.py) and when connecting the
+client in the [workflow starter
+code](https://github.com/temporalio/samples-python/blob/main/context_propagation/starter.py).
+
+
 ### Nexus
 
 ⚠️  **Nexus support is currently at an experimental release stage. Backwards-incompatible changes are anticipated until a stable release is announced.** ⚠️

temporalio/activity.py

@@ -19,6 +19,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from typing import (
+    TYPE_CHECKING,
     Any,
     Callable,
     Iterator,
@@ -42,6 +43,9 @@
 
 from .types import CallableType
 
+if TYPE_CHECKING:
+    from temporalio.client import Client
+
 
 @overload
 def defn(fn: CallableType) -> CallableType: ...
@@ -179,6 +183,7 @@ class _Context:
         temporalio.converter.PayloadConverter,
     ]
     runtime_metric_meter: Optional[temporalio.common.MetricMeter]
+    client: Optional[Client]
     cancellation_details: _ActivityCancellationDetailsHolder
     _logger_details: Optional[Mapping[str, Any]] = None
     _payload_converter: Optional[temporalio.converter.PayloadConverter] = None
@@ -271,13 +276,37 @@ def wait_sync(self, timeout: Optional[float] = None) -> None:
         self.thread_event.wait(timeout)
 
 
+def client() -> Client:
+    """Return a Temporal Client for use in the current activity.
+
+    The client is only available in `async def` activities.
+
+    In tests it is not available automatically, but you can pass a client when creating a
+    :py:class:`temporalio.testing.ActivityEnvironment`.
+
+    Returns:
+        :py:class:`temporalio.client.Client` for use in the current activity.
+
+    Raises:
+        RuntimeError: When the client is not available.
+    """
+    client = _Context.current().client
+    if not client:
+        raise RuntimeError(
+            "No client available. The client is only available in `async def` "
+            "activities; not in `def` activities. In tests you can pass a "
+            "client when creating ActivityEnvironment."
+        )
+    return client
+
+
 def in_activity() -> bool:
     """Whether the current code is inside an activity.
 
     Returns:
         True if in an activity, False otherwise.
     """
-    return not _current_context.get(None) is None
+    return _current_context.get(None) is not None
 
 
 def info() -> Info:
@@ -574,8 +603,10 @@ def _apply_to_callable(
                 fn=fn,
                 # iscoroutinefunction does not return true for async __call__
                 # TODO(cretz): Why can't MyPy handle this?
-                is_async=inspect.iscoroutinefunction(fn)
-                or inspect.iscoroutinefunction(fn.__call__),  # type: ignore
+                is_async=(
+                    inspect.iscoroutinefunction(fn)
+                    or inspect.iscoroutinefunction(fn.__call__)  # type: ignore
+                ),
                 no_thread_cancel_exception=no_thread_cancel_exception,
             ),
         )

temporalio/client.py

@@ -936,9 +936,6 @@ async def execute_update_with_start_workflow(
         the call will not return successfully until the update has been delivered to a
         worker.
 
-        .. warning::
-           This API is experimental
-
         Args:
             update: Update function or name on the workflow. arg: Single argument to the
                 update.
@@ -1542,6 +1539,12 @@ def __init__(
         result_run_id: Optional[str] = None,
         first_execution_run_id: Optional[str] = None,
         result_type: Optional[Type] = None,
+        start_workflow_response: Optional[
+            Union[
+                temporalio.api.workflowservice.v1.StartWorkflowExecutionResponse,
+                temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse,
+            ]
+        ] = None,
     ) -> None:
         """Create workflow handle."""
         self._client = client
@@ -1550,6 +1553,7 @@ def __init__(
         self._result_run_id = result_run_id
         self._first_execution_run_id = first_execution_run_id
         self._result_type = result_type
+        self._start_workflow_response = start_workflow_response
         self.__temporal_eagerly_started = False
 
     @property
@@ -5347,11 +5351,7 @@ class StartWorkflowUpdateInput:
 
 @dataclass
 class UpdateWithStartUpdateWorkflowInput:
-    """Update input for :py:meth:`OutboundInterceptor.start_update_with_start_workflow`.
-
-    .. warning::
-       This API is experimental
-    """
+    """Update input for :py:meth:`OutboundInterceptor.start_update_with_start_workflow`."""
 
     update_id: Optional[str]
     update: str
@@ -5365,11 +5365,7 @@ class UpdateWithStartUpdateWorkflowInput:
 
 @dataclass
 class UpdateWithStartStartWorkflowInput:
-    """StartWorkflow input for :py:meth:`OutboundInterceptor.start_update_with_start_workflow`.
-
-    .. warning::
-       This API is experimental
-    """
+    """StartWorkflow input for :py:meth:`OutboundInterceptor.start_update_with_start_workflow`."""
 
     # Similar to StartWorkflowInput but without e.g. run_id, start_signal,
     # start_signal_args, request_eager_start.
@@ -5405,11 +5401,7 @@ class UpdateWithStartStartWorkflowInput:
 
 @dataclass
 class StartWorkflowUpdateWithStartInput:
-    """Input for :py:meth:`OutboundInterceptor.start_update_with_start_workflow`.
-
-    .. warning::
-       This API is experimental
-    """
+    """Input for :py:meth:`OutboundInterceptor.start_update_with_start_workflow`."""
 
     start_workflow_input: UpdateWithStartStartWorkflowInput
     update_workflow_input: UpdateWithStartUpdateWorkflowInput
@@ -5683,11 +5675,7 @@ async def start_workflow_update(
     async def start_update_with_start_workflow(
         self, input: StartWorkflowUpdateWithStartInput
     ) -> WorkflowUpdateHandle[Any]:
-        """Called for every :py:meth:`Client.start_update_with_start_workflow` and :py:meth:`Client.execute_update_with_start_workflow` call.
-
-        .. warning::
-            This API is experimental
-        """
+        """Called for every :py:meth:`Client.start_update_with_start_workflow` and :py:meth:`Client.execute_update_with_start_workflow` call."""
         return await self.next.start_update_with_start_workflow(input)
 
     ### Async activity calls
@@ -5832,6 +5820,7 @@ async def start_workflow(
             result_run_id=resp.run_id,
             first_execution_run_id=first_execution_run_id,
             result_type=input.ret_type,
+            start_workflow_response=resp,
         )
         setattr(handle, "__temporal_eagerly_started", eagerly_started)
         return handle

temporalio/contrib/openai_agents/workflow.py

@@ -1,18 +1,27 @@
 """Workflow-specific primitives for working with the OpenAI Agents SDK in a workflow context"""
 
+import functools
+import inspect
 import json
 from datetime import timedelta
-from typing import Any, Callable, Optional, Type
+from typing import Any, Callable, Optional, Type, Union, overload
 
 import nexusrpc
 from agents import (
+    Agent,
     RunContextWrapper,
     Tool,
 )
-from agents.function_schema import function_schema
+from agents.function_schema import DocstringStyle, function_schema
 from agents.tool import (
     FunctionTool,
+    ToolErrorFunction,
+    ToolFunction,
+    ToolParams,
+    default_tool_error_function,
+    function_tool,
 )
+from agents.util._types import MaybeAwaitable
 
 from temporalio import activity
 from temporalio import workflow as temporal_workflow
@@ -78,6 +87,25 @@ def activity_as_tool(
             "Bare function without tool and activity decorators is not supported",
             "invalid_tool",
         )
+    if ret.name is None:
+        raise ApplicationError(
+            "Input activity must have a name to be made into a tool",
+            "invalid_tool",
+        )
+    # If the provided callable has a first argument of `self`, partially apply it with the same metadata
+    # The actual instance will be picked up by the activity execution, the partially applied function will never actually be executed
+    params = list(inspect.signature(fn).parameters.keys())
+    if len(params) > 0 and params[0] == "self":
+        partial = functools.partial(fn, None)
+        setattr(partial, "__name__", fn.__name__)
+        partial.__annotations__ = getattr(fn, "__annotations__")
+        setattr(
+            partial,
+            "__temporal_activity_definition",
+            getattr(fn, "__temporal_activity_definition"),
+        )
+        partial.__doc__ = fn.__doc__
+        fn = partial
     schema = function_schema(fn)
 
     async def run_activity(ctx: RunContextWrapper[Any], input: str) -> Any:
@@ -94,9 +122,8 @@ async def run_activity(ctx: RunContextWrapper[Any], input: str) -> Any:
         # Add the context to the arguments if it takes that
         if schema.takes_context:
             args = [ctx] + args
-
         result = await temporal_workflow.execute_activity(
-            fn,
+            ret.name,  # type: ignore
             args=args,
             task_queue=task_queue,
             schedule_to_close_timeout=schedule_to_close_timeout,

temporalio/nexus/_link_conversion.py

@@ -23,7 +23,7 @@
 LINK_EVENT_TYPE_PARAM_NAME = "eventType"
 
 
-def workflow_handle_to_workflow_execution_started_event_link(
+def workflow_execution_started_event_link_from_workflow_handle(
     handle: temporalio.client.WorkflowHandle[Any, Any],
 ) -> temporalio.api.common.v1.Link.WorkflowEvent:
     """Create a WorkflowEvent link corresponding to a started workflow"""

temporalio/nexus/_operation_context.py

@@ -18,6 +18,7 @@
 from typing_extensions import Concatenate
 
 import temporalio.api.common.v1
+import temporalio.api.workflowservice.v1
 import temporalio.client
 import temporalio.common
 from temporalio.nexus import _link_conversion
@@ -142,25 +143,30 @@ def _get_workflow_event_links(
     def _add_outbound_links(
         self, workflow_handle: temporalio.client.WorkflowHandle[Any, Any]
     ):
+        # If links were not sent in StartWorkflowExecutionResponse then construct them.
+        wf_event_links: list[temporalio.api.common.v1.Link.WorkflowEvent] = []
         try:
-            link = _link_conversion.workflow_event_to_nexus_link(
-                _link_conversion.workflow_handle_to_workflow_execution_started_event_link(
-                    workflow_handle
-                )
+            if isinstance(
+                workflow_handle._start_workflow_response,
+                temporalio.api.workflowservice.v1.StartWorkflowExecutionResponse,
+            ):
+                if workflow_handle._start_workflow_response.HasField("link"):
+                    if link := workflow_handle._start_workflow_response.link:
+                        if link.HasField("workflow_event"):
+                            wf_event_links.append(link.workflow_event)
+            if not wf_event_links:
+                wf_event_links = [
+                    _link_conversion.workflow_execution_started_event_link_from_workflow_handle(
+                        workflow_handle
+                    )
+                ]
+            self.nexus_context.outbound_links.extend(
+                _link_conversion.workflow_event_to_nexus_link(link)
+                for link in wf_event_links
             )
         except Exception as e:
             logger.warning(
-                f"Failed to create WorkflowExecutionStarted event link for workflow {id}: {e}"
-            )
-        else:
-            self.nexus_context.outbound_links.append(
-                # TODO(nexus-prerelease): Before, WorkflowRunOperation was generating an EventReference
-                # link to send back to the caller. Now, it checks if the server returned
-                # the link in the StartWorkflowExecutionResponse, and if so, send the link
-                # from the response to the caller. Fallback to generating the link for
-                # backwards compatibility. PR reference in Go SDK:
-                # https://github.com/temporalio/sdk-go/pull/1934
-                link
+                f"Failed to create WorkflowExecutionStarted event links for workflow {workflow_handle}: {e}"
             )
         return workflow_handle