feat: introducing special activities due to temporal async constraints

Author: prassanna-ravishankar

examples/tutorials/10_agentic/10_temporal/040_openai_temporal_integration/README.md

@@ -1,6 +1,6 @@
-# Simplified OpenAI Agent Chat - Agent Platform Integration
+# OpenAI Temporal Integration Tutorial
 
-This tutorial demonstrates the new **Agent Platform Integration** for Agentex that dramatically simplifies agent development while preserving all Agentex infrastructure benefits.
+This tutorial demonstrates the **Agent Platform Integration** for Agentex that provides a streamlined approach to agent development while maintaining full Agentex infrastructure compatibility.
 
 ## Before vs After Comparison
 
@@ -13,31 +13,30 @@ This tutorial demonstrates the new **Agent Platform Integration** for Agentex th
 | **Error handling** | Manual try/catch and retry logic | Built-in recovery |
 | **ACP integration** | Manual message creation/sending | Automatic via bridge |
 
-## Key Benefits
+## Key Features
 
-### 🚀 **Dramatically Reduced Complexity**
-- **90% reduction in code** - from 277 lines to ~30 lines
-- **No manual orchestration** - agent execution is automatically durable
-- **No activity definitions** - tool calls are automatically temporal activities
+### **Reduced Complexity**
+- Simplified codebase: from 277 lines to ~30 lines
+- Automatic agent execution durability
+- Built-in tool call orchestration
 
-### 🔧 **Preserved Agentex Infrastructure**
-- **ACP protocol compatibility** - external clients unchanged
-- **Kubernetes deployment** - same Helm charts and configs
-- **Multi-tenant hosting** - same agent discovery and routing
-- **Authentication & monitoring** - same observability stack
+### **Infrastructure Compatibility**
+- Full ACP protocol compatibility
+- Existing deployment configurations work unchanged
+- Same authentication and monitoring systems
+- Multi-tenant hosting support maintained
 
-### 🎯 **Platform Agnostic Design**
-- **OpenAI Agents SDK** - this tutorial (implemented)
-- **LangChain** - future extension point
-- **CrewAI** - future extension point
-- **Custom frameworks** - extensible via strategy pattern
+### **Platform Extensibility**
+- OpenAI Agents SDK integration (implemented)
+- Extensible architecture for LangChain, CrewAI
+- Strategy pattern for custom frameworks
 
 ## Implementation Details
 
 ### Workflow Definition
 ```python
 @workflow.defn(name=environment_variables.WORKFLOW_NAME)
-class SimplifiedOpenAIChatAgent(OpenAIAgentWorkflow):
+class At040OpenAITemporalIntegration(OpenAIAgentWorkflow):
     async def create_agent(self) -> Agent:
         return Agent(
             name="Tool-Enabled Assistant",
@@ -51,41 +50,41 @@ class SimplifiedOpenAIChatAgent(OpenAIAgentWorkflow):
 ```python
 worker = AgentexWorker(
     task_queue=environment_variables.WORKFLOW_TASK_QUEUE,
-    agent_platform="openai",  # Automatic optimization
+    agent_platform="openai",  # Platform optimization
 )
-await worker.run(activities=[], workflow=SimplifiedOpenAIChatAgent)
+await worker.run(activities=[], workflow=At040OpenAITemporalIntegration)
 ```
 
-## Architecture Benefits
+## Technical Architecture
 
-### Automatic Durability
-- **Agent executions** become Temporal activities automatically
-- **Tool calls** are durable with automatic retries
-- **Conversation state** persists across workflow restarts
+### Durability Features
+- Agent executions are automatically temporal activities
+- Tool calls include built-in retry mechanisms
+- Conversation state persists across workflow restarts
 
-### Performance Optimizations  
-- **Activity exclusion** - OpenAI provider activities automatically excluded
-- **Direct SDK integration** - bypasses activity overhead for simple cases
-- **Platform-specific configuration** - optimized worker settings per platform
+### Performance Features
+- Automatic exclusion of unused provider activities
+- Direct SDK integration reduces overhead
+- Platform-specific worker configuration
 
-### Future Extensibility
-- **Strategy pattern** - easy to add new agent platforms
-- **Unified interface** - same workflow pattern across all platforms
-- **Agentex compatibility** - seamless integration with existing infrastructure
+### Extensibility
+- Strategy pattern for adding new agent platforms
+- Consistent workflow interface across platforms
+- Full compatibility with existing Agentex infrastructure
 
 ## Running the Tutorial
 
 1. **Set environment variables:**
    ```bash
