Add openai_agents streaming sample

.github/CODEOWNERS

@@ -23,4 +23,5 @@
 /tests/google_adk_agents/ @temporalio/sdk @temporalio/ai-sdk
 /tests/langgraph_plugin/  @temporalio/sdk @temporalio/ai-sdk
 /tests/langsmith_tracing/ @temporalio/sdk @temporalio/ai-sdk
+/tests/openai_agents/     @temporalio/sdk @temporalio/ai-sdk
 /tests/strands_plugin/    @temporalio/sdk @temporalio/ai-sdk

openai_agents/README.md

@@ -36,3 +36,4 @@ Each directory contains a complete example with its own README for detailed inst
 - **[Customer Service](./customer_service/README.md)** - Interactive customer service agent with escalation capabilities, demonstrating conversational workflows.
 - **[Reasoning Content](./reasoning_content/README.md)** - Example of how to retrieve the thought process of reasoning models.
 - **[Financial Research Agent](./financial_research_agent/README.md)** - Multi-agent financial research system with planner, search, analyst, writer, and verifier agents collaborating.
+- **[Streaming](./streaming/README.md)** - `Runner.run_streamed` with buffered token streaming to external subscribers via `temporalio.contrib.workflow_streams`. **Experimental.**

openai_agents/streaming/README.md

