Add open ai agents sdk plugin to enable temporal support

Author: danielmillerp

src/agentex/lib/core/clients/temporal/temporal_client.py

@@ -71,11 +71,12 @@
 
 
 class TemporalClient:
-    def __init__(self, temporal_client: Client | None = None):
+    def __init__(self, temporal_client: Client | None = None, enable_openai_agents_plugin: bool = False):
         self._client: Client = temporal_client
+        self._enable_openai_agents_plugin = enable_openai_agents_plugin
 
     @classmethod
-    async def create(cls, temporal_address: str):
+    async def create(cls, temporal_address: str, enable_openai_agents_plugin: bool = False):
         if temporal_address in [
             "false",
             "False",
@@ -88,8 +89,11 @@ async def create(cls, temporal_address: str):
         ]:
             _client = None
         else:
-            _client = await get_temporal_client(temporal_address)
-        return cls(_client)
+            _client = await get_temporal_client(
+                temporal_address, 
+                enable_openai_agents_plugin=enable_openai_agents_plugin
+            )
+        return cls(_client, enable_openai_agents_plugin)
 
     async def setup(self, temporal_address: str):
         self._client = await self._get_temporal_client(
@@ -109,7 +113,10 @@ async def _get_temporal_client(self, temporal_address: str) -> Client:
         ]:
             return None
         else:
-            return await get_temporal_client(temporal_address)
+            return await get_temporal_client(
+                temporal_address, 
+                enable_openai_agents_plugin=self._enable_openai_agents_plugin
+            )
 
     async def start_workflow(
         self,

src/agentex/lib/core/clients/temporal/utils.py

@@ -1,6 +1,7 @@
 from temporalio.client import Client
 from temporalio.contrib.pydantic import pydantic_data_converter
 from temporalio.runtime import OpenTelemetryConfig, Runtime, TelemetryConfig
+from temporalio.contrib.openai_agents import OpenAIAgentsPlugin
 
 # class DateTimeJSONEncoder(AdvancedJSONEncoder):
 #     def default(self, o: Any) -> Any:
@@ -38,12 +39,34 @@
 # )
 
 
-async def get_temporal_client(temporal_address: str, metrics_url: str = None) -> Client:
+async def get_temporal_client(
+    temporal_address: str, 
+    metrics_url: str = None,
+    enable_openai_agents_plugin: bool = False
+) -> Client:
+    """
+    Create a Temporal client with optional OpenAI Agents plugin integration.
+    
+    Args:
+        temporal_address: Temporal server address
+        metrics_url: Optional metrics endpoint URL  
+        enable_openai_agents_plugin: Whether to enable OpenAI Agents plugin
+        
+    Returns:
+        Configured Temporal client
+    """
+    plugins = []
+    
+    # Add OpenAI Agents plugin if enabled
+    if enable_openai_agents_plugin:
+        plugins.append(OpenAIAgentsPlugin())
+    
     if not metrics_url:
         client = await Client.connect(
             target_host=temporal_address,
             # data_converter=custom_data_converter,
             data_converter=pydantic_data_converter,
+            plugins=plugins if plugins else None,
         )
     else:
         runtime = Runtime(telemetry=TelemetryConfig(metrics=OpenTelemetryConfig(url=metrics_url)))
@@ -52,5 +75,6 @@ async def get_temporal_client(temporal_address: str, metrics_url: str = None) ->
             # data_converter=custom_data_converter,
             data_converter=pydantic_data_converter,
             runtime=runtime,
+            plugins=plugins if plugins else None,
         )
     return client

src/agentex/lib/core/temporal/services/temporal_task_service.py

@@ -23,6 +23,28 @@ def __init__(
         self._temporal_client = temporal_client
         self._env_vars = env_vars
 
+    @classmethod
+    async def create(
+        cls,
+        env_vars: EnvironmentVariables,
+        enable_openai_agents_plugin: bool = False,
+    ):
+        """
+        Create a TemporalTaskService with optional OpenAI Agents plugin integration.
+        
+        Args:
+            env_vars: Environment variables configuration
+            enable_openai_agents_plugin: Whether to enable OpenAI Agents plugin
+            
+        Returns:
+            Configured TemporalTaskService instance
+        """
+        temporal_client = await TemporalClient.create(
+            temporal_address=env_vars.TEMPORAL_ADDRESS or "localhost:7233",
+            enable_openai_agents_plugin=enable_openai_agents_plugin,
+        )
+        return cls(temporal_client, env_vars)
+
     async def submit_task(self, agent: Agent, task: Task, params: dict[str, Any] | None) -> str:
         """
         Submit a task to the async runtime for execution.

src/agentex/lib/core/temporal/workers/worker.py

@@ -7,7 +7,7 @@
 from typing import Any, overload
 
 from aiohttp import web
-from temporalio.client import Client
+from temporalio.client import Client, Plugin as ClientPlugin
 from temporalio.converter import (
     AdvancedJSONEncoder,
     CompositePayloadConverter,
@@ -19,6 +19,7 @@
 )
 from temporalio.runtime import OpenTelemetryConfig, Runtime, TelemetryConfig
 from temporalio.worker import (
+    Plugin as WorkerPlugin,
     UnsandboxedWorkflowRunner,
     Worker,
 )
@@ -65,11 +66,25 @@ def __init__(self) -> None:
     payload_converter_class=DateTimePayloadConverter,
 )
 
+def _validate_plugins(plugins: list) -> None:
+    """Validate that all items in the plugins list are valid Temporal plugins."""
+    for i, plugin in enumerate(plugins):
+        if not isinstance(plugin, (ClientPlugin, WorkerPlugin)):
+            raise TypeError(
+                f"Plugin at index {i} must be an instance of temporalio.client.Plugin "
+                f"or temporalio.worker.Plugin, got {type(plugin).__name__}"
+            )
+
+
+
+async def get_temporal_client(temporal_address: str, metrics_url: str = None, plugins: list = []) -> Client:
+
+    if plugins != []: # We don't need to validate the plugins if they are empty
+        _validate_plugins(plugins)
 
-async def get_temporal_client(temporal_address: str, metrics_url: str = None) -> Client:
     if not metrics_url:
         client = await Client.connect(
-            target_host=temporal_address, data_converter=custom_data_converter
+            target_host=temporal_address, data_converter=custom_data_converter, plugins=plugins
         )
     else:
         runtime = Runtime(
@@ -90,6 +105,7 @@ def __init__(
         max_workers: int = 10,
         max_concurrent_activities: int = 10,
         health_check_port: int = 80,
+        plugins: list = [],
     ):
         self.task_queue = task_queue
         self.activity_handles = []
@@ -98,6 +114,7 @@ def __init__(
         self.health_check_server_running = False
         self.healthy = False
         self.health_check_port = health_check_port
+        self.plugins = plugins
 
     @overload
     async def run(
@@ -126,6 +143,7 @@ async def run(
         await self._register_agent()
         temporal_client = await get_temporal_client(
             temporal_address=os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
+            plugins=self.plugins,
         )
 
         # Enable debug mode if AgentEx debug is enabled (disables deadlock detection)