DO NOT MERGE: custom activity examples

Author: Charlotte Zhuang

examples/tutorials/10_agentic/10_temporal/010_agent_chat/project/custom_activities.py

@@ -0,0 +1,58 @@
+import asyncio
+from typing import Any, Dict, List
+
+from temporalio import activity
+
+from agentex.lib.utils.logging import make_logger
+
+logger = make_logger(__name__)
+
+
+@activity.defn
+async def process_special_data(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Custom activity for processing special data in the custom task queue."""
+    logger.info(f"Processing special data: {data}")
+    
+    # Simulate some processing
+    await asyncio.sleep(0.1)
+    
+    return {
+        "processed": True,
+        "original_data": data,
+        "result": f"Processed {len(data)} items"
+    }
+
+
+@activity.defn
+async def validate_custom_input(input_data: str) -> bool:
+    """Custom activity for validating input in the custom task queue."""
+    logger.info(f"Validating input: {input_data}")
+    
+    # Simple validation logic
+    is_valid = len(input_data) > 0 and not input_data.isspace()
+    
+    return is_valid
+
+
+@activity.defn
+async def transform_message(message: str, transform_type: str = "uppercase") -> str:
+    """Custom activity for transforming messages in various ways."""
+    logger.info(f"Transforming message with type: {transform_type}")
+    
+    if transform_type == "uppercase":
+        return message.upper()
+    elif transform_type == "lowercase":
+        return message.lower()
+    elif transform_type == "reverse":
+        return message[::-1]
+    else:
+        return message
+
+
+def get_custom_activities() -> List[Any]:
+    """Return all custom activities for the special task queue."""
+    return [
+        process_special_data,
+        validate_custom_input,
+        transform_message,
+    ]

examples/tutorials/10_agentic/10_temporal/010_agent_chat/project/run_custom_worker.py

@@ -0,0 +1,30 @@
+import asyncio
+
+from agentex.lib.core.temporal.workers.worker import AgentexWorker
+from agentex.lib.environment_variables import EnvironmentVariables
+from agentex.lib.utils.logging import make_logger
+
+from custom_activities import get_custom_activities
+from workflow import At010AgentChatWorkflow
+
+environment_variables = EnvironmentVariables.refresh()
+
+logger = make_logger(__name__)
+
+CUSTOM_TASK_QUEUE_NAME = "custom-activities-queue"
+
+
+async def main() -> None:
+    # Create a worker for the custom task queue
+    worker = AgentexWorker(
+        task_queue=CUSTOM_TASK_QUEUE_NAME,
+    )
+    
+    await worker.run(
+        activities=get_custom_activities(),
+        workflow=At010AgentChatWorkflow,
+    )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/tutorials/10_agentic/10_temporal/010_agent_chat/project/workflow.py

@@ -105,6 +105,51 @@ async def on_task_event_send(self, params: SendEventParams) -> None:
                     parent_span_id=span.id if span else None,
                 )
 
+            # Example: Use custom activities from the special task queue
+            user_message = params.event.content.content
+            
+            # Validate the input using custom activity
+            is_valid = await workflow.execute_activity(
+                "validate_custom_input",
+                user_message,
+                task_queue="custom-activities-queue",
+            )
+            
+            if is_valid:
+                # Transform the message using custom activity
+                transformed = await workflow.execute_activity(
+                    "transform_message",
+                    args=[user_message, "uppercase"],
+                    task_queue="custom-activities-queue",
+                )
+                
+                # Process some custom data
+                processed_result = await workflow.execute_activity(
+                    "process_special_data",
+                    {"message": transformed, "turn": self._state.turn_number if self._state else 0},
+                    task_queue="custom-activities-queue",
+                )
+                
+                await adk.messages.create(
+                    task_id=params.task.id,
+                    trace_id=params.task.id,
+                    content=TextContent(
+                        author="agent",
+                        content=f"Custom processing result: {processed_result}",
+                    ),
+                    parent_span_id=span.id if span else None,
+                )
+            else:
+                await adk.messages.create(
+                    task_id=params.task.id,
+                    trace_id=params.task.id,
+                    content=TextContent(
+                        author="agent",
+                        content="Invalid input received.",
+                    ),
+                    parent_span_id=span.id if span else None,
+                )
+
             # Call an LLM to respond to the user's message
             # When send_as_agent_task_message=True, returns a TaskMessage
             run_result = await adk.providers.openai.run_agent_streamed_auto_send(