Add logic

Author: danielmillerp

examples/demos/procurement_agent/manifest.yaml

@@ -106,10 +106,10 @@ agent:
 
   # Optional: Set Environment variables for running your agent locally as well 
   # as for deployment later on
-  # env:
-  #   OPENAI_API_KEY: "<YOUR_OPENAI_API_KEY_HERE>"
-  #   OPENAI_BASE_URL: "<YOUR_OPENAI_BASE_URL_HERE>"
-  #   OPENAI_ORG_ID: "<YOUR_OPENAI_ORG_ID_HERE>"
+  env:
+    OPENAI_API_KEY: ""
+    # OPENAI_BASE_URL: "<YOUR_OPENAI_BASE_URL_HERE>"
+    OPENAI_ORG_ID: ""
 
 
 # Deployment Configuration

examples/demos/procurement_agent/project/acp.py

@@ -1,5 +1,11 @@
 import os
 import sys
+from temporalio.contrib.openai_agents import OpenAIAgentsPlugin
+from datetime import timedelta
+from agentex.lib.core.temporal.plugins.openai_agents.interceptors.context_interceptor import ContextInterceptor
+from agentex.lib.core.temporal.plugins.openai_agents.models.temporal_streaming_model import (
+    TemporalStreamingModelProvider,
+)
 
 # === DEBUG SETUP (AgentEx CLI Debug Support) ===
 if os.getenv("AGENTEX_DEBUG_ENABLED") == "true":
@@ -36,29 +42,16 @@
 from agentex.lib.sdk.fastacp.fastacp import FastACP
 from agentex.lib.types.fastacp import TemporalACPConfig
 
+context_interceptor = ContextInterceptor()
+streaming_model_provider = TemporalStreamingModelProvider()
 
 # Create the ACP server
 acp = FastACP.create(
     acp_type="async",
     config=TemporalACPConfig(
-        # When deployed to the cluster, the Temporal address will automatically be set to the cluster address
-        # For local development, we set the address manually to talk to the local Temporal service set up via docker compose
         type="temporal",
-        temporal_address=os.getenv("TEMPORAL_ADDRESS", "localhost:7233")
+        temporal_address=os.getenv("TEMPORAL_ADDRESS", "localhost:7233"),
+        plugins=[OpenAIAgentsPlugin(model_provider=streaming_model_provider)],
+        interceptors=[context_interceptor]
     )
-)
-
-
-# Notice that we don't need to register any handlers when we use type="temporal"
-# If you look at the code in agentex.sdk.fastacp.impl.temporal_acp
-# You can see that these handlers are automatically registered when the ACP is created
-
-# @acp.on_task_create
-# This will be handled by the method in your workflow that is decorated with @workflow.run
-
-# @acp.on_task_event_send
-# This will be handled by the method in your workflow that is decorated with @workflow.signal(name=SignalName.RECEIVE_MESSAGE)
-
-# @acp.on_task_cancel
-# This does not need to be handled by your workflow.
-# It is automatically handled by the temporal client which cancels the workflow directly 
+)

examples/demos/procurement_agent/project/activities.py