-   export WORKFLOW_NAME="simplified-openai-chat"
-   export WORKFLOW_TASK_QUEUE="simplified_openai_chat_queue"
-   export AGENT_NAME="simplified-openai-chat"
+   export WORKFLOW_NAME="at040-openai-temporal-integration"
+   export WORKFLOW_TASK_QUEUE="040_openai_temporal_integration_queue"
+   export AGENT_NAME="at040-openai-temporal-integration"
    export OPENAI_API_KEY="your-openai-api-key"
    ```
 
-2. **Start the worker:**
+2. **Run the agent:**
    ```bash
-   python project/run_worker.py
+   uv run agentex agents run --manifest manifest.yaml
    ```
 
 3. **Test via ACP API:**
@@ -95,29 +94,29 @@ await worker.run(activities=[], workflow=SimplifiedOpenAIChatAgent)
      -d '{
        "method": "task/create",
        "params": {
-         "agent_name": "simplified-openai-chat"
+         "agent_name": "at040-openai-temporal-integration"
        }
      }'
    ```
 
-## Migration Guide
+## Migration from Manual Approach
 
-To migrate from the complex manual approach to this simplified approach:
+To migrate from the manual orchestration pattern (010_agent_chat):
 
-1. **Replace workflow inheritance:**
-   - From: `BaseWorkflow` 
-   - To: `OpenAIAgentWorkflow` (or other platform workflow)
+1. **Update workflow inheritance:**
+   - Change from: `BaseWorkflow` 
+   - Change to: `OpenAIAgentWorkflow`
 
-2. **Replace manual orchestration:**
-   - From: Manual `adk.providers.openai.run_agent_streamed_auto_send()`
-   - To: Simple `create_agent()` implementation
+2. **Replace orchestration code:**
+   - Remove: Manual `adk.providers.openai.run_agent_streamed_auto_send()` calls
+   - Add: `create_agent()` method implementation
 
 3. **Update worker configuration:**
-   - Add: `agent_platform="openai"` parameter
-   - Remove: Manual activity registration
+   - Add: `agent_platform="openai"` parameter to `AgentexWorker`
+   - Activities: Use empty list `[]` for automatic optimization
 
-4. **Remove manual activities:**
-   - Delete: Custom `@activity.defn` wrappers
-   - Keep: Core business logic in simple functions
+4. **Simplify activity management:**
+   - Remove: Custom `@activity.defn` wrapper functions
+   - Retain: Core business logic as regular functions
 
-This approach maintains 100% compatibility with existing Agentex infrastructure while dramatically simplifying development.
+This maintains full compatibility with existing Agentex infrastructure.

examples/tutorials/10_agentic/10_temporal/040_openai_temporal_integration/manifest.yaml

@@ -30,7 +30,7 @@ build:
 # Only used when running the agent locally
 local_development:
   agent:
-    port: 8000  # Port where your local ACP server is running
+    port: 18000  # Port where your local ACP server is running
     host_address: host.docker.internal  # Host address for Docker networking
 
   # File paths for local development (relative to this manifest.yaml)

examples/tutorials/10_agentic/10_temporal/040_openai_temporal_integration/project/workflow.py

@@ -19,6 +19,7 @@
 
 from agentex.lib.core.temporal.agent_platforms.workflow import OpenAIAgentWorkflow
 from agentex.lib.environment_variables import EnvironmentVariables
