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

@@ -4,7 +4,7 @@
 import uuid
 from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
-from typing import Any, overload
+from typing import Any
 
 from aiohttp import web
 from temporalio.client import Client
@@ -22,6 +22,7 @@
     UnsandboxedWorkflowRunner,
     Worker,
 )
+from temporalio.contrib.openai_agents import OpenAIAgentsPlugin
 
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.utils.registration import register_agent
@@ -66,10 +67,18 @@ def __init__(self) -> None:
 )
 
 
-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:
+    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
+            target_host=temporal_address, 
+            data_converter=custom_data_converter,
+            plugins=plugins if plugins else None,
         )
     else:
         runtime = Runtime(
@@ -79,6 +88,7 @@ async def get_temporal_client(temporal_address: str, metrics_url: str = None) ->
             target_host=temporal_address,
             data_converter=custom_data_converter,
             runtime=runtime,
+            plugins=plugins if plugins else None,
         )
     return client
 
@@ -90,6 +100,7 @@ def __init__(
         max_workers: int = 10,
         max_concurrent_activities: int = 10,
         health_check_port: int = 80,
+        enable_openai_agents_plugin: bool = False,
     ):
         self.task_queue = task_queue
         self.activity_handles = []
@@ -98,49 +109,30 @@ def __init__(
         self.health_check_server_running = False
         self.healthy = False
         self.health_check_port = health_check_port
+        self.enable_openai_agents_plugin = enable_openai_agents_plugin
 
-    @overload
     async def run(
         self,
         activities: list[Callable],
-        *,
         workflow: type,
-    ) -> None: ...
-    
-    @overload
-    async def run(
-        self,
-        activities: list[Callable],
-        *,
-        workflows: list[type],
-    ) -> None: ...
-
-    async def run(
-        self,
-        activities: list[Callable],
-        *,
-        workflow: type | None = None,
-        workflows: list[type] | None = None,
     ):
         await self.start_health_check_server()
         await self._register_agent()
         temporal_client = await get_temporal_client(
             temporal_address=os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
+            enable_openai_agents_plugin=self.enable_openai_agents_plugin,
         )
 
         # Enable debug mode if AgentEx debug is enabled (disables deadlock detection)
         debug_enabled = os.environ.get("AGENTEX_DEBUG_ENABLED", "false").lower() == "true"
         if debug_enabled:
             logger.info("🐛 [WORKER] Temporal debug mode enabled - deadlock detection disabled")
 
-        if workflow is None and workflows is None:
-            raise ValueError("Either workflow or workflows must be provided")
-        
         worker = Worker(
             client=temporal_client,
             task_queue=self.task_queue,
             activity_executor=ThreadPoolExecutor(max_workers=self.max_workers),
-            workflows=[workflow] if workflows is None else workflows,
+            workflows=[workflow],
             activities=activities,
             workflow_runner=UnsandboxedWorkflowRunner(),
             max_concurrent_activities=self.max_concurrent_activities,