@@ -0,0 +1,73 @@
+from temporalio import activity
+import asyncio
+from temporalio import workflow
+from datetime import timedelta
+from agentex.lib.utils.logging import make_logger
+from project.events import SubmitalApprovalEvent, ShipmentDepartedFactoryEvent, ShipmentArrivedSiteEvent, InspectionFailedEvent
+import uuid
+from datetime import datetime
+
+logger = make_logger(__name__)
+
+@activity.defn
+async def issue_purchase_order(event: SubmitalApprovalEvent) -> str:
+    """
+    Issues a purchase order for construction materials.
+
+    Call this when:
+    - A submittal is approved (Submittal_Approved event)
+    - Human feedback requests reissuing a purchase order
+    """
+    uuid_purchase_order = str(uuid.uuid4())
+    # wait for 5 seconds as if we were calling an API to issue a purchase order
+    await asyncio.sleep(5)
+    logger.info(f"Issuing purchase order: {event}")
+    logger.info(f"Purchase order ID: {uuid_purchase_order}")
+
+    return f"Successfully issued purchase order with ID: {uuid_purchase_order}"  
+
+@activity.defn
+async def flag_potential_issue(event: ShipmentDepartedFactoryEvent) -> str:
+    """
+    Flags a potential issue with a delivery date.
+
+    Call this when:
+    - A shipment departure creates timeline concerns (Shipment_Departed_Factory event)
+    - Human feedback identifies a potential delivery issue
+    """
+    logger.info(f"Flagging potential issue: {event}")
+    logger.info(f"Potential issue flagged with delivery date: {event.eta}")
+    # imagine this is a call to an API to flag a potential issue, perhaps a notification to a team member
+    await asyncio.sleep(1)
+    return f"Potential issue flagged with delivery date: {event.eta}"
+
+@activity.defn
+async def notify_team_shipment_arrived(event: ShipmentDepartedFactoryEvent) -> str:
+    """
+    Notifies the team that a shipment has arrived.
+
+    Call this when:
+    - A shipment arrives at the site (Shipment_Arrived_Site event)
+    - Human feedback requests team notification
+    """
+    logger.info(f"Notifying team that shipment has arrived: {event.item}")
+    logger.info(f"Team notification sent for arrival of: {event.item}")
+    # imagine this is a call to an API to notify the team that a shipment has arrived, perhaps a notification to a team member
+    await asyncio.sleep(1)
+
+    return f"Notifying team that shipment has arrived: {event.item}"
+
+@activity.defn
+async def schedule_inspection(event: ShipmentDepartedFactoryEvent) -> str:
+    """
+    Schedules an inspection for delivered materials.
+
+    Call this when:
+    - A shipment arrives at the site (Shipment_Arrived_Site event)
+    - Human feedback requests scheduling an inspection
+    """
+    inspection_date = datetime.now() + timedelta(days=1)
+    logger.info(f"Scheduling inspection for: {event.item} on {inspection_date}")
+    # imagine this is a call to an API to schedule an inspection
+    await asyncio.sleep(1)
+    return f"Scheduling inspection for {event.item} on {inspection_date}"

examples/demos/procurement_agent/project/agents/procurement_agent.py

@@ -0,0 +1,96 @@
+"""Event agent for processing procurement events and taking actions."""
+
+from agents import Agent, function_tool
+from datetime import datetime, timedelta
+from temporalio.contrib import openai_agents
+from project.activities import (
+    issue_purchase_order,
+    flag_potential_issue,
+    notify_team_shipment_arrived,
+    schedule_inspection,
+)
+from temporalio import workflow
+
+
+@function_tool
+async def wait_for_human(recommended_action: str) -> str:
+    """
+    When the we are stuck and need to ask a human for help, call this tool. Please provide the recommended action to the human, should be yes or no recommendation.
+
+    Args:
+        recommended_action: The recommended action to take
+
+    Returns:
+        The human's response
+    """
+    workflow_instance = workflow.instance()
+
+    await workflow.wait_condition(
+        lambda: not workflow_instance.human_queue.empty(),
+        timeout=None,
+    )
+
+
+    while not workflow_instance.human_queue.empty():
+        human_input = await workflow_instance.human_queue.get()
+        print(f"[WORKFLOW] Processing human message from queue")
+
+        return human_input
+
+
+def new_procurement_agent(event_log: list, master_construction_schedule: str, human_input_learnings: list) -> Agent:
+    """
+    Create an agent that processes procurement events and takes actions.
+
+    Args:
+        event_log: History of events that have occurred
+        master_construction_schedule: Current construction schedule
+        human_input_learnings: Past escalations and human decisions
+
+    Returns:
+        Agent configured to process events and call tools
+    """
+    instructions = f"""
+You are a procurement agent for a commercial building construction project.
+
+Your role is to monitor procurement events, take appropriate actions, and escalate critical issues to a human with a recommended action.
+
+If the user says no or has feedback, please come up with another solution and call the wait_for_human tool again (you can call it as many times as needed).
+
+## Context
+
+Master Construction Schedule:
+{master_construction_schedule}
+
+Event History:
+{event_log}
+
+Past Learnings from Escalations:
+{human_input_learnings}
+
+Current Date: {datetime.now().isoformat()}
+    """
+
+    start_to_close_timeout = timedelta(days=1)
+
+    return Agent(
+        name="Procurement Event Agent",
+        instructions=instructions,
+        model="gpt-4o",
+        tools=[
+            openai_agents.workflow.activity_as_tool(
+                issue_purchase_order, start_to_close_timeout=start_to_close_timeout
+            ),
+            openai_agents.workflow.activity_as_tool(
+                flag_potential_issue, start_to_close_timeout=start_to_close_timeout
+            ),
+            openai_agents.workflow.activity_as_tool(
+                notify_team_shipment_arrived,
+                start_to_close_timeout=start_to_close_timeout,
+            ),
+            openai_agents.workflow.activity_as_tool(
+                schedule_inspection, start_to_close_timeout=start_to_close_timeout
+            ),
+            wait_for_human,  # function_tool runs in workflow context
+        ],
+    )

