Google GenAI integration

View Markdown

Temporal's Google GenAI integration lets you call Gemini models from inside Temporal Workflows, so a sequence of model calls keeps its place across Worker restarts, deploys, and transient failures.

Temporal gives your code Durable Execution. The Google Gen AI SDK gives you the model API: content generation, automatic function calling, chat sessions, structured output, files, and MCP. The integration connects the two so that you write ordinary Gemini SDK code and run it as a Workflow, without writing your own retry loop or checkpointing.

GoogleGenAIPlugin is what ties them together. You build a genai.Client with your credentials on the Worker and hand it to the plugin. Inside the Workflow you construct a TemporalAsyncClient, which has the same shape as the SDK's async client but routes every API call through a Temporal Activity. Each call gets its own timeout, retry policy, and Event History entry, and your credentials stay on the Worker.

Code snippets in this guide come from the Google GenAI plugin samples. Refer to the samples for the complete code.

Prerequisites

This guide assumes you are already familiar with the Google Gen AI SDK. If you aren't, refer to the Gemini API documentation for more details.
If you are new to Temporal, we recommend reading Understanding Temporal or taking the Temporal 101 course.
Ensure you have set up your local development environment by following the Set up your local development environment guide. When you're done, leave the Temporal development server running if you want to test your code locally.

Install the plugin

Install the Temporal Python SDK with Google GenAI support (requires temporalio 1.31.0 or later):

uv add "temporalio[google-genai]>=1.31.0"

If you use pip:

pip install "temporalio[google-genai]>=1.31.0"

The MCP path also needs the mcp package, which the extra does not install.

Call a model from a Workflow

Construct a TemporalAsyncClient in your Workflow and call it the way you would call genai.Client.aio. The client takes no credentials — it resolves each call to an Activity that runs on the Worker.

google_genai/hello_world/workflow.py

from temporalio import workflow
from temporalio.contrib.google_genai import TemporalAsyncClient


@workflow.defn
class HelloWorldWorkflow:
    @workflow.run
    async def run(self, prompt: str) -> str:
        client = TemporalAsyncClient()
        response = await client.models.generate_content(
            model="gemini-2.5-flash",
            contents=prompt,
        )
        return response.text or ""

On the Worker, build a real genai.Client with your credentials, wrap it in a GoogleGenAIPlugin, and pass the plugin to Client.connect. A Worker created from that Temporal Client picks up the plugin, which registers the Activity that makes the API calls and swaps in the Pydantic payload converter that serializes Gemini types.

google_genai/hello_world/run_worker.py

import asyncio
import os

from google import genai
from temporalio.client import Client
from temporalio.contrib.google_genai import GoogleGenAIPlugin
from temporalio.worker import Worker

from google_genai.hello_world.workflow import HelloWorldWorkflow


async def main() -> None:
    # The real genai.Client (with credentials) lives only on the worker.
    genai_client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
    plugin = GoogleGenAIPlugin(genai_client)

    client = await Client.connect(
        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
        plugins=[plugin],
    )

    worker = Worker(
        client,
        task_queue="google-genai-hello-world",
        workflows=[HelloWorldWorkflow],
    )
    print("Worker started. Ctrl+C to exit.")
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

Because API calls run as Activities, the Worker process is the one that needs credentials. The samples use the Gemini Developer API, which reads its key from the GOOGLE_API_KEY environment variable.

export GOOGLE_API_KEY="your-api-key"
uv run google_genai/hello_world/run_worker.py

Start the Workflow the way you start any other Temporal Workflow. The starting Client does not need the plugin.

google_genai/hello_world/run_workflow.py

import asyncio
import os

from temporalio.client import Client

from google_genai.hello_world.workflow import HelloWorldWorkflow


async def main() -> None:
    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))

    result = await client.execute_workflow(
        HelloWorldWorkflow.run,
        "Write a haiku about durable execution.",
        id="google-genai-hello-world",
        task_queue="google-genai-hello-world",
    )

    print(f"Result: {result}")


if __name__ == "__main__":
    asyncio.run(main())

Run tools as Activities

The Gemini SDK's automatic function calling loop runs inside the Workflow, so you don't write the tool loop yourself. A tool can be either of the following:

