temporalio.contrib.google

package documentation

(source)

First-class Temporal integration for the Google Gemini SDK.

Warning

This module is experimental and may change in future versions. Use with caution in production environments.

This integration lets you use the Gemini SDK's async client with full automatic function calling (AFC) support. Every API call becomes a durable Temporal activity. Tools default to plain workflow methods that run deterministically in-workflow; wrap any @activity.defn with activity_as_tool to run a tool as a durable activity instead.

No credentials are fetched in the workflow, and no auth material appears in Temporal's event history.

GoogleGenAIPlugin — registers the Gemini SDK activities using a caller-provided genai.Client on the worker side.
TemporalAsyncClient — construct from a workflow to get an AsyncClient that routes API calls through activities.
activity_as_tool — convert any @activity.defn function into a Gemini tool callable; Gemini's AFC invokes it as a Temporal activity.

The Interactions API (client.interactions) and managed agents (client.agents) are supported as whole-operation activities; streamed interactions are batched (the activity drains the SSE stream and the workflow iterates the collected events). client.webhooks is not supported in workflows. The Interactions API has no automatic function calling: declare tools as {"type": "function", ...} dicts (per the Gemini docs) and drive the tool loop yourself, executing each call via workflow.execute_activity or an activity_as_tool callable.

MCP is supported across three paths. Client-side MCP (Gemini Developer API) uses TemporalMcpClientSession: register a server with GoogleGenAIPlugin(mcp_servers={name: factory}) on the worker, then place TemporalMcpClientSession(name) in a generate_content tools list — the SDK's AFC loop drives it, with list_tools / call_tool running as activities against a pooled worker-side connection. Server-side MCP on Vertex AI (Tool(mcp_servers=[McpServer(...)])) and the Interactions API's MCP step types are executed by Google's backend and flow through unchanged as request / response data — no extra wiring needed. Client-side MCP requires the mcp package.

Streaming: set TemporalAsyncClient(streaming_topic=...) and host a temporalio.contrib.workflow_streams.WorkflowStream in the workflow's @workflow.init. Each generate_content_stream chunk is then published to that topic as it arrives, so external consumers can observe model output in real time while the workflow runs durably; the workflow's own iteration is unchanged.

Quickstart:

# ---- worker setup (outside the Temporal Python Sandbox) ----
client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
plugin = GoogleGenAIPlugin(client)

@activity.defn
async def get_weather(state: str) -> str: ...

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

Module	`testing`	Testing utilities for the Google Gemini SDK Temporal integration.
Module	`workflow`	Workflow utilities for Google Gemini SDK integration with Temporal.
Module	`_errors`	Error types for the Google Gemini SDK Temporal integration.
Module	`_gemini_activity`	Temporal activity that executes Gemini SDK API calls with real credentials.
Module	`_google_genai_plugin`	Temporal plugin for Google Gemini SDK integration.
Module	`_mcp`	Worker-side MCP activities and a pooled-connection subsystem.
Module	`_models`	Serializable Pydantic models for the Gemini SDK Temporal integration.
Module	`_temporal_agents`	Temporal-aware agents resource shim.
Module	`_temporal_api_client`	Temporal-aware BaseApiClient that routes SDK calls through activities.
Module	`_temporal_async_client`	Temporal-aware `AsyncClient`.
Module	`_temporal_file_search_stores`	Temporal-aware AsyncFileSearchStores shim.
Module	`_temporal_files`	Temporal-aware AsyncFiles shim.
Module	`_temporal_interactions`	Temporal-aware interactions resource shim.
Module	`_temporal_mcp`	Temporal-aware `mcp.ClientSession` shim.

From __init__.py:

Class	`GoogleGenAIPlugin`	A Temporal Worker Plugin configured for the Google Gemini SDK.
Class	`TemporalAsyncClient`	An `AsyncClient` whose API calls run as Temporal activities.
Class	`TemporalMcpClientSession`	`mcp.ClientSession` whose tool discovery and calls run as activities.
Exception	`GoogleGenAIError`	Error raised by the Google Gemini Temporal integration.
Function	`__getattr__`	Lazily expose `TemporalMcpClientSession` without importing `mcp` eagerly.
Function	`activity_as_tool`	Convert a Temporal activity into a Gemini-compatible async tool callable.

def __getattr__(name: str) -> Any: (source) ¶

Lazily expose TemporalMcpClientSession without importing mcp eagerly.

mcp is an optional dependency, so importing this package must not require it; the import (and any resulting ImportError) is deferred until the symbol is actually accessed.

def activity_as_tool(fn: Callable, *, activity_config: ActivityConfig | None = None) -> Callable: (source) ¶

Convert a Temporal activity into a Gemini-compatible async tool callable.

Warning

This API is experimental and may change in future versions. Use with caution in production environments.

Returns an async callable with the same name, docstring, and type signature as fn. When Gemini's automatic function calling (AFC) invokes the returned callable from within a Temporal workflow, the call is executed as a Temporal activity via workflow.execute_activity. Each tool invocation therefore appears as a separate, durable entry in the workflow event history.

Because AFC is left enabled, the Gemini SDK owns the agentic loop — no manual while loop or run_agent() helper is required. Pass the returned callable directly to GenerateContentConfig(tools=[...]).

Parameters
fn:`Callable`	A Temporal activity function decorated with `@activity.defn`.
activity_config:`ActivityConfig \| None`	Configuration for the activity execution (timeouts, retry policy, etc.). Must set `start_to_close_timeout` or `schedule_to_close_timeout` — Temporal requires one, and there is no default; otherwise the tool call raises when the activity is invoked.

Returns
`Callable`	An async callable suitable for use as a Gemini tool.

Raises
`GoogleGenAIError`	If `fn` is not decorated with `@activity.defn` or has no activity name.

temporalio.contrib.google_genai

`temporalio.contrib.google_genai`