examples/demos/procurement_agent/project/events.py

@@ -0,0 +1,29 @@
+from enum import Enum
+from pydantic import BaseModel, Field
+from datetime import datetime
+
+class EventType(Enum):
+    SUBMITTAL_APPROVED = "Submittal_Approved"
+    SHIPMENT_DEPARTED_FACTORY = "Shipment_Departed_Factory"
+    SHIPMENT_ARRIVED_SITE = "Shipment_Arrived_Site"
+    INSPECTION_FAILED = "Inspection_Failed"
+    HUMAN_INPUT = "Human_Input"
+
+class SubmitalApprovalEvent(BaseModel):
+    event_type: EventType = Field(default=EventType.SUBMITTAL_APPROVED)
+    item: str
+
+class ShipmentDepartedFactoryEvent(BaseModel):
+    event_type: EventType = Field(default=EventType.SHIPMENT_DEPARTED_FACTORY)
+    item: str
+    eta: datetime
+
+class ShipmentArrivedSiteEvent(BaseModel):
+    event_type: EventType = Field(default=EventType.SHIPMENT_ARRIVED_SITE)
+    item: str
+    date_arrived: datetime
+
+class InspectionFailedEvent(BaseModel):
+    event_type: EventType = Field(default=EventType.INSPECTION_FAILED)
+    item: str
+    inspection_date: datetime

examples/demos/procurement_agent/project/master_construction_schedule.py

@@ -0,0 +1,38 @@
+"""Master construction schedule data."""
+
+MASTER_CONSTRUCTION_SCHEDULE = """
+{
+    "project": {
+        "name": "Small Office Renovation",
+        "start_date": "2026-02-01",
+        "end_date": "2026-05-31"
+    },
+    "deliveries": [
+        {
+        "item": "Steel Beams",
+        "required_by": "2026-02-15",
+        "buffer_days": 5
+        },
+        {
+        "item": "HVAC Units",
+        "required_by": "2026-03-01",
+        "buffer_days": 7
+        },
+        {
+        "item": "Windows",
+        "required_by": "2026-03-15",
+        "buffer_days": 10
+        },
+        {
+        "item": "Flooring Materials",
+        "required_by": "2026-04-01",
+        "buffer_days": 3
+        },
+        {
+        "item": "Electrical Panels",
+        "required_by": "2026-04-15",
+        "buffer_days": 5
+        }
+    ]
+}
+"""

examples/demos/procurement_agent/project/run_worker.py

@@ -5,9 +5,14 @@
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.utils.debug import setup_debug_if_enabled
 from agentex.lib.environment_variables import EnvironmentVariables
-
+from temporalio.contrib.openai_agents import OpenAIAgentsPlugin
+from agentex.lib.core.temporal.plugins.openai_agents.interceptors.context_interceptor import ContextInterceptor
+from agentex.lib.core.temporal.plugins.openai_agents.hooks.activities import stream_lifecycle_content
+from agentex.lib.core.temporal.plugins.openai_agents.models.temporal_streaming_model import (
+    TemporalStreamingModelProvider,
+)
 from project.workflow import ProcurementAgentWorkflow
-
+from project.activities import issue_purchase_order, flag_potential_issue, notify_team_shipment_arrived, schedule_inspection
 
 environment_variables = EnvironmentVariables.refresh()
 
@@ -22,11 +27,16 @@ async def main():
     if task_queue_name is None:
         raise ValueError("WORKFLOW_TASK_QUEUE is not set")
 
-    all_activities = get_all_activities() + []  # add your own activities here
+    all_activities = get_all_activities() + [stream_lifecycle_content, issue_purchase_order, flag_potential_issue, notify_team_shipment_arrived, schedule_inspection]
+    
+    context_interceptor = ContextInterceptor()
+    streaming_model_provider = TemporalStreamingModelProvider()
 
     # Create a worker with automatic tracing
     worker = AgentexWorker(
         task_queue=task_queue_name,
+        plugins=[OpenAIAgentsPlugin(model_provider=streaming_model_provider)],
+        interceptors=[context_interceptor],
     )
 
     await worker.run(