A Temporal Activity wrapped with activity_as_tool. The model's call to it runs as its own Activity, with its own timeout and retries, and appears in the Event History. Use this for anything that does I/O or is otherwise non-deterministic.
A plain Workflow method passed directly. It runs in the Workflow with no Activity dispatch, so it must be deterministic.

This Workflow passes one of each on a single call.

google_genai/tools/workflow.py

from datetime import timedelta

from google.genai import types
from temporalio import activity, workflow
from temporalio.contrib.google_genai import TemporalAsyncClient, activity_as_tool
from temporalio.workflow import ActivityConfig


@activity.defn
async def get_weather(city: str) -> str:
    """Look up the current weather for a city."""
    # Stub — replace with a real HTTP call in production.
    return f"It's 72F and sunny in {city}."


@workflow.defn
class ToolsWorkflow:
    @workflow.run
    async def run(self, prompt: str) -> str:
        client = TemporalAsyncClient()
        response = 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),
                        ),
                    ),
                    self.recommend_activity,
                ],
            ),
        )
        return response.text or ""

    async def recommend_activity(self, weather: str) -> str:
        """Recommend something to do given a weather description."""
        if "sunny" in weather.lower():
            return "Go for a hike."
        return "Visit a museum."

activity_as_tool keeps the wrapped function's name, docstring, and type signature, which is what the model uses to decide when to call it. Its activity_config must set start_to_close_timeout or schedule_to_close_timeout; there is no default, and the tool call fails without one.

google_genai/tools/run_worker.py

worker = Worker(
    client,
    task_queue="google-genai-tools",
    workflows=[ToolsWorkflow],
    activities=[get_weather],
)

Hold a multi-turn conversation

client.chats works inside a Workflow. The chat session keeps its history in Workflow state, and each send_message call runs as its own Activity, so a conversation that spans hours or days survives a Worker restart.

google_genai/chat/workflow.py

from temporalio import workflow
from temporalio.contrib.google_genai import TemporalAsyncClient


@workflow.defn
class ChatWorkflow:
    @workflow.run
    async def run(self, prompts: list[str]) -> list[str]:
        client = TemporalAsyncClient()
        chat = client.chats.create(model="gemini-2.5-flash")
        replies: list[str] = []
        for prompt in prompts:
            response = await chat.send_message(prompt)
            replies.append(response.text or "")
        return replies

To drive the turns from the outside instead of from a list, take each prompt as a Signal and wait on it with workflow.wait_condition.

Return structured output

The plugin installs Temporal's Pydantic payload converter, so a Pydantic model passes through Temporal payloads unchanged. Pass the model as response_schema and read the parsed result from response.parsed.

google_genai/structured_output/workflow.py

from google.genai import types
from pydantic import BaseModel
from temporalio import workflow
from temporalio.contrib.google_genai import TemporalAsyncClient


class Recipe(BaseModel):
    name: str
    ingredients: list[str]
    steps: list[str]


@workflow.defn
class StructuredOutputWorkflow:
    @workflow.run
    async def run(self, prompt: str) -> Recipe:
        client = TemporalAsyncClient()
        response = await client.models.generate_content(
            model="gemini-2.5-flash",
            contents=prompt,
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                response_schema=Recipe,
            ),
        )
        recipe = response.parsed
        assert isinstance(recipe, Recipe)
        return recipe

Use MCP tool servers

To give a model tools from an MCP server, register the server on the Worker and reference it by name in the Workflow. Connecting to an MCP server is external I/O, so the plugin holds the connection on the Worker and runs list_tools and call_tool as Activities against it.

google_genai/mcp/run_worker.py

@asynccontextmanager
async def echo_session() -> AsyncIterator[ClientSession]:
    """Yield a connected, initialized session to the stdio echo MCP server."""
    params = StdioServerParameters(command=sys.executable, args=[ECHO_SERVER])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session


async def main() -> None:
    genai_client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
    plugin = GoogleGenAIPlugin(genai_client, mcp_servers={"echo": echo_session})

In the Workflow, pass a TemporalMcpClientSession with the same name in the tools list. Automatic function calling discovers and calls the server's tools from there.

google_genai/mcp/workflow.py

from datetime import timedelta

from google.genai import types
from temporalio import workflow
from temporalio.contrib.google_genai import (
    TemporalAsyncClient,
    TemporalMcpClientSession,
)
from temporalio.workflow import ActivityConfig


