Repository Public
1This tutorial walks you through building a durable AI agent that uses the Gemini2API for reasoning and [Temporal](https://temporal.io/) for durability. It uses3Temporal's built-in [Gemini SDK integration](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/google_genai).45The agent can call tools, like looking up weather alerts or geolocating an IP6address, and will loop until it has enough information to respond.78What makes this different from a typical agent demo is **durability**. Every LLM9call and every tool invocation is persisted by Temporal. If the process crashes,10the network drops, or an API times out, Temporal automatically retries and11resumes from the last completed step. No conversation history is lost, and no12tool calls are incorrectly repeated.1314## Architecture1516The architecture consists of three parts:1718- **Workflow:** A single `generate_content` call. The Gemini SDK's automatic function calling (AFC) loop runs *inside* the Workflow, and Temporal makes every step of it durable.19- **Activities:** Individual units of work that Temporal makes durable. The Gemini API calls become Activities automatically.20- **Worker:** The process that executes the Workflows and Activities, and the only place your API key lives.2122In this example, you will place all three of these pieces in a single file23(`durable_agent_worker.py`). In a real-world implementation, you would separate24them to allow for various deployment and scalability advantages. You will supply25prompts to the agent with the Temporal CLI, so there is no client code to write.2627## Prerequisites2829To complete this guide, you'll need:3031- A Gemini API key. You can create one for free in [Google AI Studio](https://aistudio.google.com/apikey).32- [Python](https://www.python.org/downloads/) version 3.10 or later.33- [uv](https://docs.astral.sh/uv/getting-started/installation/) for dependency management.34- The [Temporal CLI](https://docs.temporal.io/cli) for running a local development server and starting Workflows.3536## Setup3738Before you begin, ensure you have a39[Temporal development server](https://docs.temporal.io/cli#start-dev-server)40running locally:4142 temporal server start-dev4344Next, create a project and install the required dependencies:4546 uv init durable-gemini-agent47 cd durable-gemini-agent48 uv add "temporalio[google-genai]" httpx python-dotenv4950uv creates and manages the virtual environment for you, so every Python command51later in this tutorial runs through `uv run`.5253Create a `.env` file in your project directory with your Gemini API key. You54can get an API key from55[Google AI Studio](https://aistudio.google.com/apikey).5657 echo "GOOGLE_API_KEY=your-api-key-here" > .env5859> [!NOTE]60> **Note:** Only the worker process needs the API key. Whatever starts the Workflow does not require it, and neither does the Workflow itself---the real `genai.Client` is constructed on the worker and never crosses into Workflow code, so no auth material can appear in Temporal's event history.6162## Implementation6364The rest of this tutorial walks through `durable_agent_worker.py` from top to65bottom, building up the agent piece by piece. Create the file and follow along.6667> [!NOTE]68> **Note:** The `temporalio.contrib.google_genai` integration is experimental. Its API may change in future versions of the Temporal Python SDK.6970### Imports and sandbox setup7172Start with the imports that must be defined up-front. The73`workflow.unsafe.imports_passed_through()` block tells Temporal's Workflow74sandbox to let `httpx` pass through without restriction. Importing `httpx`75executes `class _CookieCompatRequest(urllib.request.Request)`, and the sandbox76blocks subclassing that stdlib class.7778Your tools use `httpx`, and `activity_as_tool()` needs the Workflow to import79those tool functions so Gemini can derive their schemas from the signatures. So80`httpx` reaches the sandbox no matter how you split the files---moving the tools81into their own module doesn't avoid it.8283 from temporalio import workflow8485 with workflow.unsafe.imports_passed_through():86 import httpx8788You don't need to list `google.genai` here. The Temporal plugin you configure89later adds it---along with `pydantic_core` and `annotated_types`---to the sandbox90passthrough set for you.9192### System instructions9394Next, define the agent's personality. The system instructions tell the model how95to behave. This agent is instructed to respond in haikus when no tools are96needed.9798 SYSTEM_INSTRUCTIONS = """99 You are a helpful agent that can use tools to help the user.100 You will be given an input from the user and a list of tools to use.101 You may or may not need to use the tools to satisfy the user ask.102 If no tools are needed, respond in haikus.103 """104105### Tool definitions106107Now define the tools the agent can use. Each tool is an ordinary Temporal108Activity: an async function decorated with `@activity.defn`, with type-annotated109parameters and a descriptive docstring. Gemini builds the function declaration110from that signature and docstring, so document each parameter in the `Args`111section.112113 import json114115 from temporalio import activity116117 NWS_API_BASE = "https://api.weather.gov"118 USER_AGENT = "weather-app/1.0"119120 @activity.defn121 async def get_weather_alerts(state: str) -> str:122 """Get weather alerts for a US state.123124 Args:125 state: Two-letter US state code (e.g. CA, NY)126 """127 headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}128 url = f"{NWS_API_BASE}/alerts/active/area/{state}"129130 async with httpx.AsyncClient() as client:131 response = await client.get(url, headers=headers, timeout=5.0)132 response.raise_for_status()133 return json.dumps(response.json())134135Next, define tools for IP address geolocation:136137 @activity.defn138 async def get_ip_address() -> str:139 """Get the public IP address of the current machine."""140 async with httpx.AsyncClient() as client:141 response = await client.get("https://icanhazip.com")142 response.raise_for_status()143 return response.text.strip()144145 @activity.defn146 async def get_location_info(ipaddress: str) -> str:147 """Get the location information for an IP address including city, state, and country.148149 Args:150 ipaddress: An IP address to look up151 """152 async with httpx.AsyncClient() as client:153 response = await client.get(f"http://ip-api.com/json/{ipaddress}")154 response.raise_for_status()155 result = response.json()156 return f"{result['city']}, {result['regionName']}, {result['country']}"157158That's the whole tool layer. There is no tool registry, no `FunctionDeclaration`159construction, and no dispatch table---the next section wraps these Activities with160`activity_as_tool()`, which passes each parameter through to the Activity161positionally. Tools with zero, one, or several parameters all work.162163### The agent Workflow164165Now you have all the pieces to finish building the agent. The `AgentWorkflow`166class makes one `generate_content` call. `TemporalAsyncClient` is a drop-in167`AsyncClient` whose every API call runs as a Temporal Activity, and168`activity_as_tool()` turns each of your Activities into a Gemini tool.169170When the model asks for a tool, the SDK's AFC loop---running inside the171Workflow---dispatches it through `workflow.execute_activity`, appends the result172to the conversation, and calls the model again. That loop is the agent, and it173is durable because each step is an Activity recorded in Temporal's event174history.175176 from datetime import timedelta177178 from google.genai import types179 from temporalio.contrib.google_genai import TemporalAsyncClient, activity_as_tool180 from temporalio.workflow import ActivityConfig181182 TOOL_CONFIG = ActivityConfig(start_to_close_timeout=timedelta(seconds=30))183184 @workflow.defn185 class AgentWorkflow:186 """Agent workflow that uses Gemini for LLM calls and executes tools."""187188 @workflow.run189 async def run(self, prompt: str) -> str:190 client = TemporalAsyncClient()191192 response = await client.models.generate_content(193 model="gemini-3.8-flash",194 contents=prompt,195 config=types.GenerateContentConfig(196 system_instruction=SYSTEM_INSTRUCTIONS,197 tools=[198 activity_as_tool(get_weather_alerts, activity_config=TOOL_CONFIG),199 activity_as_tool(get_ip_address, activity_config=TOOL_CONFIG),200 activity_as_tool(get_location_info, activity_config=TOOL_CONFIG),201 ],202 ),203 )204205 # Leave this in place. You will un-comment it during a durability206 # test later on.207 # await workflow.sleep(timedelta(seconds=10))208209 return response.text or ""210211A few things to note:212213- Construct `TemporalAsyncClient` **inside** the Workflow. It carries no credentials; it only knows how to turn API calls into Activity invocations.214- `activity_config` must set `start_to_close_timeout` or `schedule_to_close_timeout`. Temporal requires a timeout and there is no default for tool Activities.215- The Gemini API Activities default to a 60-second `start_to_close_timeout`. Override it with `TemporalAsyncClient(activity_config=...)` if your model calls need longer.216217The agent is fully durable. If the worker crashes after several turns, Temporal218picks up exactly where it left off without re-invoking already executed LLM219calls or tool calls.220221### Retries222223Temporal owns retries, so don't enable the Gemini SDK's own retry loop. Set224retry behavior with a `retry_policy` on the Activity config instead:225226 from temporalio.common import RetryPolicy227228 TOOL_CONFIG = ActivityConfig(229 start_to_close_timeout=timedelta(seconds=30),230 retry_policy=RetryPolicy(maximum_attempts=3),231 )232233API failures are also classified for you. Transient statuses (408, 429, 5xx)234stay retryable so the Activity's retry policy applies; other statuses (such as a235400 for a malformed request) are non-retryable, so the Workflow fails fast236instead of burning attempts on an error that won't resolve.237238You can extend that classification. The integration surfaces each API failure as239an `ApplicationError` whose type is the Gemini exception class name---`ClientError`240for 4xx, `ServerError` for 5xx---so listing a name in241`non_retryable_error_types` moves it out of the transient set. For example, to242stop retrying Gemini-side outages and fail the Workflow on the first 5xx, apply243the policy to the Gemini API Activities through `TemporalAsyncClient`:244245 from temporalio.common import RetryPolicy246247 client = TemporalAsyncClient(248 activity_config=ActivityConfig(249 start_to_close_timeout=timedelta(seconds=60),250 retry_policy=RetryPolicy(251 maximum_attempts=5,252 non_retryable_error_types=["ServerError"],253 ),254 ),255 )256257### Worker startup258259Finally, wire everything together. The Temporal worker connects to the Temporal260service and acts as a scheduler for the Workflow and Activity tasks.261262This is where the real `genai.Client` is created, with your API key.263`GoogleGenAIPlugin` takes that client and registers the Gemini API Activities,264installs the Pydantic data converter, and configures the Workflow sandbox.265266 import asyncio267 import os268269 from dotenv import load_dotenv270 from google import genai271 from temporalio.client import Client272 from temporalio.contrib.google_genai import GoogleGenAIPlugin273 from temporalio.envconfig import ClientConfig274 from temporalio.worker import Worker275276 async def main():277 gemini = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])278 plugin = GoogleGenAIPlugin(gemini)279280 config = ClientConfig.load_client_connect_config()281 config.setdefault("target_host", "localhost:7233")282 client = await Client.connect(**config, plugins=[plugin])283284 worker = Worker(285 client,286 task_queue="gemini-agent",287 workflows=[288 AgentWorkflow,289 ],290 activities=[291 get_weather_alerts,292 get_ip_address,293 get_location_info,294 ],295 )296 await worker.run()297298 if __name__ == "__main__":299 load_dotenv()300 asyncio.run(main())301302The plugin removes three pieces of boilerplate you would otherwise need here:303304- No `data_converter=pydantic_data_converter`---the plugin installs the Pydantic payload converter itself.305- No `activity_executor=ThreadPoolExecutor`---every Activity is async.306- No Gemini Activities in the `activities` list---the plugin registers them. You only register your own tools.307308## Run the agent309310That's the entire agent. You don't need to write a client---the Temporal CLI can311start the Workflow for you.312313If you haven't already, start the Temporal development server:314315 temporal server start-dev316317In a new terminal window, start the agent worker:318319 uv run durable_agent_worker.py320321In a third terminal window, submit a query to your agent:322323 temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \324 --input '"are there any weather alerts for where I am?"'325326Note the task queue: it's the same one the worker polls. Starting the Workflow327dispatches a Workflow task carrying the user prompt to that queue, which is what328initiates the agent. `execute` blocks until the Workflow completes and prints the329result. If you'd rather not wait, use `temporal workflow start` with an explicit330`--workflow-id`, then collect the result later with331`temporal workflow result -w your-workflow-id`. Temporal generates the Workflow332ID for you when you omit `--workflow-id`.333334`--input` takes JSON, so a bare string prompt needs its own quotes inside the335shell quotes. The CLI needs no Gemini API key, and no data converter336configuration either: the Workflow's argument and return value are both plain337strings, which the default JSON payload converter handles.338339Open the Temporal UI at340`http://localhost:8233/namespaces/default/workflows` to watch the agentic loop341unfold. You'll see `gemini_api_client_async_request` Activities---one per model342turn---interleaved with one Activity per tool call, each labeled with a343`tool_call` summary. That interleaving *is* the AFC loop, made durable and344observable.345346Try a few different prompts to see the agent reason and call tools. Each command347is the same as the one above with a new `--input`:348349 temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \350 --input '"are there any weather alerts for New York?"'351 temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \352 --input '"where am I?"'353 temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \354 --input '"what is my ip address?"'355 temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \356 --input '"tell me a joke"'357358The last prompt doesn't require any tools, so the agent responds in a haiku359based on the `SYSTEM_INSTRUCTIONS`.360361## Test durability362363Building on Temporal ensures your agent survives failures seamlessly. You can364test this using two distinct experiments.365366### Simulating a network outage367368In this test, you'll temporarily disable your computer's internet connection,369submit a Workflow, watch Temporal automatically retry, and then restore the370network to see it recover.3713721. Disconnect your machine from the internet (for example, turn off your Wi-Fi).3732. Submit a Workflow:374375 temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \376 --input '"tell me a joke"'3773783. Check the Temporal UI (`http://localhost:8233`). You will see the Gemini API379 Activity failing and Temporal automatically managing the retries in the380 background.3813824. Reconnect to the internet.3833845. The next automated retry will successfully reach the Gemini API, and your385 terminal will print the final result.386387### Surviving a worker crash388389In this test, you kill the worker mid-execution and restart it. Temporal replays390the Workflow history (event sourcing) and resumes from the last completed391Activity---already-completed LLM invocations and tool calls are not repeated.3923931. To give yourself time to kill the worker, open `durable_agent_worker.py` and394 uncomment the durable timer in `AgentWorkflow.run`:395396 await workflow.sleep(timedelta(seconds=10))397398 `workflow.sleep` is a Temporal timer, not a local one. It is recorded in399 history and survives the restart, which is what makes this test reliable.4002. Restart the worker:401402 uv run durable_agent_worker.py4034043. Submit a query that triggers several tools:405406 temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \407 --input '"are there any weather alerts where I am?"'4084094. Once the tool calls have completed and the timer is running, kill the worker410 process (`Ctrl-C` in the worker terminal, or `kill %1` if running in the411 background).4124135. Restart the worker:414415 uv run durable_agent_worker.py416417Temporal replays the Workflow history. The LLM calls and tool invocations that418already completed are **not** re-executed---their results are instantly replayed419from history (the event log), the timer resumes, and the Workflow finishes420successfully.421422## Going further423424The integration supports more than this tutorial covers. See the425[plugin documentation](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/google_genai)426for details:427428- **Streaming.** Use `generate_content_stream` as usual. To let an external consumer (a chat UI) observe chunks in real time while the Workflow runs durably, set `TemporalAsyncClient(streaming_topic=...)` and host a `WorkflowStream` in the Workflow.429- **MCP.** Register a client-side MCP server on the worker with `GoogleGenAIPlugin(mcp_servers={...})` and reference it by name in the Workflow with `TemporalMcpClientSession`. Tool discovery and calls run as Activities against a pooled worker-side connection.430- **Vertex AI.** Pass `vertexai=True` to both the worker-side `genai.Client` and the Workflow-side `TemporalAsyncClient`, setting `project` and `location` explicitly on the Workflow side so replay stays deterministic.431432## Further resources433434- [Temporal documentation](https://docs.temporal.io/)435- [Temporal Python SDK](https://docs.temporal.io/develop/python)436- [Temporal Gemini SDK integration](https://docs.temporal.io/develop/python/integrations/google-genai)437- [Google GenAI SDK](https://googleapis.github.io/python-genai/)