.

Author: smoreinis

examples/tutorials/10_async/10_temporal/060_open_ai_agents_sdk_hello_world/060_open_ai_agents_sdk_hello_world/project/run_worker.py

@@ -1,17 +1,13 @@
 import asyncio
 
+from project.workflow import ExampleTutorialWorkflow
 from temporalio.contrib.openai_agents import OpenAIAgentsPlugin
 
-from project.workflow import At060OpenAiAgentsSdkHelloWorldWorkflow
 from agentex.lib.utils.debug import setup_debug_if_enabled
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.environment_variables import EnvironmentVariables
 from agentex.lib.core.temporal.activities import get_all_activities
 from agentex.lib.core.temporal.workers.worker import AgentexWorker
-from agentex.lib.core.temporal.plugins.openai_agents.models.temporal_streaming_model import (
-    TemporalStreamingModelProvider,
-)
-from agentex.lib.core.temporal.plugins.openai_agents.interceptors.context_interceptor import ContextInterceptor
 
 environment_variables = EnvironmentVariables.refresh()
 
@@ -29,40 +25,13 @@ async def main():
     # Add activities to the worker
     all_activities = get_all_activities() + []  # add your own activities here
 
-    # ============================================================================
-    # STREAMING SETUP: Interceptor + Model Provider
-    # ============================================================================
-    # This is where the streaming magic is configured! Two key components:
-    #
-    # 1. ContextInterceptor
-    #    - Threads task_id through activity headers using Temporal's interceptor pattern
-    #    - Outbound: Reads _task_id from workflow instance, injects into activity headers
-    #    - Inbound: Extracts task_id from headers, sets streaming_task_id ContextVar
-    #    - This enables runtime context without forking the Temporal plugin!
-    #
-    # 2. TemporalStreamingModelProvider
-    #    - Returns TemporalStreamingModel instances that read task_id from ContextVar
-    #    - TemporalStreamingModel.get_response() streams tokens to Redis in real-time
-    #    - Still returns complete response to Temporal for determinism/replay safety
-    #    - Uses AgentEx ADK streaming infrastructure (Redis XADD to stream:{task_id})
-    #
-    # Together, these enable real-time LLM streaming while maintaining Temporal's
-    # durability guarantees. No forked components - uses STANDARD OpenAIAgentsPlugin!
-    context_interceptor = ContextInterceptor()
-    temporal_streaming_model_provider = TemporalStreamingModelProvider()
-
     # Create a worker with automatic tracing
-    # IMPORTANT: We use the STANDARD temporalio.contrib.openai_agents.OpenAIAgentsPlugin
-    # No forking needed! The interceptor + model provider handle all streaming logic.
-    worker = AgentexWorker(
-        task_queue=task_queue_name,
-        plugins=[OpenAIAgentsPlugin(model_provider=temporal_streaming_model_provider)],
-        interceptors=[context_interceptor],
-    )
+    # We are also adding the Open AI Agents SDK plugin to the worker.
+    worker = AgentexWorker(task_queue=task_queue_name, plugins=[OpenAIAgentsPlugin()])
 
     await worker.run(
         activities=all_activities,
-        workflow=At060OpenAiAgentsSdkHelloWorldWorkflow,
+        workflow=ExampleTutorialWorkflow,
     )
 
 

examples/tutorials/10_async/10_temporal/060_open_ai_agents_sdk_hello_world/060_open_ai_agents_sdk_hello_world/tests/test_agent.py

@@ -1,136 +1,39 @@
 """
-Sample tests for AgentEx ACP agent.
+Tests for example-tutorial (OpenAI Agents SDK Hello World)
 
-This test suite demonstrates how to test the main AgentEx API functions:
-- Non-streaming event sending and polling
-- Streaming event sending
+Prerequisites:
+    - AgentEx services running (make dev)
+    - Temporal server running
+    - Agent running: agentex agents run --manifest manifest.yaml
 
-To run these tests:
-1. Make sure the agent is running (via docker-compose or `agentex agents run`)
-2. Set the AGENTEX_API_BASE_URL environment variable if not using default
-3. Run: pytest test_agent.py -v
-
-Configuration:
-- AGENTEX_API_BASE_URL: Base URL for the AgentEx server (default: http://localhost:5003)
-- AGENT_NAME: Name of the agent to test (default: example-tutorial)
+Run: pytest tests/test_agent.py -v
 """
 
-import os
-import uuid
-
 import pytest
