Add LangGraph workflow streams sample

.github/CODEOWNERS

@@ -1,4 +1,5 @@
 * @temporalio/sdk
+/langgraph_plugin/ @temporalio/sdk @temporalio/ai-sdk
 
 # SDK & Nexus own the README, pyproject.toml, and uv.lock
 /README.md @temporalio/sdk @temporalio/nexus

langgraph_plugin/README.md

@@ -16,6 +16,7 @@ Samples are organized by API style:
 | **Continue-as-new** | [graph_api/continue_as_new](graph_api/continue_as_new) | [functional_api/continue_as_new](functional_api/continue_as_new) | Multi-stage data pipeline that uses `continue-as-new` with task result caching so previously-completed stages are not re-executed. |
 | **ReAct Agent** | [graph_api/react_agent](graph_api/react_agent) | [functional_api/react_agent](functional_api/react_agent) | Tool-calling agent loop. Graph API uses conditional edges; Functional API uses a `while` loop. |
 | **Control Flow** | -- | [functional_api/control_flow](functional_api/control_flow) | Demonstrates parallel task execution, `for` loops, and `if/else` branching -- patterns that are natural in the Functional API. |
+| **Streaming** | [graph_api/streaming](graph_api/streaming) | -- | Streams live output from a running workflow via [Workflow Streams](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/workflow_streams): node tokens via `get_stream_writer()` + `streaming_topic`, plus workflow-side `astream` progress published with `WorkflowStream.topic().publish()`. |
 | **LangSmith Tracing** | [graph_api/langsmith_tracing](graph_api/langsmith_tracing) | [functional_api/langsmith_tracing](functional_api/langsmith_tracing) | Combines `LangGraphPlugin` with Temporal's `LangSmithPlugin` for durable execution + full observability of LLM calls. Requires API keys. |
 
 ## Prerequisites
@@ -67,6 +68,7 @@ uv run langgraph_plugin/<api>/langsmith_tracing/main.py
 - **Continue-as-new with caching** -- `cache()` captures completed task results; passing the cache to the next execution avoids re-running them.
 - **Conditional routing** -- Graph API's `add_conditional_edges` and Functional API's native `if/else`/`while` for agent loops.
 - **Parallel execution** -- Functional API launches multiple tasks concurrently by creating futures before awaiting them.
