streaming

Streaming

Forward Gemini model output to an external subscriber in real time. TemporalAsyncClient(streaming_topic="gemini") publishes each generate_content_stream chunk onto a workflow-hosted WorkflowStream as it arrives. A subscriber connects with WorkflowStreamClient and reads the topic live while the workflow runs durably.

What This Sample Demonstrates

• TemporalAsyncClient(streaming_topic=...) publishing chunks to a topic
• Hosting a WorkflowStream in @workflow.init (required for streaming)
• Consuming the stream externally via WorkflowStreamClient.subscribe(...)
• Holding the workflow open on a signal so the subscriber can drain the stream
• Timing out both sides of the handshake so neither waits forever

Timeouts

The rendezvous between workflow and subscriber has two failure modes, and both sides are bounded here rather than waiting forever:

• The subscriber stops reading only when a chunk carries finish_reason. If generation fails mid-stream that chunk never arrives, so the consume loop is wrapped in asyncio.wait_for.
• The workflow stops waiting only on the finish signal. If the subscriber crashes before signaling, nothing releases it — so workflow.wait_condition takes a timeout and the workflow completes without the signal.

Running the Sample

Prerequisites: install dependencies, set GOOGLE_API_KEY, and start a Temporal dev server. See the suite README.

# Terminal 1
uv run google_genai/streaming/run_worker.py

# Terminal 2
uv run google_genai/streaming/run_workflow.py

Files

File	Description
workflow.py	StreamingWorkflow — streams chunks to the gemini topic
run_worker.py	Registers GoogleGenAIPlugin, starts the worker
run_workflow.py	Starts the workflow and consumes the stream live