+from agentex.lib.types.acp import CreateTaskParams
 from agents import Agent
 
 environment_variables = EnvironmentVariables.refresh()
@@ -36,6 +37,11 @@ class At040OpenAITemporalIntegration(OpenAIAgentWorkflow):
     agent configuration approach.
     """
 
+    @workflow.run
+    async def on_task_create(self, params: CreateTaskParams) -> None:
+        """Task creation handler - delegates to platform base class"""
+        await super().on_task_create(params)
+    
     async def create_agent(self) -> Agent:
         """
         Define the OpenAI agent configuration.

src/agentex/lib/core/temporal/activities/__init__.py

@@ -212,6 +212,18 @@ def get_all_activities(sgp_client=None, exclude_platforms=None):
             openai_activities.run_agent_auto_send,
             openai_activities.run_agent_streamed_auto_send,
         ])
+    else:
+        # When excluding OpenAI provider activities, include agent platform activities
+        from agentex.lib.core.temporal.agent_platforms import (
+            openai_agent_execution,
+            langchain_agent_execution,
+            crewai_agent_execution,
+        )
+        activities.extend([
+            openai_agent_execution,
+            langchain_agent_execution,
+            crewai_agent_execution,
+        ])
 
     # SGP activities
     if sgp_client is not None:

src/agentex/lib/core/temporal/agent_platforms/__init__.py

@@ -4,14 +4,18 @@
 from .registry import AgentPlatformRegistry
 from .workflow import AgentPlatformWorkflow
 from .tools import AgentexToolAdapter, create_openai_tools_from_activities, create_search_tool
+from .activities import openai_agent_execution, langchain_agent_execution, crewai_agent_execution
 
 __all__ = [
     "ACPAgentBridge",
     "AgentExecutionStrategy", 
-    "OpenAIExecutionStrategy",
+    "OpenAIExecutionStrategy", 
     "AgentPlatformRegistry",
     "AgentPlatformWorkflow",
     "AgentexToolAdapter",
     "create_openai_tools_from_activities",
     "create_search_tool",
+    "openai_agent_execution",
+    "langchain_agent_execution", 
+    "crewai_agent_execution",
 ]

src/agentex/lib/core/temporal/agent_platforms/activities.py

@@ -0,0 +1,78 @@
+"""
+Agent platform activities - Temporal activities for platform-specific agent execution.
+
+These activities provide the proper async execution context for different agent
+frameworks that may not work directly within Temporal workflow constraints.
+"""
+
+from typing import Any, Dict
+
+from temporalio import activity
+from agents import Agent, Runner, ModelSettings
+
+from agentex.lib.utils.logging import make_logger
+
+logger = make_logger(__name__)
+
+@activity.defn(name="openai_agent_execution")
+async def openai_agent_execution(params: Dict[str, Any]) -> str:
+    """
+    Execute OpenAI agent in proper async activity context.
+    
+    OpenAI Agents SDK requires full async context including executor access
+    which isn't available in Temporal workflow execution environment.
+    
+    Args:
+        params: Dictionary with agent_config, user_input, task_id
+        
+    Returns:
+        str: Agent's response output
+    """
+    agent_config = params["agent_config"]
+    user_input = params["user_input"] 
+    task_id = params["task_id"]
+    
+    try:
+        # Handle serialized config - convert dicts back to proper types
+        cleaned_config = {}
+        for key, value in agent_config.items():
+            if key == "model_settings" and isinstance(value, dict):
+                # Convert dict back to ModelSettings instance
+                cleaned_config[key] = ModelSettings(**value)
+            else:
+                cleaned_config[key] = value
+        
+        # Create agent from cleaned config
+        agent = Agent(**cleaned_config)
+        
+        # Execute agent with user input in proper async context
+        logger.debug(f"Executing OpenAI agent for task {task_id} with input: {user_input[:100]}...")
+        result = await Runner.run(starting_agent=agent, input=user_input)
+        
+        # Extract final output
+        output = result.final_output if hasattr(result, 'final_output') else str(result)
+        
+        logger.debug(f"OpenAI agent execution completed for task {task_id}")
+        return output
+        
+    except Exception as e:
+        logger.error(f"OpenAI agent execution failed for task {task_id}: {e}")
+        raise
+
+
+@activity.defn(name="langchain_agent_execution") 
+async def langchain_agent_execution(params: Dict[str, Any]) -> str:
+    """
+    Execute LangChain agent in proper async activity context (future implementation).
+    """
+    # TODO: Implement when LangChain strategy is added
+    raise NotImplementedError("LangChain agent execution not yet implemented")
+
+
+@activity.defn(name="crewai_agent_execution")
+async def crewai_agent_execution(params: Dict[str, Any]) -> str:
+    """
+    Execute CrewAI agent in proper async activity context (future implementation).  
+    """
+    # TODO: Implement when CrewAI strategy is added
+    raise NotImplementedError("CrewAI agent execution not yet implemented")