+- **Streaming** -- Workflow Streams expose a durable, offset-addressed event channel that external clients subscribe to while the workflow is still running. Nodes emit fine-grained tokens via `get_stream_writer()` (routed by the plugin's `streaming_topic`), and the workflow can publish coarse `astream` progress to its own topic.
 
 ## Related
 

langgraph_plugin/graph_api/streaming/README.md

@@ -0,0 +1,31 @@
+# Streaming (Graph API)
+
+Streams a LangGraph run to an external client while the workflow is still running, using Temporal's durable, offset-addressed [`WorkflowStream`](https://docs.temporal.io/). The graph writes a short story about a topic and emits both fine-grained tokens and node-completion progress on separate topics.
+
+## What This Sample Demonstrates
+
+- **Node token streaming** — the `write_story` node calls LangGraph's `get_stream_writer()` to emit tokens. The plugin's `streaming_topic="tokens"` routes those writes onto the `"tokens"` topic.
+- **Workflow-side `astream` publish** — the workflow drives the graph with `app.astream(...)` and publishes each node-completion chunk onto a `"progress"` topic it owns.
+- A single client subscribing to all topics and demultiplexing on `item.topic`.
+- Waiting for the client to acknowledge (via signal) before completing, since the stream disappears when the workflow ends.
+- **Idempotent consumption** — each token chunk carries a monotonic sequence id so the client can dedupe, because streaming is at-least-once per activity attempt (a retried node re-runs and re-publishes its writes).
+
+## Running the Sample
+
+Prerequisites: `uv sync --group langgraph` and a running Temporal dev server (`temporal server start-dev`).
+
+```bash
+# Terminal 1
+uv run langgraph_plugin/graph_api/streaming/run_worker.py
+
+# Terminal 2
+uv run langgraph_plugin/graph_api/streaming/run_workflow.py
+```
+
+## Files
+
+| File | Description |
+|------|-------------|
+| `workflow.py` | Graph node functions, graph definition, and `StreamingWorkflow` that publishes to the stream |
+| `run_worker.py` | Registers graph with `LangGraphPlugin` (`streaming_topic="tokens"`), starts worker |
+| `run_workflow.py` | Starts the workflow, subscribes to the stream, prints tokens and progress, then acks |

langgraph_plugin/graph_api/streaming/run_worker.py

@@ -0,0 +1,35 @@
+"""Worker for the streaming sample (Graph API)."""
+
+import asyncio
+import os
+
+from temporalio.client import Client
+from temporalio.contrib.langgraph import LangGraphPlugin
+from temporalio.worker import Worker
+
+from langgraph_plugin.graph_api.streaming.workflow import (
+    StreamingWorkflow,
+    make_streaming_graph,
+)
+
+
+async def main() -> None:
+    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
+    # streaming_topic routes node get_stream_writer() output onto the "tokens" topic.
+    plugin = LangGraphPlugin(
+        graphs={"streaming": make_streaming_graph()},
+        streaming_topic="tokens",
+    )
+
+    worker = Worker(
+        client,
+        task_queue="langgraph-streaming",
+        workflows=[StreamingWorkflow],
+        plugins=[plugin],
+    )
+    print("Worker started. Ctrl+C to exit.")
+    await worker.run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

langgraph_plugin/graph_api/streaming/run_workflow.py

@@ -0,0 +1,51 @@
+"""Start the streaming workflow and subscribe to its Workflow Stream (Graph API)."""
+
+import asyncio
+import os
+from datetime import timedelta
+
+from temporalio.client import Client
+from temporalio.contrib.workflow_streams import WorkflowStreamClient
+
+from langgraph_plugin.graph_api.streaming.workflow import StreamingWorkflow
+
+
+async def main() -> None:
+    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
+
+    handle = await client.start_workflow(
+        StreamingWorkflow.run,
+        "a brave robot",
+        id="streaming-workflow",
+        task_queue="langgraph-streaming",
+    )
+
+    # Subscribe to all topics on the workflow's stream and demultiplex on topic.
+    ws = WorkflowStreamClient.create(client, handle.id)
+    # Streaming is at-least-once per activity attempt, so a retried node may
+    # re-publish tokens. Dedupe on the chunk's seq to consume idempotently.
+    seen_tokens: set[int] = set()
+    async for item in ws.subscribe(
+        from_offset=0,
+        result_type=dict,
+        poll_cooldown=timedelta(milliseconds=50),
+    ):
+        if item.topic == "tokens":
+            seq = item.data["seq"]
+            if seq in seen_tokens:
+                continue  # duplicate from a node retry; already consumed.
+            seen_tokens.add(seq)
+            print(item.data["token"], end="", flush=True)
+        elif item.topic == "progress":
+            if item.data.get("done"):
+                # Let the workflow know we are done consuming so it can complete.
+                await handle.signal(StreamingWorkflow.ack_stream)
+                break
+            print(f"\n[progress] {item.data}")
+
+    result = await handle.result()
+    print(f"\n\nFinal result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

langgraph_plugin/graph_api/streaming/workflow.py

@@ -0,0 +1,97 @@
+"""Streaming with the LangGraph Graph API and Temporal Workflow Streams.
+
+A workflow's :class:`WorkflowStream` is a durable, offset-addressed event channel
+external clients can subscribe to while the workflow is still running. This sample
+demonstrates both ways the LangGraph plugin produces stream items:
+
+- **Node token streaming** -- the ``write_story`` node calls LangGraph's
+  ``get_stream_writer()`` to emit fine-grained tokens. The plugin is configured with
+  ``streaming_topic="tokens"`` (see ``run_worker.py``), which routes those writes onto
+  the ``"tokens"`` topic.
+- **Workflow-side ``astream`` publish** -- the workflow drives the graph with
+  ``app.astream(...)`` and publishes each node-completion chunk onto a ``"progress"``
+  topic it owns.
+
+A single client subscribes to all topics and demultiplexes on ``item.topic``.
+"""
+
+from datetime import timedelta
+
+from langgraph.config import get_stream_writer
+from langgraph.graph import START, StateGraph
+from temporalio import workflow
+from temporalio.contrib.langgraph import graph as temporal_graph
+from temporalio.contrib.workflow_streams import WorkflowStream
+from typing_extensions import TypedDict
+
+
+class State(TypedDict):
+    topic: str
+    story: str
+
+
+async def outline(state: State) -> dict[str, str]:
+    """Produce a short opening line. Runs first so ``astream`` emits an early chunk."""
+    return {"story": f"A story about {state['topic']}:"}
+
+
+async def write_story(state: State) -> dict[str, str]:
+    """Write the story, emitting each word as a token via the stream writer.
+
+    Streaming is at-least-once per activity attempt: if this node retries
+    (transient failure, worker restart) it re-runs from scratch and re-publishes
+    its writes, so subscribers may see the same token twice. Each chunk therefore
+    carries a monotonic ``seq`` so consumers can dedupe idempotently. A retry
+    re-emits the same ``seq`` values, letting the client drop the duplicates.
+    """
+    writer = get_stream_writer()
+    words = f"{state['story']} Once upon a time, there was {state['topic']}.".split()
+    for seq, word in enumerate(words):
+        writer({"seq": seq, "token": word + " "})
+    return {"story": " ".join(words)}
+
+
+def make_streaming_graph() -> StateGraph:
+    g = StateGraph(State)
+    activity_metadata = {
+        "execute_in": "activity",
+        "start_to_close_timeout": timedelta(seconds=10),
+    }
+    g.add_node("outline", outline, metadata=activity_metadata)
+    g.add_node("write_story", write_story, metadata=activity_metadata)
+    g.add_edge(START, "outline")
+    g.add_edge("outline", "write_story")
+    return g
+
+
+@workflow.defn
+class StreamingWorkflow:
+    def __init__(self) -> None:
+        # WorkflowStream must be constructed during workflow initialization.
+        self.stream = WorkflowStream()
+        self._stream_acked = False
+
+    @workflow.signal
+    def ack_stream(self) -> None:
+        """Signalled by the client once it has finished consuming the stream."""
+        self._stream_acked = True
+
+    @workflow.run
+    async def run(self, topic: str) -> str:
+        app = temporal_graph("streaming").compile()
+        progress = self.stream.topic("progress")
+
+        story = ""
+        async for chunk in app.astream({"topic": topic, "story": ""}):
+            # Each chunk is {node_name: {state updates}}. Forward it as progress.
+            progress.publish(chunk)
+            for node_update in chunk.values():
+                if "story" in node_update:
+                    story = node_update["story"]
+
+        progress.publish({"done": True})
+
+        # The stream disappears when the workflow completes, so wait until the
+        # client acknowledges it has finished consuming before returning.
+        await workflow.wait_condition(lambda: self._stream_acked)
+        return story

tests/langgraph_plugin/streaming_test.py

@@ -0,0 +1,71 @@
+import uuid
+from datetime import timedelta
+from typing import Any
+
+from temporalio.client import Client
+from temporalio.contrib.langgraph import LangGraphPlugin
+from temporalio.contrib.workflow_streams import WorkflowStreamClient
+from temporalio.worker import Worker
+
+from langgraph_plugin.graph_api.streaming.workflow import (
+    StreamingWorkflow,
+    make_streaming_graph,
+)
+
+
+async def test_streaming_graph_api(client: Client) -> None:
+    task_queue = f"streaming-test-{uuid.uuid4()}"
+    plugin = LangGraphPlugin(
+        graphs={"streaming": make_streaming_graph()},
+        streaming_topic="tokens",
+    )
+
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[StreamingWorkflow],
+        plugins=[plugin],
+    ):
+        handle = await client.start_workflow(
+            StreamingWorkflow.run,
+            "a brave robot",
+            id=f"streaming-{uuid.uuid4()}",
+            task_queue=task_queue,
+        )
+
+        ws = WorkflowStreamClient.create(client, handle.id)
+        tokens: list[dict[str, Any]] = []
+        progress: list[dict[str, Any]] = []
+        async for item in ws.subscribe(
+            from_offset=0,
+            result_type=dict,
+            poll_cooldown=timedelta(milliseconds=10),
+        ):
+            if item.topic == "tokens":
+                tokens.append(item.data)
+            elif item.topic == "progress":
+                if item.data.get("done"):
+                    await handle.signal(StreamingWorkflow.ack_stream)
+                    break
+                progress.append(item.data)
+
+        result = await handle.result()
+
+    # Tokens reassemble into the final story. Streaming is at-least-once per
+    # activity attempt, so dedupe on seq (keeping first-seen order) before
+    # reassembling, exactly as an idempotent consumer would.
+    assert tokens, "expected at least one token"
+    assert all("seq" in t and "token" in t for t in tokens)
+    seen: set[int] = set()
+    deduped: list[dict[str, Any]] = []
+    for t in tokens:
+        if t["seq"] not in seen:
+            seen.add(t["seq"])
+            deduped.append(t)
+    assembled = "".join(t["token"] for t in deduped).strip()
+    assert assembled == result
+
+    # Workflow-side astream publish: one chunk per node, in order.
+    assert [list(chunk)[0] for chunk in progress] == ["outline", "write_story"]
+    assert result == progress[-1]["write_story"]["story"]
+    assert "a brave robot" in result