Metadata-Version: 2.4
Name: temporalio-google-genai
Version: 0.1.0
Summary: Temporal integration for google genai
Author: Temporal Technologies Inc
Author-email: Temporal Technologies Inc <sdk@temporal.io>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Requires-Dist: temporalio>=1.34.0,<2
Requires-Dist: google-genai>=2.21.0,<3
Requires-Dist: pydantic>=2.0.0,<3
Requires-Dist: mcp>=1.24,<2 ; extra == 'mcp'
Requires-Python: >=3.10
Project-URL: Homepage, https://github.com/temporalio/ai-integrations/tree/main/python/google_genai
Project-URL: Repository, https://github.com/temporalio/ai-integrations
Project-URL: Documentation, https://docs.temporal.io/develop/python/integrations/google-genai
Provides-Extra: mcp
Description-Content-Type: text/markdown

# Google Gemini SDK Integration for Temporal

> Release stage: [Public Preview](https://docs.temporal.io/develop/python/integrations/google-genai).

## Overview

This plugin lets you use the [Google Gemini SDK](https://googleapis.github.io/python-genai/)
(`google-genai`) inside Temporal workflows with durable execution. Every Gemini
API call becomes a **Temporal activity**, so model calls, tool calls, file
operations, interactions, and managed agents are retried, recorded in history,
and survive worker restarts.

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
  tool wrappers (`activity_as_tool`) work naturally — no manual agent loop.
- **Temporal owns retries.** Configure them via the activity `retry_policy`; the
  SDK's own retry loop is rejected to avoid double-retry (see
  [Retries & errors](#retries--errors)).

## Install

```bash
uv add temporalio-google-genai
```

## Hello World

```python
import os
from datetime import timedelta

from google import genai
from google.genai import types

from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.google_genai import (
    GoogleGenAIPlugin,
    TemporalAsyncClient,
    activity_as_tool,
)
from temporalio.worker import Worker
from temporalio.workflow import ActivityConfig


# ---- a tool, as a normal Temporal activity (runs on the worker) ----
@activity.defn
async def get_weather(city: str) -> str:
    return f"It's sunny in {city}."


# ---- the workflow (runs in the Temporal sandbox) ----
@workflow.defn
class WeatherAgent:
    @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),
                        ),
                    ),
                ],
            ),
        )
        return response.text or ""


# ---- worker setup (outside the sandbox: real client + credentials) ----
async def main() -> None:
    gemini = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
    plugin = GoogleGenAIPlugin(gemini)

    client = await Client.connect("localhost:7233", plugins=[plugin])
    async with Worker(
        client,
        task_queue="gemini",
        workflows=[WeatherAgent],
        activities=[get_weather],
    ):
        result = await client.execute_workflow(
            WeatherAgent.run,
            "What's the weather in Tokyo?",
            id="weather-1",
            task_queue="gemini",
        )
        print(result)
```

Construct `TemporalAsyncClient` **inside** the workflow; construct the real
`genai.Client` and `GoogleGenAIPlugin` **on the worker**.

## What this plugin gives you

| Surface | Workflow API | Runs as |
| --- | --- | --- |
| Model calls | `client.models.generate_content` / `generate_content_stream` | activity (AFC loop in workflow) |
| Tools | `activity_as_tool(fn, ...)` | one activity per tool call |
| Files | `client.files.upload` / `download` | activity |
| File search | `client.file_search_stores.upload_to_file_search_store` | activity |
| Interactions | `client.interactions.create` / `get` / `cancel` / `delete` | whole-operation activity |
| Managed agents | `client.agents.create` / `get` / `list` / `delete` | whole-operation activity |
| MCP (client-side) | `TemporalMcpClientSession(name)` in `tools=[...]` | `list_tools` / `call_tool` activities |

Streamed responses are batched: the activity drains the stream and the workflow
iterates the collected chunks/events. `client.webhooks` is not supported in
workflows and raises.

## Tool calling

`activity_as_tool` wraps any `@activity.defn` function as a Gemini tool. When the
model calls it, the AFC loop (running in the workflow) dispatches it as a
durable activity:

```python
activity_as_tool(
    get_weather,
    activity_config=ActivityConfig(start_to_close_timeout=timedelta(seconds=30)),
)
```

A timeout is required — `activity_config` must set `start_to_close_timeout` or
`schedule_to_close_timeout` (Temporal needs one; there is no default for tools).

## MCP support

Install the optional dependency for client-side MCP:

```bash
uv add "temporalio-google-genai[mcp]"
```

This adapter requires MCP Python SDK v1. The base package does not require MCP
and can be installed alongside MCP v2 integrations. Server-side MCP on Vertex AI
and the Interactions API do not require this extra.

Client-side MCP (Gemini Developer API) is wired through the plugin: register the
server on the worker and reference it by name in the workflow.

```python
from contextlib import asynccontextmanager
import sys

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from temporalio.google_genai import TemporalMcpClientSession


# ---- worker: a factory yielding a connected, initialized session ----
@asynccontextmanager
async def weather_mcp():
    params = StdioServerParameters(command=sys.executable, args=["weather_server.py"])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session


plugin = GoogleGenAIPlugin(
    genai.Client(api_key=os.environ["GOOGLE_API_KEY"]),
    mcp_servers={"weather": weather_mcp},
    mcp_connection_idle_timeout=timedelta(minutes=5),
)


# ---- workflow: reference the server by name in the tools list ----
@workflow.defn
class McpAgent:
    @workflow.run
    async def run(self, prompt: str) -> str:
        client = TemporalAsyncClient()
        session = TemporalMcpClientSession(
            "weather",
            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 ""
```

The MCP connection lives on the worker (pooled, idle-evicted); the workflow only
carries the server name. Tool discovery and calls run as `{name}-list-tools` /
`{name}-call-tool` activities, so the full tool parameter schema reaches the
model. Set `cache_tools=True` to list a server's tools once per workflow instead
of per turn.

## Streaming

`generate_content_stream` works as usual — the workflow iterates chunks (batched
from the activity). To let an **external** consumer (a chat UI) observe chunks in
real time while the workflow runs durably, set `streaming_topic` on the client
and host a [`WorkflowStream`](https://github.com/temporalio/sdk-python/tree/6adc0d84290a79952dee3ef02c36f6ed9334874a/temporalio/contrib/workflow_streams/) in the workflow. Each
streamed `GenerateContentResponse` is published to that topic as it arrives:

```python
from temporalio.contrib.workflow_streams import WorkflowStream


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

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

Consume the stream from outside the workflow:

```python
from temporalio.contrib.workflow_streams import WorkflowStreamClient


async def consume(client, workflow_id):
    stream = WorkflowStreamClient.create(client, workflow_id)
    async for item in stream.subscribe(
        ["gemini"], result_type=types.GenerateContentResponse,
    ):
        print(item.data.text, end="", flush=True)
```

The workflow's own iteration is unchanged (it still receives batched chunks for
the SDK to parse); the topic is purely for external real-time observation. If
`streaming_topic` is set but the workflow hosts no `WorkflowStream`, the call
raises `GoogleGenAIError`. Tune flush cadence with
`TemporalAsyncClient(streaming_topic=..., streaming_batch_interval=...)`
(default 100ms).

## Retries & errors

Temporal owns retries. Configure them with the activity `retry_policy` via
`activity_config`. The plugin **rejects** the SDK's own retry config so retries
don't compound:

- Constructing the plugin with a `genai.Client` that has
  `http_options.retry_options` raises `ValueError`.
- Setting `http_options.retry_options` on a per-request call raises
  `GoogleGenAIError`.

API-call activities classify failures: transient statuses (408, 429, 5xx) stay
retryable (the activity's `retry_policy` applies); other statuses (e.g. 4xx) are
non-retryable so the workflow fails fast.

## Vertex AI

Pass `vertexai=True` to both the worker-side `genai.Client` and the
workflow-side `TemporalAsyncClient`. On the workflow side you must also set
`project` and `location` **explicitly**:

```python
# worker
genai.Client(vertexai=True, project="my-project", location="us-central1")

# workflow
TemporalAsyncClient(vertexai=True, project="my-project", location="us-central1")
```

Normally the SDK auto-discovers `project`/`location` from the environment
(credentials, ADC, metadata server). That discovery
would be non-deterministic and break replay. Setting them by hand
keeps it deterministic.

## Composing with other plugins

`GoogleGenAIPlugin` is a `temporalio.plugin.SimplePlugin`; pass it in the
`plugins=[...]` list alongside others (e.g. OpenTelemetry). It contributes a
Pydantic data converter, the Gemini activities, a sandbox-passthrough config for
`google.genai` (and `mcp`), and registers `GoogleGenAIError` as a workflow
failure type. When composing data converters, construct the plugins so their
converters are compatible.
