LangGraph streaming with workflow streams

[brianstrauch]

• Wire LangGraph's get_stream_writer() to Temporal: inside activity-wrapped nodes, the writer is backed by a WorkflowStreamClient that signals chunks back to the owning workflow's WorkflowStream on a topic, letting external subscribers tail node output via WorkflowStreamClient.
• Thread an optional stream_writer through set_langgraph_config so the activity wrapper can install one per invocation; default behavior (no writer) is unchanged.

Test plan

• test_streaming_via_workflow_streams — node calls get_stream_writer() inside an activity; external WorkflowStreamClient receives {token: a/b/c} then {done: True}, and the workflow result reflects the accumulated state.
• test_workflow_publishes_astream_chunks — workflow iterates app.astream(...) and republishes each chunk on its own topic; subscriber sees per-node progress.

[Copilot]

Pull request overview

This PR adds an experimental integration path between LangGraph’s get_stream_writer() streaming mechanism and Temporal “Workflow Streams”, so activity-wrapped LangGraph nodes can publish incremental chunks that external subscribers can tail via WorkflowStreamClient.

Changes:

• Add streaming_topic and streaming_batch_interval options to LangGraphPlugin and thread them into the activity wrapper.
• Extend LangGraph runtime config restoration to accept an optional stream_writer, enabling activity invocations to install a per-call writer.
• Add tests covering (1) activity-side get_stream_writer() streaming to workflow streams and (2) workflow-side forwarding of astream() chunks to an external topic.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
tests/contrib/langgraph/test_streaming.py	Adds workflow-stream-backed streaming tests (activity-side writer + workflow-side republish of astream() chunks).
temporalio/contrib/langgraph/_plugin.py	Introduces plugin-level streaming configuration and passes it into the activity wrapper.
temporalio/contrib/langgraph/_langgraph_config.py	Adds stream_writer plumbing to set_langgraph_config so activities can inject a writer into the reconstructed Runtime.
temporalio/contrib/langgraph/_activity.py	Uses WorkflowStreamClient.from_within_activity() to back the injected writer with workflow-stream topic publishes when streaming is enabled.
temporalio/contrib/langgraph/__init__.py	Minor export ordering change (no functional changes).

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.