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/agent_patterns/README.md

@@ -40,7 +40,7 @@ uv run openai_agents/agent_patterns/run_agents_as_tools_workflow.py
 ```
 
 ### Agent Routing and Handoffs
-Route requests to specialized agents based on content analysis (adapted for non-streaming):
+Route requests to specialized agents based on content analysis (adapted to consume the run's output in one piece; see [Streaming](../streaming/README.md) for streaming output to external subscribers):
 ```bash
 uv run openai_agents/agent_patterns/run_routing_workflow.py
 ```
@@ -94,4 +94,4 @@ This is really useful for latency: for example, you might have a very fast model
 
 The following patterns from the [reference repository](https://github.com/openai/openai-agents-python/tree/main/examples/agent_patterns) are not included in this Temporal adaptation:
 
-- **Streaming Guardrails**: Requires streaming capabilities which are not yet available in the Temporal integration
+- **Streaming Guardrails**: The pattern interrupts generation by inspecting deltas from inside the run loop. The Temporal integration does support streaming (see [Streaming](../streaming/README.md)), but the model call runs in an activity and the workflow only sees its events once that activity returns, so there is no in-run delta to act on mid-response.

openai_agents/basic/README.md

@@ -76,4 +76,5 @@ uv run openai_agents/basic/run_previous_response_id_workflow.py
 The following examples from the [reference repository](https://github.com/openai/openai-agents-python/tree/main/examples/basic) are not included in this Temporal adaptation:
 
 - **Session** - Stores state in local SQLite database, not appropriate for distributed workflows
-- **Stream Items/Stream Text** - Streaming is not supported in Temporal OpenAI Agents SDK integration
+
+**Stream Items/Stream Text** are adapted in [`../streaming/`](../streaming/README.md) rather than here. They need `streaming_topic` set on the plugin's `ModelActivityParameters`, so they run on their own worker instead of sharing this directory's.

openai_agents/handoffs/README.md

@@ -41,4 +41,4 @@ The workflow returns both the final response and complete message history for in
 
 The following patterns from the [reference repository](https://github.com/openai/openai-agents-python/tree/main/examples/handoffs) are not included in this Temporal adaptation:
 
-- **Message Filter Streaming**: Streaming capabilities are not yet available in the Temporal integration
+- **Message Filter Streaming**: Differs from the included message-filter example only in rendering the same run's output as it streams. The Temporal integration does support that — see [Streaming](../streaming/README.md) — but it is demonstrated there rather than duplicated here.

openai_agents/reasoning_content/README.md

@@ -34,4 +34,4 @@ uv run openai_agents/reasoning_content/run_reasoning_content_workflow.py
 
 ## Note on Streaming
 
-The original OpenAI Agents SDK example includes streaming capabilities, but since Temporal workflows do not support streaming yet, this example contains only the non-streaming approach.
+The original OpenAI Agents SDK example includes a streaming variant. This example keeps only the non-streaming approach for brevity; the integration does support streaming model output to external subscribers, which is covered in [Streaming](../streaming/README.md).

openai_agents/streaming/README.md

@@ -0,0 +1,130 @@
+# 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`. It has to be constructed from a
+  method named `__init__` — `WorkflowStream` inspects its caller's frame and
+  raises otherwise — and `@workflow.init` is what makes the workflow's run
+  argument (carrying `stream_state` for continue-as-new) available there.
+
+## 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 partial events on the stream — they are flushed
+  before the failure is reported — and the retry publishes a whole new
+  response. `stream_events()` in the workflow only sees the successful attempt,
+  so the workflow's return value stays correct while a naive subscriber renders
+  the truncated attempt followed by the full one.
+
+  The plugin's streaming activity publishes no retry marker, so subscribers
+  detect this in band: every OpenAI stream event carries a `sequence_number`
+  that starts at 0 per response, and a number that fails to advance means a new
+  attempt. `run_stream_text_workflow.py` prints a notice at that seam;
+  `workflow_streams/run_llm.py` shows the fuller treatment, where an activity
+  you own publishes an explicit `RetryEvent` from `activity.info().attempt` and
+  the consumer erases the failed attempt's output.

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,98 @@
+"""Start StreamTextWorkflow and render its model output as it streams.
+
+Delivery is at-least-once per model-activity attempt: an attempt that fails
+mid-response leaves its partial deltas on the stream, and the retry publishes
+a whole new response. Unlike ``workflow_streams/activities/llm_activity.py``,
+which publishes an explicit ``RetryEvent`` on ``activity.info().attempt > 1``,
+the plugin's streaming activity emits no retry marker — so this subscriber
+infers a new attempt from the stream itself and says so, rather than silently
+running the two responses together. (``workflow_streams/run_llm.py`` goes a
+step further and erases the failed attempt's output with ANSI escapes.)
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from typing import Any, cast
+
+from agents.items import TResponseStreamEvent
+from openai.types.responses import ResponseCompletedEvent, 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.
+    last_sequence = -1
+    response_in_flight = False
+    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)
+
+        # Every event carries a sequence_number that starts at 0 per response,
+        # so a number that does not advance means a new response is streaming.
+        # That is a retry only if the previous one never completed: each turn
+        # of a multi-turn run is its own response and restarts the count too.
+        # The retry is an independently sampled answer rather than a
+        # continuation, so mark the seam instead of letting the failed
+        # attempt's partial text run into the new one. The workflow's return
+        # value is unaffected — stream_events() there sees only the attempt
+        # that succeeded.
+        sequence = event.sequence_number
+        if sequence <= last_sequence and response_in_flight:
+            print("\n\n[model activity retried — output restarts here]\n")
+        last_sequence = sequence
+        response_in_flight = not isinstance(event, ResponseCompletedEvent)
+
+        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())