Skip to content

Commit 4ec9ab0

Browse files
JasonSteving99brianstrauchclaude
authored
Add First-Class Google Gen AI SDK Integration to Contrib (#1378)
* Add First-Class Gemini SDK Integration to Contrib # Temporal Integration for the Google Gemini SDK This adds a first-class integration that lets users call the Gemini SDK's `AsyncClient` directly from within Temporal workflows. Every API call and tool invocation becomes a durable Temporal activity — giving full crash recovery, visibility in workflow event history, and replay safety — while keeping credentials entirely on the worker side. ## How it works The integration shims three layers of the Gemini SDK so that workflows can use `client.models`, `client.files`, `client.file_search_stores`, `client.chats`, and all other SDK modules naturally: ### `TemporalApiClient` (`_temporal_api_client.py`) A `BaseApiClient` subclass that replaces the SDK's HTTP layer. Instead of making network calls, `async_request` and `async_request_streamed` serialize the request and dispatch it through `workflow.execute_activity`. The real HTTP call happens inside the activity on the worker, where the actual `genai.Client` with real credentials lives. Sync methods raise immediately. Per-request `http_options` are validated (non-serializable fields like `httpx_client` are rejected), and `timeout` is mapped to Temporal's `start_to_close_timeout`. ### `TemporalAsyncFiles` / `TemporalAsyncFileSearchStores` (`_temporal_files.py`, `_temporal_file_search_stores.py`) Subclasses of `AsyncFiles` and `AsyncFileSearchStores` that override `upload`, `download`, `register_files`, and `upload_to_file_search_store` to dispatch the entire operation as a Temporal activity. This avoids filesystem access (`os` module) and credential token refresh in the workflow sandbox. Methods like `get`, `delete`, `list` are inherited and work through the `TemporalApiClient`'s `async_request` activity. File uploads accept `str` paths (resolved on the worker), `os.PathLike`, or `io.IOBase` (bytes serialized across the activity boundary). ### `TemporalAsyncClient` (`_temporal_async_client.py`) An `AsyncClient` subclass that wires in `TemporalAsyncFiles` and `TemporalAsyncFileSearchStores`. All other SDK modules (`models`, `tunings`, `caches`, `batches`, `live`, `tokens`, `operations`) are inherited unchanged since they only use `async_request` under the hood. ### `GeminiPlugin` (`_gemini_plugin.py`) A `SimplePlugin` that registers all activities, configures the Pydantic data converter, and passes `google.genai` through the workflow sandbox. Users pass a fully configured `genai.Client` — the plugin never constructs one itself. An optional `extra_credentials` parameter supports operations like `register_files` that need separate GCS credentials. ### `activity_as_tool` (`workflow.py`) Wraps any `@activity.defn` function so it looks like a regular async callable to Gemini's automatic function calling (AFC). When the model decides to call the tool, the SDK invokes the wrapper, which dispatches through `workflow.execute_activity`. Users can also pass plain workflow methods directly as tools — these run in-workflow without an activity. ### Batched streaming `generate_content_stream` is supported via a batched approach: the `async_request_streamed` activity collects all chunks from the real streaming response and returns them as a list. The workflow-side `TemporalApiClient` yields them back as an async generator so the SDK sees the expected interface. ## Usage ```python # Worker side client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"]) plugin = GeminiPlugin(client) # Workflow side @workflow.defn class MyWorkflow: @workflow.run async def run(self, query: str) -> str: client = gemini_client() response = await client.models.generate_content( model="gemini-2.5-flash", contents=query, config=types.GenerateContentConfig( tools=[activity_as_tool(my_tool)], ), ) return response.text ``` ## Testing 31 integration tests covering: - Basic `generate_content` and multi-chunk streaming - AFC tool calling (single-arg, multi-arg, workflow methods, sequential multi-tool, failure propagation) - Per-request `http_options` propagation (headers, api_version, base_url) - File upload via str path and `io.BytesIO`, file download - File search store upload - Multi-turn chat via `client.chats` - `TemporalAsyncClient` wiring verification - `TemporalApiClient` error paths (sync raises, low-level upload/download raises) - `activity_as_tool` validation and signature preservation - A full end-to-end integration test that exercises all real activity implementations (generate, stream, file upload, download, store upload, RAG query, store delete) with a mocked `genai.Client` — ensuring the actual activity code in `_gemini_activity.py` is covered, not just the workflow-side shims. * update lock * address PR feedback * upper bound google-genai dep * move to `.../contrib/google_genai/` * rename `gemini_client` -> `google_genai_client` * Rename `GeminiPlugin` -> `GoogleGenAIPlugin` * add to codeowners * Fix docstring errors breaking poe gen-docs pydoctor runs with warnings-as-errors: the bullet list needed a blank line before it, and the :func: reference must use the unqualified name since pydoctor relocates __all__ re-exports to the package page. * google_genai: add MCP support, interactions/agents, and durability tests Client-side MCP (Gemini Developer API): TemporalMcpClientSession subclasses mcp.ClientSession and routes list_tools/call_tool through {server}-list-tools and {server}-call-tool activities, so the SDK's in-workflow AFC loop drives MCP tools while the real session lives on the worker. Servers register on the plugin via mcp_servers={name: factory} with a pooled, idle-evicted worker-side connection (mcp_connection_idle_timeout). Server-side MCP (Vertex McpServer config and Interactions API MCP steps) flows through unchanged as data. Also in this change: - Interactions API and managed agents support, plus files/file-search activities - Collapse activity_config defaults to a single documented 60s start_to_close - activity_as_tool requires an explicit timeout (matches openai_agents/strands) - Replay and side-effect (ActivityTaskScheduled count) tests, incl. MCP - mcp is an optional dep: lazy import + TYPE_CHECKING so the package imports without it; declared in the dev group for tests Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * google_genai: add GoogleGenAIError and make Temporal own retries - Add GoogleGenAIError(ApplicationError) and register it via the plugin's workflow_failure_exception_types so it terminally fails the workflow rather than retrying the task; activity_as_tool validation now raises it. - Reject the SDK's own retry config instead of silently overriding it: the plugin raises ValueError if the genai.Client has http_options.retry_options, and the workflow-side client raises GoogleGenAIError on per-request retry_options (previously dropped silently). Both point users to the activity retry_policy via activity_config. - Classify API-call activity errors: catch google.genai.errors.APIError and re-raise as ApplicationError with non_retryable set by HTTP status (408/429/5xx retryable, 4xx fail fast), with the SDK error class name as the type. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * google_genai: add README, fix plugin name, document determinism, widen passthrough - Add a user-facing README (overview, install, hello world, tool calling, MCP, retries/errors, Vertex, composing). - Rename the plugin to the conventional "google_genai.GoogleGenAIPlugin". - Document the replay-determinism survey: the generate_content/AFC/MCP paths are replay-safe; note the one in-workflow caveat (Vertex batches.create auto-naming). - Add pydantic_core and annotated_types to sandbox passthrough so the SDK's in-workflow Pydantic validation doesn't reimport them after workflow load. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * google_genai: README — explain Vertex project/location, trim MCP section - Vertex AI: document that project/location must be set explicitly on the workflow-side client; auto-discovery can't run in the sandbox and would make in-workflow request formatting non-deterministic. - MCP: drop the server-side/interactions bullet (works unchanged, no wiring) and focus the section on the client-side path the plugin actually wires. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * google_genai: pin google-genai < 2.8.0 (in-workflow AFC regression) google-genai 2.8.0 regressed automatic function calling for plain workflow-method tools: the tool executes (its function response is correct) but its in-workflow state mutation is no longer visible on replay/query, failing test_workflow_method_as_tool. The activity_as_tool path is unaffected. Cap at < 2.8.0 until a fixed release ships upstream. Also simplify exclude-newer-package to disable the cutoff for google-adk outright (= false) rather than a dated pin that needs later cleanup. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * google_genai: fix poe lint (exports, import order, type nits) - Export GoogleGenAIError in __all__ (fixes unused-import in __init__ and the "not exported" warning where tests import it from the package). - Sort imports in _gemini_activity.py (ruff I001). - Suppress reportUnusedClass on _TemporalApiClient (used in the sibling module) and reportUnusedFunction on the autouse MCP fixture, matching repo convention. - Use collections.abc.AsyncIterator and annotate the test helper parameter. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * google_genai: add public testing utilities Add temporalio/contrib/google_genai/testing.py so users can test workflows that use TemporalAsyncClient without real Gemini API calls — the agent-framework "test fakes are table-stakes" expectation, matching openai_agents/testing.py. - text_response / function_call_response build canned generate_content bodies - GeminiTestServer scripts model responses (incl. AFC turns and streaming) and exposes a GoogleGenAIPlugin via .plugin(); records requests for assertions Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * google_genai: stream generate_content_stream via Workflow Streams Set TemporalAsyncClient(streaming_topic=...) and host a WorkflowStream in the workflow's @workflow.init; each generate_content_stream chunk is then published to that topic (as a parsed GenerateContentResponse) as it arrives, so external WorkflowStreamClient consumers observe model output in real time while the workflow runs durably. The workflow's own iteration is unchanged (still batched). - _models: _GeminiApiRequest carries streaming_topic + streaming_batch_interval_ms - TemporalAsyncClient/_TemporalApiClient: streaming_topic + streaming_batch_interval; fail fast (GoogleGenAIError) if a topic is set but no WorkflowStream is hosted - streamed activity publishes via WorkflowStreamClient.from_within_activity; publishing is best-effort and never breaks the batched return - README Streaming section + module docstring; tests in test_gemini_streaming.py Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * google_genai: fix gen-docs cross-reference link targets pydoctor (warnings-as-errors) couldn't resolve the `~`-prefixed `:class:` cross-references added for streaming/testing docstrings. Drop the `~` prefix to use the full dotted path, matching the form already used elsewhere (e.g. openai_agents references temporalio.contrib.workflow_streams.WorkflowStream). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * google_genai: closure-wrap workflow-method tools for config deep-copy google-genai >= 2.8.0 deep-copies the request config internally (config.model_copy(deep=True)), which clones a bound-method tool's __self__ — so a workflow-method tool runs against a throwaway clone and its in-workflow state mutation is silently lost. TemporalAsyncClient now closure-wraps bound-method tools before handing the config to the SDK; copy.deepcopy leaves plain functions intact, so the closure keeps the tool bound to the real workflow instance. Plain functions and activity_as_tool wrappers are unaffected. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * google_genai: migrate interactions/agents to public API; require google-genai 2.10 google-genai 2.9.0 regenerated the vendored interactions client (google.genai._interactions -> google.genai._gaos), removing the private symbols the interactions/agents shims imported (construct_type, AsyncStream, and the AsyncInteractionsResource/AsyncAgentsResource base classes). Retarget the shims to the public google.genai.interactions surface: - TemporalAsyncInteractions/TemporalAsyncAgents are now standalone classes (they already overrode every method), not private-resource subclasses. - _deserialize uses public pydantic only: a TypeAdapter dispatches the InteractionSSEEvent discriminated union (and nested unions), and model_validate rehydrates plain models (recursing nested objects). - _TemporalInteractionAsyncStream is a plain async iterator/context manager, no longer subclassing the SDK's AsyncStream. - Worker-side activities use public types; the stream drain is typed structurally. Bump the pin to google-genai>=2.10.0,<3 (with an exclude-newer-package override so uv can select it past the cutoff) and update the tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Fix lint errors from google-genai 2.10 migration - Mark the Agent import from google.genai.interactions with a reportPrivateImportUsage ignore (pyright can't follow the module's dynamically-built __all__), matching the source modules. - Narrow types in the bound-method tool-wrapping unit tests so pyright and mypy accept attribute/subscript access on the helper results. - Add missing __init__ docstrings flagged by pydocstyle. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Fix ADK multi-agent test mock under google-adk 2.4 google-adk 2.4 rewrites cross-agent history into "For context:" text parts after a transfer, so ResearchModel's dedup-against-history never matched and it re-served "transfer to researcher" to the researcher itself, failing test-latest-deps with "Agent 'researcher' cannot transfer to itself". Key the scripted responses off the calling agent's instruction text instead, which is stateless (replay-safe) and independent of ADK history rewriting. Passes on both the locked google-adk 2.2.0 and latest 2.4.0. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Brian Strauch <brian@brianstrauch.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
1 parent 59c1164 commit 4ec9ab0

25 files changed

Lines changed: 5651 additions & 1028 deletions

.github/CODEOWNERS

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,14 @@
1111
# as well as @temporalio/sdk, so the SDK team can continue to
1212
# manage repo-wide concerns.
1313
/temporalio/contrib/google_adk_agents/ @temporalio/ai-sdk @temporalio/sdk
14+
/temporalio/contrib/google_genai/ @temporalio/ai-sdk @temporalio/sdk
1415
/temporalio/contrib/langgraph/ @temporalio/ai-sdk @temporalio/sdk
1516
/temporalio/contrib/langsmith/ @temporalio/ai-sdk @temporalio/sdk
1617
/temporalio/contrib/openai_agents/ @temporalio/ai-sdk @temporalio/sdk
1718
/temporalio/contrib/strands/ @temporalio/ai-sdk @temporalio/sdk
1819
/temporalio/contrib/workflow_streams/ @temporalio/ai-sdk @temporalio/sdk
1920
/tests/contrib/google_adk_agents/ @temporalio/ai-sdk @temporalio/sdk
21+
/tests/contrib/google_genai/ @temporalio/ai-sdk @temporalio/sdk
2022
/tests/contrib/langgraph/ @temporalio/ai-sdk @temporalio/sdk
2123
/tests/contrib/langsmith/ @temporalio/ai-sdk @temporalio/sdk
2224
/tests/contrib/openai_agents/ @temporalio/ai-sdk @temporalio/sdk

pyproject.toml

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ grpc = ["grpcio>=1.48.2,<2"]
2929
opentelemetry = ["opentelemetry-api>=1.11.1,<2", "opentelemetry-sdk>=1.11.1,<2"]
3030
pydantic = ["pydantic>=2.0.0,<3"]
3131
openai-agents = ["openai-agents>=0.17.5", "mcp>=1.9.4, <2"]
32-
google-adk = ["google-adk>=1.27.0,<2"]
32+
google-adk = ["google-adk>=2.2.0,<3"]
3333
langgraph = ["langgraph>=1.1.0"]
3434
langsmith = ["langsmith>=0.7.34,<0.9"]
3535
lambda-worker-otel = [
@@ -40,6 +40,7 @@ lambda-worker-otel = [
4040
"opentelemetry-sdk-extension-aws>=2.0.0,<3",
4141
]
4242
aioboto3 = ["aioboto3>=10.4.0", "types-aioboto3[s3]>=10.4.0"]
43+
google-genai = ["google-genai>=2.10.0,<3.0.0"]
4344
strands-agents = ["strands-agents>=1.39.0"]
4445

4546
[project.urls]
@@ -88,6 +89,7 @@ dev = [
8889
"async-timeout>=4.0,<6; python_version < '3.11'",
8990
"strands-agents>=1.39.0",
9091
"strands-agents-tools>=0.5.2",
92+
"mcp>=1.9.4,<2",
9193
]
9294

9395
[tool.poe.tasks]
@@ -260,4 +262,4 @@ exclude = ["temporalio/bridge/target/**/*"]
260262
# Prevent uv commands from building the package by default
261263
package = false
262264
exclude-newer = "2 weeks"
263-
exclude-newer-package = { openai-agents = false }
265+
exclude-newer-package = { google-adk = false, google-genai = false, openai-agents = false }
Lines changed: 285 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,285 @@
1+
# Google Gemini SDK Integration for Temporal
2+
3+
> ⚠️ **Experimental.** This integration may change in future versions. Use with
4+
> caution in production.
5+
6+
## Overview
7+
8+
This plugin lets you use the [Google Gemini SDK](https://googleapis.github.io/python-genai/)
9+
(`google-genai`) inside Temporal workflows with durable execution. Every Gemini
10+
API call becomes a **Temporal activity**, so model calls, tool calls, file
11+
operations, interactions, and managed agents are retried, recorded in history,
12+
and survive worker restarts.
13+
14+
Key properties:
15+
16+
- **Credentials never enter the workflow.** The real `genai.Client` lives only
17+
on the worker, inside activities; no API keys or tokens appear in event
18+
history.
19+
- **The SDK's automatic function calling (AFC) loop runs in the workflow**, so
20+
tool wrappers (`activity_as_tool`) work naturally — no manual agent loop.
21+
- **Temporal owns retries.** Configure them via the activity `retry_policy`; the
22+
SDK's own retry loop is rejected to avoid double-retry (see
23+
[Retries & errors](#retries--errors)).
24+
25+
## Install
26+
27+
```bash
28+
uv add temporalio google-genai
29+
# For client-side MCP support, also:
30+
uv add mcp
31+
```
32+
33+
## Hello World
34+
35+
```python
36+
import os
37+
from datetime import timedelta
38+
39+
from google import genai
40+
from google.genai import types
41+
42+
from temporalio import activity, workflow
43+
from temporalio.client import Client
44+
from temporalio.contrib.google_genai import (
45+
GoogleGenAIPlugin,
46+
TemporalAsyncClient,
47+
activity_as_tool,
48+
)
49+
from temporalio.worker import Worker
50+
from temporalio.workflow import ActivityConfig
51+
52+
53+
# ---- a tool, as a normal Temporal activity (runs on the worker) ----
54+
@activity.defn
55+
async def get_weather(city: str) -> str:
56+
return f"It's sunny in {city}."
57+
58+
59+
# ---- the workflow (runs in the Temporal sandbox) ----
60+
@workflow.defn
61+
class WeatherAgent:
62+
@workflow.run
63+
async def run(self, prompt: str) -> str:
64+
client = TemporalAsyncClient()
65+
response = await client.models.generate_content(
66+
model="gemini-2.5-flash",
67+
contents=prompt,
68+
config=types.GenerateContentConfig(
69+
tools=[
70+
activity_as_tool(
71+
get_weather,
72+
activity_config=ActivityConfig(
73+
start_to_close_timeout=timedelta(seconds=30),
74+
),
75+
),
76+
],
77+
),
78+
)
79+
return response.text or ""
80+
81+
82+
# ---- worker setup (outside the sandbox: real client + credentials) ----
83+
async def main() -> None:
84+
gemini = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
85+
plugin = GoogleGenAIPlugin(gemini)
86+
87+
client = await Client.connect("localhost:7233", plugins=[plugin])
88+
async with Worker(
89+
client,
90+
task_queue="gemini",
91+
workflows=[WeatherAgent],
92+
activities=[get_weather],
93+
):
94+
result = await client.execute_workflow(
95+
WeatherAgent.run,
96+
"What's the weather in Tokyo?",
97+
id="weather-1",
98+
task_queue="gemini",
99+
)
100+
print(result)
101+
```
102+
103+
Construct `TemporalAsyncClient` **inside** the workflow; construct the real
104+
`genai.Client` and `GoogleGenAIPlugin` **on the worker**.
105+
106+
## What this plugin gives you
107+
108+
| Surface | Workflow API | Runs as |
109+
| --- | --- | --- |
110+
| Model calls | `client.models.generate_content` / `generate_content_stream` | activity (AFC loop in workflow) |
111+
| Tools | `activity_as_tool(fn, ...)` | one activity per tool call |
112+
| Files | `client.files.upload` / `download` | activity |
113+
| File search | `client.file_search_stores.upload_to_file_search_store` | activity |
114+
| Interactions | `client.interactions.create` / `get` / `cancel` / `delete` | whole-operation activity |
115+
| Managed agents | `client.agents.create` / `get` / `list` / `delete` | whole-operation activity |
116+
| MCP (client-side) | `TemporalMcpClientSession(name)` in `tools=[...]` | `list_tools` / `call_tool` activities |
117+
118+
Streamed responses are batched: the activity drains the stream and the workflow
119+
iterates the collected chunks/events. `client.webhooks` is not supported in
120+
workflows and raises.
121+
122+
## Tool calling
123+
124+
`activity_as_tool` wraps any `@activity.defn` function as a Gemini tool. When the
125+
model calls it, the AFC loop (running in the workflow) dispatches it as a
126+
durable activity:
127+
128+
```python
129+
activity_as_tool(
130+
get_weather,
131+
activity_config=ActivityConfig(start_to_close_timeout=timedelta(seconds=30)),
132+
)
133+
```
134+
135+
A timeout is required — `activity_config` must set `start_to_close_timeout` or
136+
`schedule_to_close_timeout` (Temporal needs one; there is no default for tools).
137+
138+
## MCP support
139+
140+
Client-side MCP (Gemini Developer API) is wired through the plugin: register the
141+
server on the worker and reference it by name in the workflow.
142+
143+
```python
144+
from contextlib import asynccontextmanager
145+
import sys
146+
147+
from mcp import ClientSession, StdioServerParameters
148+
from mcp.client.stdio import stdio_client
149+
150+
from temporalio.contrib.google_genai import TemporalMcpClientSession
151+
152+
153+
# ---- worker: a factory yielding a connected, initialized session ----
154+
@asynccontextmanager
155+
async def weather_mcp():
156+
params = StdioServerParameters(command=sys.executable, args=["weather_server.py"])
157+
async with stdio_client(params) as (read, write):
158+
async with ClientSession(read, write) as session:
159+
await session.initialize()
160+
yield session
161+
162+
163+
plugin = GoogleGenAIPlugin(
164+
genai.Client(api_key=os.environ["GOOGLE_API_KEY"]),
165+
mcp_servers={"weather": weather_mcp},
166+
mcp_connection_idle_timeout=timedelta(minutes=5),
167+
)
168+
169+
170+
# ---- workflow: reference the server by name in the tools list ----
171+
@workflow.defn
172+
class McpAgent:
173+
@workflow.run
174+
async def run(self, prompt: str) -> str:
175+
client = TemporalAsyncClient()
176+
session = TemporalMcpClientSession(
177+
"weather",
178+
activity_config=ActivityConfig(start_to_close_timeout=timedelta(seconds=30)),
179+
)
180+
response = await client.models.generate_content(
181+
model="gemini-2.5-flash",
182+
contents=prompt,
183+
config=types.GenerateContentConfig(tools=[session]),
184+
)
185+
return response.text or ""
186+
```
187+
188+
The MCP connection lives on the worker (pooled, idle-evicted); the workflow only
189+
carries the server name. Tool discovery and calls run as `{name}-list-tools` /
190+
`{name}-call-tool` activities, so the full tool parameter schema reaches the
191+
model. Set `cache_tools=True` to list a server's tools once per workflow instead
192+
of per turn.
193+
194+
## Streaming
195+
196+
`generate_content_stream` works as usual — the workflow iterates chunks (batched
197+
from the activity). To let an **external** consumer (a chat UI) observe chunks in
198+
real time while the workflow runs durably, set `streaming_topic` on the client
199+
and host a [`WorkflowStream`](../workflow_streams/) in the workflow. Each
200+
streamed `GenerateContentResponse` is published to that topic as it arrives:
201+
202+
```python
203+
from temporalio.contrib.workflow_streams import WorkflowStream
204+
205+
206+
@workflow.defn
207+
class StreamingAgent:
208+
@workflow.init
209+
def __init__(self, prompt: str) -> None:
210+
self.stream = WorkflowStream() # required when streaming_topic is set
211+
212+
@workflow.run
213+
async def run(self, prompt: str) -> str:
214+
client = TemporalAsyncClient(streaming_topic="gemini")
215+
text = []
216+
async for chunk in await client.models.generate_content_stream(
217+
model="gemini-2.5-flash", contents=prompt,
218+
):
219+
text.append(chunk.text or "")
220+
return "".join(text)
221+
```
222+
223+
Consume the stream from outside the workflow:
224+
225+
```python
226+
from temporalio.contrib.workflow_streams import WorkflowStreamClient
227+
228+
229+
async def consume(client, workflow_id):
230+
stream = WorkflowStreamClient.create(client, workflow_id)
231+
async for item in stream.subscribe(
232+
["gemini"], result_type=types.GenerateContentResponse,
233+
):
234+
print(item.data.text, end="", flush=True)
235+
```
236+
237+
The workflow's own iteration is unchanged (it still receives batched chunks for
238+
the SDK to parse); the topic is purely for external real-time observation. If
239+
`streaming_topic` is set but the workflow hosts no `WorkflowStream`, the call
240+
raises `GoogleGenAIError`. Tune flush cadence with
241+
`TemporalAsyncClient(streaming_topic=..., streaming_batch_interval=...)`
242+
(default 100ms).
243+
244+
## Retries & errors
245+
246+
Temporal owns retries. Configure them with the activity `retry_policy` via
247+
`activity_config`. The plugin **rejects** the SDK's own retry config so retries
248+
don't compound:
249+
250+
- Constructing the plugin with a `genai.Client` that has
251+
`http_options.retry_options` raises `ValueError`.
252+
- Setting `http_options.retry_options` on a per-request call raises
253+
`GoogleGenAIError`.
254+
255+
API-call activities classify failures: transient statuses (408, 429, 5xx) stay
256+
retryable (the activity's `retry_policy` applies); other statuses (e.g. 4xx) are
257+
non-retryable so the workflow fails fast.
258+
259+
## Vertex AI
260+
261+
Pass `vertexai=True` to both the worker-side `genai.Client` and the
262+
workflow-side `TemporalAsyncClient`. On the workflow side you must also set
263+
`project` and `location` **explicitly**:
264+
265+
```python
266+
# worker
267+
genai.Client(vertexai=True, project="my-project", location="us-central1")
268+
269+
# workflow
270+
TemporalAsyncClient(vertexai=True, project="my-project", location="us-central1")
271+
```
272+
273+
Normally the SDK auto-discovers `project`/`location` from the environment
274+
(credentials, ADC, metadata server). That discovery
275+
would be non-deterministic and break replay. Setting them by hand
276+
keeps it deterministic.
277+
278+
## Composing with other plugins
279+
280+
`GoogleGenAIPlugin` is a `temporalio.plugin.SimplePlugin`; pass it in the
281+
`plugins=[...]` list alongside others (e.g. OpenTelemetry). It contributes a
282+
Pydantic data converter, the Gemini activities, a sandbox-passthrough config for
283+
`google.genai` (and `mcp`), and registers `GoogleGenAIError` as a workflow
284+
failure type. When composing data converters, construct the plugins so their
285+
converters are compatible.

0 commit comments

Comments
 (0)