-import pytest_asyncio
-from test_utils.async_utils import (
-    poll_messages,
-    send_event_and_poll_yielding,
-)
-
-from agentex import AsyncAgentex
-from agentex.types.task_message import TaskMessage
-from agentex.types.agent_rpc_params import ParamsCreateTaskRequest
-
-# Configuration from environment variables
-AGENTEX_API_BASE_URL = os.environ.get("AGENTEX_API_BASE_URL", "http://localhost:5003")
-AGENT_NAME = os.environ.get("AGENT_NAME", "at060-open-ai-agents-sdk-hello-world")
-
 
-@pytest_asyncio.fixture
-async def client():
-    """Create an AsyncAgentex client instance for testing."""
-    client = AsyncAgentex(base_url=AGENTEX_API_BASE_URL)
-    yield client
-    await client.close()
+from agentex.lib.testing import async_test_agent, assert_valid_agent_response
 
+AGENT_NAME = "example-tutorial"
 
-@pytest.fixture
-def agent_name():
-    """Return the agent name for testing."""
-    return AGENT_NAME
 
+@pytest.mark.asyncio
+async def test_agent_basic():
+    """Test basic agent functionality."""
+    async with async_test_agent(agent_name=AGENT_NAME) as test:
+        response = await test.send_event("Test message", timeout_seconds=60.0)
+        assert_valid_agent_response(response)
 
-@pytest_asyncio.fixture
-async def agent_id(client, agent_name):
-    """Retrieve the agent ID based on the agent name."""
-    agents = await client.agents.list()
-    for agent in agents:
-        if agent.name == agent_name:
-            return agent.id
-    raise ValueError(f"Agent with name {agent_name} not found.")
 
-
-class TestNonStreamingEvents:
-    """Test non-streaming event sending and polling."""
-
-    @pytest.mark.asyncio
-    async def test_send_event_and_poll(self, client: AsyncAgentex, agent_id: str):
-        """Test sending an event and polling for the response."""
-        task_response = await client.agents.create_task(agent_id, params=ParamsCreateTaskRequest(name=uuid.uuid1().hex))
-        task = task_response.result
-        assert task is not None
-
-        # Poll for the initial task creation message
-        async for message in poll_messages(
-            client=client,
-            task_id=task.id,
-            timeout=30,
-            sleep_interval=1.0,
-        ):
-            assert isinstance(message, TaskMessage)
-            if message.content and message.content.type == "text" and message.content.author == "agent":
-                # Check for the Haiku Assistant welcome message
-                assert "Haiku Assistant" in message.content.content
-                assert "Temporal" in message.content.content
+@pytest.mark.asyncio
+async def test_agent_streaming():
+    """Test streaming responses."""
+    async with async_test_agent(agent_name=AGENT_NAME) as test:
+        events = []
+        async for event in test.send_event_and_stream("Stream test", timeout_seconds=60.0):
+            events.append(event)
+            if event.get("type") == "done":
                 break
-
-        # Send event and poll for response with streaming updates
-        user_message = "Hello how is life?"
-        print(f"[DEBUG 060 POLL] Sending message: '{user_message}'")
-
-        # Use yield_updates=True to get all streaming chunks as they're written
-        final_message = None
-        async for message in send_event_and_poll_yielding(
-            client=client,
-            agent_id=agent_id,
-            task_id=task.id,
-            user_message=user_message,
-            timeout=30,
-            sleep_interval=1.0,
-            yield_updates=True,  # Get updates as streaming writes chunks
-        ):
-            if message.content and message.content.type == "text" and message.content.author == "agent":
-                print(
-                    f"[DEBUG 060 POLL] Received update - Status: {message.streaming_status}, "
-                    f"Content length: {len(message.content.content)}"
-                )
-                final_message = message
-
-                # Stop polling once we get a DONE message
-                if message.streaming_status == "DONE":
-                    print(f"[DEBUG 060 POLL] Streaming complete!")
-                    break
-
-        # Verify the final message has content (the haiku)
-        assert final_message is not None, "Should have received an agent message"
-        assert final_message.content is not None, "Final message should have content"
-        assert len(final_message.content.content) > 0, "Final message should have haiku content"
-
-        print(f"[DEBUG 060 POLL] ✅ Successfully received haiku response!")
-        print(f"[DEBUG 060 POLL] Final haiku:\n{final_message.content.content}")
-        pass
-
-
-class TestStreamingEvents:
-    """Test streaming event sending (backend verification via polling)."""
-
-    @pytest.mark.asyncio
-    async def test_send_event_and_stream(self, client: AsyncAgentex, agent_id: str):
-        """
-        Streaming test placeholder.
-
-        NOTE: SSE streaming is tested via the UI (agentex-ui subscribeTaskState).
-        Backend streaming functionality is verified in test_send_event_and_poll.
-        """
-        pass
+        assert len(events) > 0
 
 
 if __name__ == "__main__":