Add First-Class Google Gen AI SDK Integration to Contrib

[JasonSteving99]

First-class Temporal integration for the Google Gen AI SDK

Adds temporalio.contrib.google_genai, letting you use the Gemini SDK's
AsyncClient directly inside Temporal workflows. Every Gemini API call, tool
call, file operation, interaction, managed-agent operation, and MCP call becomes
a durable Temporal activity — crash recovery, full event-history visibility,
and replay safety — while credentials stay entirely on the worker.

⚠️ Experimental; the API may change in future versions.

Key properties

• Credentials never enter the workflow. The real genai.Client lives only on
  the worker, inside activities; no API keys or tokens appear in event history.
• The SDK's automatic function calling (AFC) loop runs in the workflow, so
  activity_as_tool wrappers work with no manual agent loop.
• Temporal owns retries. The SDK's own retry config is rejected (not silently
  overridden); use the activity retry_policy. API errors are classified
  retryable/non-retryable.

What's supported

Surface	Workflow API	Runs as
Model calls	client.models.generate_content / generate_content_stream	activity (AFC loop in-workflow)
Tools	activity_as_tool(fn, ...) or a plain workflow method	activity / in-workflow
Files	client.files.upload / download / register_files	activity
File search	client.file_search_stores.*	activity
Interactions	client.interactions.create / get / cancel / delete (+ streamed)	whole-operation activity
Managed agents	client.agents.create / get / list / delete	whole-operation activity
MCP (client-side)	TemporalMcpClientSession(name) in tools=[...]	{name}-list-tools / {name}-call-tool activities
Streaming → external consumers	streaming_topic= + WorkflowStream	published to a workflow-stream topic

client.webhooks is not supported in workflows and raises.

How it works

The integration shims the SDK at the layers where I/O happens, so workflows use
the native client surface unchanged:

• TemporalAsyncClient subclasses AsyncClient and injects a private
  _TemporalApiClient (a BaseApiClient whose async_request dispatches HTTP
  through workflow.execute_activity instead of the network). The SDK's
  request-formatting and AFC loop run in the workflow; only the raw HTTP call
  crosses into an activity. models, tunings, caches, batches, tokens,
  etc. are inherited unchanged.
• Files / file search are replaced with shims that run upload/download/
  register as whole activities (avoiding filesystem + token refresh in the
  sandbox).
• Interactions and managed agents bypass BaseApiClient via a vendored
  client, so each operation is routed as a whole through its own activity;
  streamed interactions are drained in the activity and replayed to the workflow.
• GoogleGenAIPlugin (SimplePlugin) registers the activities, installs a
  Pydantic data converter, configures sandbox passthrough (google.genai,
  mcp, pydantic_core, annotated_types), and registers GoogleGenAIError
  as a workflow-failure type. You pass a fully-configured genai.Client; the
  plugin never builds one.

Tool calling

activity_as_tool wraps any @activity.defn as a Gemini tool; the AFC loop
(in-workflow) dispatches it as a durable activity. A timeout is required (no
default — matching openai_agents/strands). Plain workflow methods can also be
passed as tools and run in-workflow.

MCP

• Client-side (Gemini Developer API): register a server on the worker —
  GoogleGenAIPlugin(client, mcp_servers={name: factory}) — and place
  TemporalMcpClientSession(name) in a generate_content tools list. The
  SDK's AFC loop drives it; list_tools/call_tool run as activities against a
  pooled, idle-evicted worker-side connection. The full tool parameter schema
  reaches the model.
• Server-side (Vertex Tool(mcp_servers=[McpServer(...)])) and Interactions
  API MCP steps are executed by Google's backend and flow through unchanged.

