# Google GenAI integration > For the complete documentation index, see [llms.txt](https://docs.temporal.io/llms.txt). > Any documentation page is available as raw Markdown by appending `.md` to its URL. 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. > **Public Preview** 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 @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 @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_thing_to_do, ], ), ) return response.text or "" async def recommend_thing_to_do(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 @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 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 if not isinstance(recipe, Recipe): # ``parsed`` is None when the model returns malformed JSON. Fail the # workflow with an ApplicationError rather than asserting: an # assertion is a workflow task failure, which Temporal retries # forever, so the run would hang instead of failing visibly. raise ApplicationError( f"Gemini did not return a valid Recipe: {response.text!r}", non_retryable=True, ) 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 @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 @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 "") # Bound the wait: if the subscriber dies without signaling, complete # anyway instead of waiting forever. try: await workflow.wait_condition(lambda: self._done, timeout=FINISH_TIMEOUT) except asyncio.TimeoutError: workflow.logger.warning( "No finish signal after %s; completing without a subscriber.", FINISH_TIMEOUT, ) 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 async def consume(client: Client, workflow_id: str) -> None: """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() return async def main() -> None: # The stream publishes Pydantic GenerateContentResponse chunks, so the # consumer needs the Pydantic data converter to decode them. client = await Client.connect( os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"), data_converter=pydantic_data_converter, ) ``` 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 @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 @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 @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.