Merge branch 'main' into otel-refactor

Author: Sushisource

.github/workflows/build-binaries.yml

@@ -74,8 +74,3 @@ jobs:
         with:
           name: packages-${{ matrix.package-suffix }}
           path: dist
-
-      - name: Deliberately fail to prevent releasing nexus-rpc w/ GitHub link in pyproject.toml
-        run: |
-          echo "This is a deliberate failure to prevent releasing nexus-rpc with a GitHub link in pyproject.toml"
-          exit 1

README.md

@@ -4,6 +4,8 @@
 [![PyPI](https://img.shields.io/pypi/v/temporalio.svg?style=for-the-badge)](https://pypi.org/project/temporalio)
 [![MIT](https://img.shields.io/pypi/l/temporalio.svg?style=for-the-badge)](LICENSE)
 
+**📣 News: Integration between OpenAI Agents SDK and Temporal is now in public preview. [Learn more](temporalio/contrib/openai_agents/README.md).**
+
 [Temporal](https://temporal.io/) is a distributed, scalable, durable, and highly available orchestration engine used to
 execute asynchronous, long-running business logic in a scalable and resilient way.
 
@@ -94,7 +96,11 @@ 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)
+    - [Plugins](#plugins)
+      - [Client Plugins](#client-plugins)
+      - [Worker Plugins](#worker-plugins)
     - [Workflow Replay](#workflow-replay)
     - [Observability](#observability)
       - [Metrics](#metrics)
@@ -1256,6 +1262,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 +1317,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.** ⚠️
@@ -1416,6 +1487,192 @@ https://github.com/temporalio/samples-python/tree/nexus/hello_nexus).
     ```
 
 
+### Plugins
+
+Plugins provide a way to extend and customize the behavior of Temporal clients and workers through a chain of 
+responsibility pattern. They allow you to intercept and modify client creation, service connections, worker 
+configuration, and worker execution. Common customizations may include but are not limited to:
+
+1. DataConverter
+2. Activities
+3. Workflows
+4. Interceptors
+
+A single plugin class can implement both client and worker plugin interfaces to share common logic between both 
+contexts. When used with a client, it will automatically be propagated to any workers created with that client.
+
+#### Client Plugins
+
+Client plugins can intercept and modify client configuration and service connections. They are useful for adding 
+authentication, modifying connection parameters, or adding custom behavior during client creation.
+
+Here's an example of a client plugin that adds custom authentication:
+
+```python
+from temporalio.client import Plugin, ClientConfig
+import temporalio.service
+
+class AuthenticationPlugin(Plugin):
+    def __init__(self, api_key: str):
+        self.api_key = api_key
+        
+    def init_client_plugin(self, next: Plugin) -> None:
+      self.next_client_plugin = next
+
+    def configure_client(self, config: ClientConfig) -> ClientConfig:
+        # Modify client configuration
+        config["namespace"] = "my-secure-namespace"
+        return self.next_client_plugin.configure_client(config)
+
+    async def connect_service_client(
+        self, config: temporalio.service.ConnectConfig
+    ) -> temporalio.service.ServiceClient:
+        # Add authentication to the connection
+        config.api_key = self.api_key
+        return await self.next_client_plugin.connect_service_client(config)
+
+# Use the plugin when connecting
+client = await Client.connect(
+    "my-server.com:7233",
+    plugins=[AuthenticationPlugin("my-api-key")]
+)
+```
+
+#### Worker Plugins
+
+Worker plugins can modify worker configuration and intercept worker execution. They are useful for adding monitoring, 
+custom lifecycle management, or modifying worker settings. Worker plugins can also configure replay. 
+They should do this in the case that they modified the worker in a way which would also need to be present 
+for replay to function. For instance, changing the data converter or adding workflows. 
+
+Here's an example of a worker plugin that adds custom monitoring:
+
+```python
+import temporalio
+from contextlib import asynccontextmanager
+from typing import AsyncIterator
+from temporalio.worker import Plugin, WorkerConfig, ReplayerConfig, Worker, Replayer, WorkflowReplayResult
+import logging
+
+class MonitoringPlugin(Plugin):
+    def __init__(self):
+        self.logger = logging.getLogger(__name__)
+
+    def init_worker_plugin(self, next: Plugin) -> None:
+        self.next_worker_plugin = next
+        
+    def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
+        # Modify worker configuration
+        original_task_queue = config["task_queue"]
+        config["task_queue"] = f"monitored-{original_task_queue}"
+        self.logger.info(f"Worker created for task queue: {config['task_queue']}")
+        return self.next_worker_plugin.configure_worker(config)
+
+    async def run_worker(self, worker: Worker) -> None:
+        self.logger.info("Starting worker execution")
+        try:
+            await self.next_worker_plugin.run_worker(worker)
+        finally:
+            self.logger.info("Worker execution completed")
+      
+  def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
+      return self.next_worker_plugin.configure_replayer(config)
+  
+  @asynccontextmanager
+  async def run_replayer(
+      self,
+      replayer: Replayer,
+      histories: AsyncIterator[temporalio.client.WorkflowHistory],
+  ) -> AsyncIterator[AsyncIterator[WorkflowReplayResult]]:
+        self.logger.info("Starting replay execution")
+        try:
+          async with self.next_worker_plugin.run_replayer(replayer, histories) as results:
+              yield results
+        finally:
+            self.logger.info("Replay execution completed")
+
+# Use the plugin when creating a worker
+worker = Worker(
+    client,
+    task_queue="my-task-queue",
+    workflows=[MyWorkflow],
+    activities=[my_activity],
+    plugins=[MonitoringPlugin()]
+)
+```
+
+For plugins that need to work with both clients and workers, you can implement both interfaces in a single class:
+
+```python
+import temporalio
+from contextlib import AbstractAsyncContextManager
+from typing import AsyncIterator
+from temporalio.client import Plugin as ClientPlugin, ClientConfig
+from temporalio.worker import Plugin as WorkerPlugin, WorkerConfig, ReplayerConfig, Worker, Replayer, WorkflowReplayResult
+
+
+class UnifiedPlugin(ClientPlugin, WorkerPlugin):
+    def init_client_plugin(self, next: ClientPlugin) -> None:
+        self.next_client_plugin = next
+
+    def init_worker_plugin(self, next: WorkerPlugin) -> None:
+        self.next_worker_plugin = next
+  
+    def configure_client(self, config: ClientConfig) -> ClientConfig:
+        # Client-side customization
+        config["data_converter"] = pydantic_data_converter
+        return self.next_client_plugin.configure_client(config)
+  
+    async def connect_service_client(
+        self, config: temporalio.service.ConnectConfig
+    ) -> temporalio.service.ServiceClient:
+        # Add authentication to the connection
+        config.api_key = self.api_key
+        return await self.next_client_plugin.connect_service_client(config)
+        
+    def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
+        # Worker-side customization
+        return self.next_worker_plugin.configure_worker(config)
+  
+    async def run_worker(self, worker: Worker) -> None:
+        print("Starting unified worker")
+        await self.next_worker_plugin.run_worker(worker)
+        
+    def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
+        config["data_converter"] = pydantic_data_converter
+        return config
+    
+    async def run_replayer(
+        self,
+        replayer: Replayer,
+        histories: AsyncIterator[temporalio.client.WorkflowHistory],
+    ) -> AbstractAsyncContextManager[AsyncIterator[WorkflowReplayResult]]:
+        return self.next_worker_plugin.run_replayer(replayer, histories)
+              
+# Create client with the unified plugin
+client = await Client.connect(
+    "localhost:7233",
+    plugins=[UnifiedPlugin()]
+)
+
+# Worker will automatically inherit the plugin from the client
+worker = Worker(
+    client,
+    task_queue="my-task-queue",
+    workflows=[MyWorkflow],
+    activities=[my_activity]
+)
+```
+
+**Important Notes:**
+
+- Plugins are executed in reverse order (last plugin wraps the first), forming a chain of responsibility
+- Client plugins that also implement worker plugin interfaces are automatically propagated to workers
+- Avoid providing the same plugin to both client and worker to prevent double execution
+- Plugin methods should call the plugin provided during initialization to maintain the plugin chain
+- Each plugin's `name()` method returns a unique identifier for debugging purposes
+
+
 ### Workflow Replay
 
 Given a workflow's history, it can be replayed locally to check for things like non-determinism errors. For example,

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "temporalio"
-version = "1.14.1"
+version = "1.15.0"
 description = "Temporal.io Python SDK"
 authors = [{ name = "Temporal Technologies Inc", email = "[email protected]" }]
 requires-python = ">=3.9"
@@ -11,7 +11,7 @@ keywords = [
     "workflow",
 ]
 dependencies = [
-    "nexus-rpc>=1.1.0",
+    "nexus-rpc==1.1.0",
     "protobuf>=3.20,<6",
     "python-dateutil>=2.8.2,<3 ; python_version < '3.11'",
     "types-protobuf>=3.20",
@@ -26,7 +26,7 @@ opentelemetry = [
 ]
 pydantic = ["pydantic>=2.0.0,<3"]
 openai-agents = [
-    "openai-agents >= 0.1,<0.2",
+    "openai-agents >= 0.2.3,<0.3",
     "eval-type-backport>=0.2.2; python_version < '3.10'"
 ]
 
@@ -57,6 +57,7 @@ dev = [
     "pytest-cov>=6.1.1",
     "httpx>=0.28.1",
     "pytest-pretty>=1.3.0",
+    "openai-agents[litellm] >= 0.2.3,<0.3"
 ]
 
 [tool.poe.tasks]
@@ -165,6 +166,7 @@ reportAny = "none"
 reportCallInDefaultInitializer = "none"
 reportExplicitAny = "none"
 reportIgnoreCommentWithoutRule = "none"
+reportImplicitAbstractClass = "none"
 reportImplicitOverride = "none"
 reportImplicitStringConcatenation = "none"
 reportImportCycles = "none"
@@ -184,11 +186,6 @@ exclude = [
   "temporalio/bridge/proto",
   "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",
   "temporalio/worker/_replayer.py",
   "temporalio/worker/_worker.py",
   "temporalio/worker/workflow_sandbox/_importer.py",
@@ -203,9 +200,7 @@ exclude = [
   "tests/contrib/pydantic/workflows.py",
   "tests/test_converter.py",
   "tests/test_service.py",
-  "tests/test_workflow.py",
   "tests/worker/test_activity.py",
-  "tests/worker/test_workflow.py",
   "tests/worker/workflow_sandbox/test_importer.py",
   "tests/worker/workflow_sandbox/test_restrictions.py",
   # TODO: these pass locally but fail in CI with
@@ -236,6 +231,3 @@ exclude = [
 [tool.uv]
 # Prevent uv commands from building the package by default
 package = false
-
-[tool.uv.sources]
-nexus-rpc = { git = "https://github.com/nexus-rpc/sdk-python.git", rev = "35f574c711193a6e2560d3e6665732a5bb7ae92c" }