# Google GenAI integration

Temporal's Google GenAI integration lets you call [Gemini](https://ai.google.dev/gemini-api/docs) 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](/temporal#durable-execution). The
[Google Gen AI SDK](https://googleapis.github.io/python-genai/) 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.

> **Pre-release**

Code snippets in this guide come from the
[Google GenAI plugin samples](https://github.com/temporalio/samples-python/tree/main/google_genai). 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](https://ai.google.dev/gemini-api/docs) for more details.
- If you are new to Temporal, we recommend reading [Understanding Temporal](/evaluate/understanding-temporal) or taking
  the [Temporal 101](https://learn.temporal.io/courses/temporal_101/) course.
- Ensure you have set up your local development environment by following the
  [Set up your local development environment](/develop/python/set-up-your-local-python) 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):

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

If you use pip:

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

The [MCP](#use-mcp-tool-servers) 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](https://github.com/temporalio/samples-python/blob/main/google_genai/hello_world/workflow.py)
```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](https://github.com/temporalio/samples-python/blob/main/google_genai/hello_world/run_worker.py)
```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.

```bash
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](https://github.com/temporalio/samples-python/blob/main/google_genai/hello_world/run_workflow.py)
```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](/develop/python/workflows/basics#workflow-logic-requirements).

This Workflow passes one of each on a single call.

[google_genai/tools/workflow.py](https://github.com/temporalio/samples-python/blob/main/google_genai/tools/workflow.py)
```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.

Register the Activity on the Worker alongside the Workflow.

[google_genai/tools/run_worker.py](https://github.com/temporalio/samples-python/blob/main/google_genai/tools/run_worker.py)
```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](https://github.com/temporalio/samples-python/blob/main/google_genai/chat/workflow.py)
```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](/develop/python/workflows/message-passing#signals) 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](https://github.com/temporalio/samples-python/blob/main/google_genai/structured_output/workflow.py)
```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](https://modelcontextprotocol.io/) 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.

Register each server with a factory that yields a connected, initialized `mcp.ClientSession`.

[google_genai/mcp/run_worker.py](https://github.com/temporalio/samples-python/blob/main/google_genai/mcp/run_worker.py)
```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](https://github.com/temporalio/samples-python/blob/main/google_genai/mcp/workflow.py)
```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](https://github.com/temporalio/samples-python/blob/main/google_genai/streaming/workflow.py)
```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](https://github.com/temporalio/samples-python/blob/main/google_genai/streaming/run_workflow.py)
```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](https://github.com/temporalio/samples-python/blob/main/google_genai/files/workflow.py)
```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](https://github.com/temporalio/samples-python/blob/main/google_genai/interactions/workflow.py)
```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](https://github.com/temporalio/samples-python/blob/main/google_genai/vertex_ai/workflow.py)
```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](https://github.com/temporalio/samples-python/blob/main/google_genai/vertex_ai/run_worker.py)
```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`:

```python
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](https://github.com/temporalio/samples-python/tree/main/google_genai) cover each
pattern as a self-contained, runnable scenario:

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