Add custom streaming to open ai agents sdk and temporal integration

Author: danielmillerp

src/agentex/lib/core/clients/temporal/utils.py

@@ -3,6 +3,7 @@
 from typing import Any
 
 from temporalio.client import Client, Plugin as ClientPlugin
+from temporalio.worker import Interceptor
 from temporalio.runtime import Runtime, TelemetryConfig, OpenTelemetryConfig
 from temporalio.contrib.pydantic import pydantic_data_converter
 from temporalio.contrib.openai_agents import OpenAIAgentsPlugin
@@ -61,6 +62,24 @@ def validate_client_plugins(plugins: list[Any]) -> None:
             )
 
 
+def validate_worker_interceptors(interceptors: list[Any]) -> None:
+    """
+    Validate that all items in the interceptors list are valid Temporal worker interceptors.
+
+    Args:
+        interceptors: List of interceptors to validate
+
+    Raises:
+        TypeError: If any interceptor is not a valid Interceptor instance
+    """
+    for i, interceptor in enumerate(interceptors):
+        if not isinstance(interceptor, Interceptor):
+            raise TypeError(
+                f"Interceptor at index {i} must be an instance of temporalio.worker.Interceptor, "
+                f"got {type(interceptor).__name__}"
+            )
+
+
 async def get_temporal_client(temporal_address: str, metrics_url: str | None = None, plugins: list[Any] = []) -> Client:
     """
     Create a Temporal client with plugin integration.

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

@@ -0,0 +1,58 @@
+"""OpenAI Agents SDK Temporal Plugin with Streaming Support.
+
+This module provides streaming capabilities for the OpenAI Agents SDK in Temporal
+using interceptors to thread task_id through workflows to activities.
+
+The streaming implementation works by:
+1. Using Temporal interceptors to thread task_id through the execution
+2. Streaming LLM responses to Redis in real-time from activities
+3. Returning complete responses to maintain Temporal determinism
+
+Example:
+    >>> from agentex.lib.core.temporal.plugins.openai_agents import (
+    ...     TemporalStreamingModelProvider,
+    ...     TemporalTracingModelProvider,
+    ...     ContextInterceptor,
+    ... )
+    >>> from temporalio.contrib.openai_agents import OpenAIAgentsPlugin, ModelActivityParameters
+    >>> from datetime import timedelta
+    >>>
+    >>> # Create streaming model provider
+    >>> model_provider = TemporalStreamingModelProvider()
+    >>>
+    >>> # Create STANDARD plugin with streaming model provider
+    >>> plugin = OpenAIAgentsPlugin(
+    ...     model_params=ModelActivityParameters(
+    ...         start_to_close_timeout=timedelta(seconds=120),
+    ...     ),
+    ...     model_provider=model_provider,
+    ... )
+    >>>
+    >>> # Register interceptor with worker
+    >>> interceptor = ContextInterceptor()
+    >>> # Add interceptor to worker configuration
+"""
+
+from agentex.lib.core.temporal.plugins.openai_agents import (
+    ContextInterceptor,
+    TemporalStreamingHooks,
+    TemporalStreamingModel,
+    TemporalTracingModelProvider,
+    TemporalStreamingModelProvider,
+    streaming_task_id,
+    streaming_trace_id,
+    stream_lifecycle_content,
+    streaming_parent_span_id,
+)
+
+__all__ = [
+    "TemporalStreamingModel",
+    "TemporalStreamingModelProvider",
+    "TemporalTracingModelProvider",
+    "ContextInterceptor",
+    "streaming_task_id",
+    "streaming_trace_id",
+    "streaming_parent_span_id",
+    "TemporalStreamingHooks",
+    "stream_lifecycle_content",
+]