src/agentex/lib/core/temporal/agent_platforms/strategies.py

@@ -74,8 +74,8 @@ class OpenAIExecutionStrategy(AgentExecutionStrategy):
     """OpenAI Agents SDK execution strategy"""
 
     def __init__(self):
-        from agentex.lib.core.services.adk.providers.openai import OpenAIService
-        self.openai_service = OpenAIService()
+        # No dependencies needed - we use OpenAI Agents SDK directly
+        pass
 
     @property
     def platform_name(self) -> str:
@@ -89,43 +89,34 @@ async def execute_agent(
         trace_id: str = None
     ) -> str:
         """
-        Execute OpenAI agent using direct Agent SDK integration.
+        Execute OpenAI agent using Temporal activity for proper async context.
         
-        This bypasses the existing run_agent_streamed_auto_send activity to use
-        the OpenAI Agents SDK directly for better performance and durability.
+        OpenAI Agents SDK requires full async context which isn't available in 
+        Temporal workflow execution. We need to run it as an activity.
         """
         try:
-            # Import OpenAI Agents SDK
-            from agents import Agent, Runner
-            
-            # Create agent from config
-            agent = Agent(**agent_config)
+            from temporalio import workflow
 
-            # Execute agent with user input
-            logger.debug(f"Executing OpenAI agent for task {task_id} with input: {user_input[:100]}...")
-            result = await Runner.run(starting_agent=agent, input=user_input)
+            # Execute OpenAI agent as a Temporal activity for proper async context
+            logger.debug(f"Executing OpenAI agent activity for task {task_id} with input: {user_input[:100]}...")
 
-            # Extract final output
-            output = result.final_output if hasattr(result, 'final_output') else str(result)
+            result = await workflow.execute_activity(
+                activity="openai_agent_execution",
+                arg={"agent_config": agent_config, "user_input": user_input, "task_id": task_id},
+                start_to_close_timeout=workflow.timedelta(minutes=5),
+            )
 
             logger.debug(f"OpenAI agent execution completed for task {task_id}")
-            return output
+            return result
 
-        except ImportError as e:
-            logger.error(f"OpenAI Agents SDK not available: {e}")
-            raise RuntimeError("OpenAI Agents SDK is required for OpenAI strategy") from e
         except Exception as e:
             logger.error(f"OpenAI agent execution failed for task {task_id}: {e}")
             raise
 
     async def create_agent_from_config(self, config: Dict[str, Any]) -> Any:
         """Create OpenAI Agent instance from configuration"""
-        try:
-            from agents import Agent
-            return Agent(**config)
-        except ImportError as e:
-            logger.error(f"OpenAI Agents SDK not available: {e}")
-            raise RuntimeError("OpenAI Agents SDK is required for OpenAI strategy") from e
+        from agents import Agent
+        return Agent(**config)
 
     def get_worker_config(self, platform_config: Dict[str, Any]) -> Dict[str, Any]:
         """