@workflow.defn
class McpWorkflow:
    @workflow.run
    async def run(self, prompt: str) -> str:
        client = TemporalAsyncClient()
        session = TemporalMcpClientSession(
            "echo",
            cache_tools=True,
            activity_config=ActivityConfig(
                start_to_close_timeout=timedelta(seconds=30),
            ),
        )
        response = await client.models.generate_content(
            model="gemini-2.5-flash",
            contents=prompt,
            config=types.GenerateContentConfig(tools=[session]),
        )
        return response.text or ""

cache_tools=True reuses the first list_tools result for the rest of the run instead of listing tools before every call. The Worker keeps each MCP connection open between uses and disconnects it after five minutes idle; change that with mcp_connection_idle_timeout on the plugin.

Server-side MCP needs no wiring. Vertex AI's Tool(mcp_servers=[McpServer(...)]) and the Interactions API's MCP steps run on Google's backend, so they pass through as ordinary request and response data.

Stream model output

generate_content_stream can forward chunks to an external subscriber while the Workflow is still running. Set streaming_topic on the client and host a WorkflowStream in the Workflow's @workflow.init; each chunk is published to that topic as it arrives. The Workflow's own iteration over the stream is unchanged.

google_genai/streaming/workflow.py

from temporalio import workflow
from temporalio.contrib.google_genai import TemporalAsyncClient
from temporalio.contrib.workflow_streams import WorkflowStream


@workflow.defn
class StreamingWorkflow:
    @workflow.init
    def __init__(self, prompt: str) -> None:
        # Hosting a WorkflowStream is required when streaming_topic is set.
        self.stream = WorkflowStream()
        self._done = False

    @workflow.run
    async def run(self, prompt: str) -> str:
        client = TemporalAsyncClient(streaming_topic="gemini")
        chunks: list[str] = []
        async for chunk in await client.models.generate_content_stream(
            model="gemini-2.5-flash",
            contents=prompt,
        ):
            chunks.append(chunk.text or "")
        await workflow.wait_condition(lambda: self._done)
        return "".join(chunks)

    @workflow.signal
    def finish(self) -> None:
        self._done = True

A consumer subscribes to the topic with WorkflowStreamClient. The published chunks are Pydantic GenerateContentResponse objects, so the subscribing Client needs pydantic_data_converter.

google_genai/streaming/run_workflow.py

# Subscribe to the "gemini" topic and print chunks as the model produces them.
stream = WorkflowStreamClient.create(client, workflow_id)
async for item in stream.subscribe(
    ["gemini"],
    from_offset=0,
    result_type=types.GenerateContentResponse,
    poll_cooldown=timedelta(milliseconds=50),
):
    chunk: types.GenerateContentResponse = item.data
    if chunk.text:
        print(chunk.text, end="", flush=True)
    if chunk.candidates and chunk.candidates[0].finish_reason:
        print()
        break

The streaming Activity batches published chunks and flushes them every 100 milliseconds by default. Adjust that with streaming_batch_interval. Delivery is at-least-once per Activity attempt: if the streaming Activity retries, the model call re-runs and republishes, so subscribers should tolerate duplicates and treat the Workflow result as the source of truth.

Upload files and reference them in a prompt

client.files runs as Activities too, so the file is read on the Worker rather than in the Workflow. Upload it, then pass the returned handle in contents.

google_genai/files/workflow.py

from typing import cast

from google.genai import types
from temporalio import workflow
from temporalio.contrib.google_genai import TemporalAsyncClient


@workflow.defn
class FilesWorkflow:
    @workflow.run
    async def run(self, file_path: str, prompt: str) -> str:
        client = TemporalAsyncClient()
        uploaded = await client.files.upload(
            file=file_path,
            config=types.UploadFileConfig(mime_type="text/plain"),
        )
        contents = cast(types.ContentListUnion, [prompt, uploaded])
        response = await client.models.generate_content(
            model="gemini-2.5-flash",
            contents=contents,
        )
        return response.text or ""

The file path resolves on the Worker, so the Worker needs access to it. Operations that require separate Google Cloud credentials, such as files.register_files, use the extra_credentials you pass to the plugin.

Use the Interactions API and managed agents