@@ -0,0 +1,122 @@
+# Streaming OpenAI Agents
+
+> **Experimental.** These samples use the streaming support in
+> `temporalio.contrib.openai_agents` together with
+> `temporalio.contrib.workflow_streams`. Both are experimental and their APIs
+> may change in future versions.
+
+*Adapted from the [OpenAI Agents SDK basic examples](https://github.com/openai/openai-agents-python/tree/main/examples/basic)*
+
+Before running these examples, be sure to review the [prerequisites and background on the integration](../README.md).
+
+The OpenAI Agents SDK streams model output via `Runner.run_streamed`, which
+yields events as the model produces them. Inside a Temporal workflow the model
+call runs in an activity, so the workflow cannot iterate the live HTTP stream
+directly. Instead the plugin runs `model.stream_response()` in a streaming
+activity, and that activity publishes each event to the workflow's
+[`WorkflowStream`](../../workflow_streams/README.md) so external subscribers
+see events as they are produced.
+
+Publishing is batched: the activity coalesces events over
+`ModelActivityParameters.streaming_batch_interval` (default 100ms) before
+signalling the workflow. Call this **buffered token streaming** — deltas reach
+subscribers within a batch window of being produced, not on every byte. At
+typical model speeds one batch carries several tokens, so output arrives in
+small bursts rather than glyph-by-glyph. Lower the interval for smoother
+output at the cost of more signals.
+
+Two things to know before reading the samples:
+
+* `streaming_topic` is **required** for `Runner.run_streamed`. If it is unset,
+  `run_streamed` raises before scheduling any activity.
+* The workflow must host a `WorkflowStream`, constructed in `@workflow.init` so
+  the publish-signal handler is registered before the activity publishes.
+  Without one, the publishes are unhandled and silently dropped.
+
+## Running the Examples
+
+First, start the worker (supports both examples):
+
+```bash
+uv run openai_agents/streaming/run_worker.py
+```
+
+Then run either example in another terminal.
+
+### `stream_text` — buffered text deltas
+
+Adapted from [`examples/basic/stream_text.py`][upstream-text]. The workflow
+just calls `Runner.run_streamed`; the subscriber renders the
+`ResponseTextDeltaEvent`s the streaming activity publishes on the `events`
+topic.
+
+Subscribers receive **native OpenAI events** (`TResponseStreamEvent`), because
+the activity publishes them straight from `Model.stream_response`. That differs
+from `stream_events()` inside the workflow, which yields the agents-SDK
+`StreamEvent` union — raw model events arrive there wrapped as
+`RawResponsesStreamEvent.data`.
+
+[upstream-text]: https://github.com/openai/openai-agents-python/blob/main/examples/basic/stream_text.py
+
+```bash
+uv run openai_agents/streaming/run_stream_text_workflow.py
+```
+
+### `stream_items` — agent-level events with a tool call
+
+Adapted from [`examples/basic/stream_items.py`][upstream-items]. Renders agent
+updates, tool calls, tool outputs, and message outputs as a play-by-play.
+
+The agents SDK builds those higher-level events from the model output, so they
+exist only inside the workflow — the streaming activity never sees them. This
+workflow therefore does its own publishing: it iterates
+`result.stream_events()` and forwards each event of interest to an `items`
+topic as a small serializable `ItemEvent`. (The agents-SDK event types carry
+the originating `Agent`, which holds tool callables and so cannot be
+serialized.) `stream_events()` resolves a turn at a time — each model call is
+one activity — so a multi-turn run like this one reaches the subscriber
+progressively rather than in one lump.
+
+[upstream-items]: https://github.com/openai/openai-agents-python/blob/main/examples/basic/stream_items.py
+
+```bash
+uv run openai_agents/streaming/run_stream_items_workflow.py
+```
+
+## How it works
+
+1. The workflow constructs a `WorkflowStream` in `@workflow.init`.
+2. `OpenAIAgentsPlugin` is configured with `streaming_topic="events"`, which
+   routes `Runner.run_streamed` to `invoke_model_activity_streaming`.
+3. Inside that activity each event from the live HTTP stream is both collected
+   (returned to the workflow when the activity completes) and published to the
+   stream via `WorkflowStreamClient.from_within_activity()`.
+4. Just before returning, the workflow publishes a terminator on a separate
+   `done` topic, then sleeps briefly so the subscriber's next poll can drain
+   the tail of the stream — the log lives in workflow memory and disappears
+   when the run completes.
+5. External code subscribes with
+   `WorkflowStreamClient.create(...).subscribe([...], result_type=RawValue)`
+   and breaks on the terminator. `RawValue` keeps the payloads undecoded so
+   each topic can be decoded against its own type. If the workflow reaches a
+   terminal state without publishing a terminator (a failure, say), the
+   iterator exhausts on its own and the following `handle.result()` raises.
+
+In the workflow, `stream_events()` resolves only after the model activity
+returns, so the workflow itself does not see deltas as they arrive — the
+streaming benefit is for external observers.
+
+## Notes
+
+* Streaming is incompatible with `use_local_activity=True`: local activities
+  support neither heartbeats nor the workflow stream signal channel.
+* The streaming activity heartbeats on a background task, so set
+  `heartbeat_timeout` well below `start_to_close_timeout` to detect a stuck
+  model call early.
+* Delivery is at-least-once per activity attempt. An attempt that fails
+  mid-response leaves its events on the stream and the retry publishes a second
+  sequence; `stream_events()` in the workflow only sees the final successful
+  attempt. The [workflow_streams module
+  documentation](https://github.com/temporalio/sdk-python/blob/main/temporalio/contrib/workflow_streams/README.md)
+  covers the trade and the conventional `RETRY` event pattern for surfacing it
+  to consumers.

openai_agents/streaming/activities/joke_activities.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+import random
+
+from temporalio import activity
+
+
+@activity.defn
+async def how_many_jokes() -> int:
+    """Return a random integer of jokes to tell between 1 and 10 (inclusive)."""
+    return random.randint(1, 10)

openai_agents/streaming/run_stream_items_workflow.py

@@ -0,0 +1,65 @@
+"""Start StreamItemsWorkflow and render its run as a play-by-play."""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+
+from temporalio.client import Client
+from temporalio.common import RawValue
+from temporalio.contrib.openai_agents import OpenAIAgentsPlugin
+from temporalio.contrib.workflow_streams import WorkflowStreamClient
+
+from openai_agents.streaming.shared import (
+    TASK_QUEUE,
+    TOPIC_DONE,
+    TOPIC_ITEMS,
+    ItemEvent,
+)
+from openai_agents.streaming.workflows.stream_items_workflow import (
+    StreamItemsInput,
+    StreamItemsWorkflow,
+)
+
+
+async def main() -> None:
+    client = await Client.connect(
+        "localhost:7233",
+        plugins=[OpenAIAgentsPlugin()],
+    )
+
+    workflow_id = f"stream-items-{uuid.uuid4().hex[:8]}"
+    handle = await client.start_workflow(
+        StreamItemsWorkflow.run,
+        StreamItemsInput(),
+        id=workflow_id,
+        task_queue=TASK_QUEUE,
+    )
+
+    stream = WorkflowStreamClient.create(client, workflow_id)
+    converter = client.data_converter.payload_converter
+
+    print("=== Run starting ===")
+    # result_type=RawValue so the two topics can be decoded per item.topic.
+    # The raw model events the streaming activity publishes on TOPIC_EVENTS are
+    # on the stream too; this subscriber just isn't interested in them.
+    async for item in stream.subscribe([TOPIC_ITEMS, TOPIC_DONE], result_type=RawValue):
+        if item.topic == TOPIC_DONE:
+            break
+        event = converter.from_payload(item.data.payload, ItemEvent)
+        if event.kind == "agent_updated":
+            print(f"Agent updated: {event.detail}")
+        elif event.kind == "tool_call":
+            print(f"-- Tool was called: {event.detail}")
+        elif event.kind == "tool_output":
+            print(f"-- Tool output: {event.detail}")
+        elif event.kind == "message_output":
+            print(f"-- Message output:\n {event.detail}")
+
+    result = await handle.result()
+    print("=== Run complete ===")
+    print(result)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

openai_agents/streaming/run_stream_text_workflow.py

@@ -0,0 +1,70 @@
+"""Start StreamTextWorkflow and render its model output as it streams."""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from typing import Any, cast
+
+from agents.items import TResponseStreamEvent
+from openai.types.responses import ResponseTextDeltaEvent
+from temporalio.client import Client
+from temporalio.common import RawValue
+from temporalio.contrib.openai_agents import OpenAIAgentsPlugin
+from temporalio.contrib.workflow_streams import WorkflowStreamClient
+
+from openai_agents.streaming.shared import TASK_QUEUE, TOPIC_DONE, TOPIC_EVENTS
+from openai_agents.streaming.workflows.stream_text_workflow import (
+    StreamTextInput,
+    StreamTextWorkflow,
+)
+
+# TResponseStreamEvent is a typing.Annotated union rather than a class, so it
+# needs a cast to satisfy from_payload's type[T] signature. The plugin's
+# pydantic converter resolves the union's discriminator at runtime.
+EVENT_TYPE = cast(type, TResponseStreamEvent)
+
+
+async def main() -> None:
+    # The plugin's data converter is what decodes the OpenAI event payloads
+    # published on TOPIC_EVENTS.
+    client = await Client.connect(
+        "localhost:7233",
+        plugins=[OpenAIAgentsPlugin()],
+    )
+
+    workflow_id = f"stream-text-{uuid.uuid4().hex[:8]}"
+    handle = await client.start_workflow(
+        StreamTextWorkflow.run,
+        StreamTextInput(prompt="Please tell me 5 jokes."),
+        id=workflow_id,
+        task_queue=TASK_QUEUE,
+    )
+
+    stream = WorkflowStreamClient.create(client, workflow_id)
+    converter = client.data_converter.payload_converter
+
+    # A single iterator over both topics — one subscriber, no cancellation race
+    # between concurrent ones. result_type=RawValue delivers the underlying
+    # Payload so heterogeneous topics can be decoded per item.topic. The loop
+    # ends on the in-band terminator, or by the iterator exhausting if the
+    # workflow reaches a terminal state without publishing one (e.g. on
+    # failure); either way handle.result() below surfaces the outcome.
+    async for item in stream.subscribe(
+        [TOPIC_EVENTS, TOPIC_DONE], result_type=RawValue
+    ):
+        if item.topic == TOPIC_DONE:
+            break
+        # Subscribers receive native OpenAI events, not the agents-SDK
+        # StreamEvent wrappers that stream_events() yields in the workflow.
+        event: Any = converter.from_payload(item.data.payload, EVENT_TYPE)
+        if isinstance(event, ResponseTextDeltaEvent):
+            print(event.delta, end="", flush=True)
+
+    result = await handle.result()
+    print("\n--- final result ---")
+    print(result)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

openai_agents/streaming/run_worker.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import timedelta
+
+from temporalio.client import Client
+from temporalio.contrib.openai_agents import (
+    ModelActivityParameters,
+    OpenAIAgentsPlugin,
+)
+from temporalio.worker import Worker
+
+from openai_agents.streaming.activities.joke_activities import how_many_jokes
+from openai_agents.streaming.shared import TASK_QUEUE, TOPIC_EVENTS
+from openai_agents.streaming.workflows.stream_items_workflow import (
+    StreamItemsWorkflow,
+)
+from openai_agents.streaming.workflows.stream_text_workflow import (
+    StreamTextWorkflow,
+)
+
+
+async def main() -> None:
+    logging.basicConfig(level=logging.INFO)
+    client = await Client.connect(
+        "localhost:7233",
+        plugins=[
+            OpenAIAgentsPlugin(
+                model_params=ModelActivityParameters(
+                    # The streaming activity heartbeats on a background task,
+                    # so a heartbeat_timeout well under start_to_close_timeout
+                    # detects a stuck model call early.
+                    heartbeat_timeout=timedelta(seconds=10),
+                    start_to_close_timeout=timedelta(minutes=5),
+                    # Required for Runner.run_streamed: the topic the streaming
+                    # activity publishes raw model events to.
+                    streaming_topic=TOPIC_EVENTS,
+                ),
+            ),
+        ],
+    )
+
+    worker = Worker(
+        client,
+        task_queue=TASK_QUEUE,
+        workflows=[StreamTextWorkflow, StreamItemsWorkflow],
+        activities=[how_many_jokes],
+    )
+    await worker.run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

openai_agents/streaming/shared.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import timedelta
+
+TASK_QUEUE = "openai-agents-streaming-task-queue"
+
+# Topic the streaming activity publishes raw model stream events to. Must match
+# OpenAIAgentsPlugin(model_params=ModelActivityParameters(streaming_topic=...)).
+# Events on this topic are native OpenAI `TResponseStreamEvent`s, not the
+# agents-SDK `StreamEvent` wrappers that `stream_events()` yields.
+TOPIC_EVENTS = "events"
+
+# Topic the stream_items workflow publishes its own higher-level events to. The
+# agents SDK builds those from the model output inside the workflow, so the
+# workflow — not the activity — is what publishes them.
+TOPIC_ITEMS = "items"
+
+# Topic the workflow publishes a terminator to once Runner.run_streamed has
+# finished. Subscribers watch both topics and break on the terminator, rather
+# than racing handle.result() against their next poll.
+TOPIC_DONE = "done"
+
+# How long a workflow holds its run open after publishing the terminator, so a
+# subscriber's next poll can drain the tail of the stream. The log lives in
+# workflow memory, so it disappears when the run completes.
+DRAIN_INTERVAL = timedelta(milliseconds=500)
+
+
+@dataclass
+class ItemEvent:
+    """One step of a run, as published on TOPIC_ITEMS.
+
+    The agents-SDK event types (`RunItemStreamEvent` and friends) carry the
+    originating `Agent`, which holds tool callables and so is not
+    serializable. Samples publish their own flattened event instead.
+    """
+
+    kind: str
+    """One of "agent_updated", "tool_call", "tool_output", "message_output"."""
+
+    detail: str