mcp is an optional dependency (lazy-imported; TemporalMcpClientSession is
exposed lazily so importing the package doesn't require it).

Streaming

Set TemporalAsyncClient(streaming_topic="...") and host a
temporalio.contrib.workflow_streams.WorkflowStream in @workflow.init. Each
generate_content_stream chunk is published to that topic (as a parsed
GenerateContentResponse) as it arrives, so an external WorkflowStreamClient
can observe model output in real time while the workflow runs durably. The
workflow's own iteration is unchanged (still batched for the SDK).

Testing utilities

temporalio.contrib.google_genai.testing ships text_response /
function_call_response builders and GeminiTestServer, which scripts model
responses and returns a GoogleGenAIPlugin — so users can test Gemini workflows
(including AFC tool loops and streaming) without a real LLM endpoint.

Usage

# ---- worker (real client + credentials) ----
gemini = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
client = await Client.connect("localhost:7233", plugins=[GoogleGenAIPlugin(gemini)])
async with Worker(client, task_queue="gemini", workflows=[WeatherAgent],
                  activities=[get_weather]):
    ...

# ---- workflow ----
@workflow.defn
class WeatherAgent:
    @workflow.run
    async def run(self, prompt: str) -> str:
        client = TemporalAsyncClient()
        resp = await client.models.generate_content(
            model="gemini-2.5-flash",
            contents=prompt,
            config=types.GenerateContentConfig(tools=[
                activity_as_tool(get_weather, activity_config=ActivityConfig(
                    start_to_close_timeout=timedelta(seconds=30))),
            ]),
        )
        return resp.text

Testing

58 tests (test_gemini.py, test_gemini_mcp.py, test_gemini_streaming.py),
including:

• generate_content, AFC tool calling (single/multi-arg, workflow-method,
  sequential multi-tool, failure propagation), batched streaming, per-request
  http_options, multi-turn chat
• files (str path + io.BytesIO), download, file-search upload, full end-to-end
  with the real activities + a mocked client
• interactions (create/stream/get/cancel/delete) and managed agents CRUD
• MCP: discovery + call, connection pooling, full-schema propagation, replay,
  exact activity-scheduling counts; Vertex/interactions MCP pass-through
• streaming to a WorkflowStream + fail-fast when no stream is hosted
• replay determinism (Replayer with the plugin) and side-effects
  (max_cached_workflows=0, exact ActivityTaskScheduled counts)
• retry rejection (client- and request-level) and error classification

Notes

• google-genai is pinned <2.8.0: 2.8.0 regressed in-workflow AFC for plain
  workflow-method tools (the tool runs but its state mutation isn't visible on
  replay/query). Lift once a fixed release ships upstream.
• On Vertex AI, pass project/location explicitly to TemporalAsyncClient —
  the SDK's environment auto-discovery can't run deterministically in the
  sandbox.
• A determinism survey found no time/uuid/random on the in-workflow
  generate/AFC path; the one in-workflow exception (batches.create auto-naming
  on Vertex) is documented.

[CLAassistant]

All committers have signed the CLA.

[semgrep-managed-scans]

Semgrep found 1 ssc-80bc2470-4c90-4cf6-811e-5a265b8f01d5 finding:

• uv.lock
  • L2102 - Triage

Risk: Affected versions of litellm are vulnerable to Exposure of Sensitive Information to an Unauthorized Actor / Use of Password Hash With Insufficient Computational Effort / Use of a Broken or Risky Cryptographic Algorithm. LiteLLM exposes password hashes via authenticated API endpoints and its login flow accepts a stored SHA-256 hash as a valid password, allowing an authenticated user to steal another user's hash and log in as that user, resulting in full authentication bypass and privilege escalation.

Fix: Upgrade this library to at least version 1.83.0 at sdk-python/uv.lock:2102.

Reference(s): GHSA-69x8-hrgq-fjj8

Semgrep found 1 ssc-9ebb5c4b-d829-4496-b42b-76424d99a7b1 finding:

• uv.lock
  • L2102 - Triage

Risk: Affected versions of litellm are vulnerable to Incorrect Authorization. This vulnerability stems from inadequate enforcement of access control policies, allowing authenticated users to perform actions beyond their intended privilege level and potentially alter sensitive system configurations or access restricted resources.

Fix: Upgrade this library to at least version 1.83.0 at sdk-python/uv.lock:2102.

Reference(s): GHSA-53mr-6c8q-9789, CVE-2026-35029

Semgrep found 1 ssc-4464d88f-a369-4669-9a83-d26a12248ef5 finding:

• uv.lock
  • L982 - Triage

Risk: Affected versions of google-adk are vulnerable to Missing Authentication for Critical Function. Google Agent Development Kit (ADK) contains a missing authentication flaw that can let an unauthenticated remote attacker reach code-injection paths and execute arbitrary code on the server running the ADK instance, including ADK Web deployments on Python, Cloud Run, and GKE.

Fix: Upgrade this library to at least version 1.28.1 at sdk-python/uv.lock:982.

Reference(s): GHSA-rg7c-g689-fr3x, CVE-2026-4810

[Copilot]

Pull request overview

Adds a first-class Temporal contrib integration for the Google Gemini (google-genai) SDK so workflows can use an AsyncClient while executing all SDK HTTP calls and tool invocations as durable Temporal activities (credentials stay on the worker).

Changes:

• Introduces temporalio.contrib.google_gemini_sdk with a worker plugin, workflow helpers, activity implementations, and serializable request/response models.
• Adds Temporal-aware shims for Gemini file operations and streamed responses.
• Adds a comprehensive integration test suite and wires the optional google-genai dependency via a new extra.

Reviewed changes

Copilot reviewed 13 out of 14 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
uv.lock	Adds google-genai dependency via a new google-gemini extra and updates lock metadata.
pyproject.toml	Defines google-gemini optional dependency group (google-genai>=1.66.0).
tests/contrib/google_gemini_sdk/init.py	Test package marker for the new contrib integration.
tests/contrib/google_gemini_sdk/test_gemini.py	Integration + unit tests for API calls, streaming, file ops, tool calling, and plugin wiring.
temporalio/contrib/google_gemini_sdk/init.py	Public module entry points and docs for GeminiPlugin, gemini_client, and activity_as_tool.
temporalio/contrib/google_gemini_sdk/workflow.py	Workflow-side helpers: gemini_client() and activity_as_tool() wrapper.
temporalio/contrib/google_gemini_sdk/justfile	Local dev commands/examples for running workers and sample workflows.
temporalio/contrib/google_gemini_sdk/_models.py	Pydantic models that cross the activity boundary (requests/responses/options).
temporalio/contrib/google_gemini_sdk/_temporal_api_client.py	Workflow-side BaseApiClient shim that routes SDK HTTP calls through activities.
temporalio/contrib/google_gemini_sdk/_temporal_async_client.py	AsyncClient subclass wiring Temporal-aware files and file_search_stores.
temporalio/contrib/google_gemini_sdk/_temporal_files.py	Temporal-aware AsyncFiles overriding upload/download/register to run on the worker.
temporalio/contrib/google_gemini_sdk/_temporal_file_search_stores.py	Temporal-aware AsyncFileSearchStores overriding upload to run on the worker.
temporalio/contrib/google_gemini_sdk/_gemini_activity.py	Worker-side activities that execute real SDK calls using the provided genai.Client.
temporalio/contrib/google_gemini_sdk/_gemini_plugin.py	GeminiPlugin that registers activities, configures the data converter, and sandbox passthrough.

---

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