client.interactions and client.agents are server-managed: the state lives on Google's backend and each operation runs as its own Activity.

google_genai/interactions/workflow.py

from typing import Any

from temporalio import workflow
from temporalio.contrib.google_genai import TemporalAsyncClient


@workflow.defn
class InteractionsWorkflow:
    @workflow.run
    async def run(self, prompt: str) -> dict[str, Any]:
        client = TemporalAsyncClient()

        # create/get return either an Interaction or a streaming response; without
        # stream=True the result is always an Interaction.
        interaction: Any = await client.interactions.create(
            model="gemini-2.5-flash",
            input=prompt,
        )
        fetched: Any = await client.interactions.get(interaction.id)
        await client.interactions.delete(interaction.id)

        return {"id": interaction.id, "status": str(fetched.status)}

Two limits apply to the Interactions API:

It has no automatic function calling. Declare tools as {"type": "function", ...} dicts and drive the tool loop yourself, running each call with workflow.execute_activity or an activity_as_tool callable.
Streamed interactions are batched. The Activity drains the server-sent event stream and the Workflow iterates the collected events.

client.webhooks is not supported in Workflows.

Run against Vertex AI

To use Vertex AI instead of the Gemini Developer API, set vertexai=True on both sides. In the Workflow, pass the project and location as Workflow arguments rather than reading environment variables, which keeps the Workflow deterministic.

google_genai/vertex_ai/workflow.py

from temporalio import workflow
from temporalio.contrib.google_genai import TemporalAsyncClient


@workflow.defn
class VertexAIWorkflow:
    @workflow.run
    async def run(self, prompt: str, project: str, location: str) -> str:
        client = TemporalAsyncClient(
            vertexai=True,
            project=project,
            location=location,
        )
        response = await client.models.generate_content(
            model="gemini-2.5-flash",
            contents=prompt,
        )
        return response.text or ""

The Worker's genai.Client uses Application Default Credentials instead of an API key. Run gcloud auth application-default login, or set GOOGLE_APPLICATION_CREDENTIALS to a service account key file.

google_genai/vertex_ai/run_worker.py

genai_client = genai.Client(
    vertexai=True,
    project=os.environ["GOOGLE_CLOUD_PROJECT"],
    location=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"),
)
plugin = GoogleGenAIPlugin(genai_client)

The vertexai setting must match on both sides. A Workflow that sets vertexai=True against a Worker configured for the Gemini Developer API sends requests the backend can't serve.

Set timeouts and retries

Every API call defaults to a 60-second start_to_close_timeout and Temporal's default retry policy. Override that for all of a client's calls with activity_config:

from datetime import timedelta

from temporalio.common import RetryPolicy
from temporalio.contrib.google_genai import TemporalAsyncClient
from temporalio.workflow import ActivityConfig

client = TemporalAsyncClient(
    activity_config=ActivityConfig(
        start_to_close_timeout=timedelta(minutes=5),
        retry_policy=RetryPolicy(maximum_attempts=3),
    ),
)

activity_as_tool and TemporalMcpClientSession take their own activity_config, so tool calls and MCP calls can use different limits than model calls.

Let Temporal own retries. The plugin rejects a genai.Client configured with http_options.retry_options, because an SDK-internal retry loop hides its attempts inside a single Activity and compounds with the Temporal retry policy.

Samples

The Google GenAI plugin samples cover each pattern as a self-contained, runnable scenario:

hello_world: one generate_content call.
tools: an Activity tool and a Workflow-method tool on the same call.
streaming: chunks forwarded to an external subscriber with WorkflowStream.
chat: a multi-turn conversation with client.chats.
structured_output: typed JSON output through a Pydantic model.
mcp: MCP tools run as Activities, with a self-contained echo server.
files: a file uploaded with client.files and referenced in a prompt.
interactions: server-managed conversations with client.interactions.
agents: managed agent create, get, list, and delete with client.agents.
vertex_ai: the same hello-world flow against Vertex AI.

Prerequisites​

Install the plugin​

Call a model from a Workflow​

Run tools as Activities​

Hold a multi-turn conversation​

Return structured output​

Use MCP tool servers​

Stream model output​

Upload files and reference them in a prompt​

Use the Interactions API and managed agents​

Run against Vertex AI​

Set timeouts and retries​

Samples​