Skip to main content

Build a Google ADK agent in an AI sandbox

This guide walks you through running a Google Agent Development Kit (ADK) agent on Windmill as a single script. The agent runs inside an AI sandbox, keeps its session state in a volume, and exposes a clean main() entry point you can call from a flow, a trigger, or the UI.

ADK is a code-first framework for building, evaluating, and orchestrating agents. It is optimized for Gemini but model-agnostic, so the same script works with other providers via LiteLLM. The same patterns shown here work with both the Python (google-adk) and the TypeScript (@google/adk) ports.

AI sandbox
Run AI agents with process isolation and persistent volumes.
ADK documentation
Google's Agent Development Kit reference and samples.

What you'll build​

A weather assistant agent that:

  • Defines a single ADK agent with one function tool.
  • Runs inside an nsjail sandbox so the agent process is isolated from the worker.
  • Persists session history to a SQLite database stored in a volume, so subsequent runs resume where the conversation left off.
  • Takes a prompt and a googleai resource as script inputs and returns the agent's final response.

Prerequisites​

  • A Windmill instance with workspace object storage configured (required for volumes).
  • Workers with nsjail available (included in standard Windmill Docker images).
  • A Google AI Studio API key — get one at aistudio.google.com/apikey.
  • A googleai resource holding the API key. The googleai resource type ships on the Windmill Hub; create a resource of that type and paste your key into api_key.

Step 1: Create the script​

Create a new script and paste the code for your language. The two annotations at the top are the only Windmill-specific parts — everything else is plain ADK.

# sandbox
# volume: adk-state .adk

#requirements:
#google-adk

import asyncio
import os
from pathlib import Path
from typing import TypedDict

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions.sqlite_session_service import SqliteSessionService
from google.genai import types


class googleai(TypedDict):
    api_key: str
    base_url: str
    platform: str


def get_weather(city: str) -> dict:
    """Return the current weather for a given city.

    Args:
        city: The city name to look up.
    """
    data = {
        "paris": {"temp_c": 18, "conditions": "Sunny"},
        "tokyo": {"temp_c": 22, "conditions": "Cloudy"},
        "new york": {"temp_c": 12, "conditions": "Rainy"},
    }
    info = data.get(city.lower())
    if info is None:
        return {"status": "error", "message": f"No weather data for {city}."}
    return {"status": "success", "city": city, **info}


root_agent = Agent(
    name="weather_agent",
    model="gemini-2.5-flash",
    description="Assistant that answers questions about the current weather.",
    instruction=(
        "You are a friendly weather assistant. "
        "Call the get_weather tool when the user asks about weather. "
        "If a city is not supported, suggest one of: Paris, Tokyo, New York."
    ),
    tools=[get_weather],
)


APP_NAME = "windmill_adk_demo"
USER_ID = "windmill_user"
STATE_DIR = Path(".adk")
SESSION_FILE = STATE_DIR / "session-id.txt"
DB_FILE = STATE_DIR / "sessions.db"


async def _run(prompt: str) -> dict:
    STATE_DIR.mkdir(parents=True, exist_ok=True)

    # SQLite-native service (REAL-typed timestamps, plain aiosqlite). ADK's
    # generic DatabaseSessionService is SQLAlchemy-based and aimed at
    # postgres/mysql; on SQLite it can fail on resume with
    # `fromisoformat: argument must be str`.
    session_service = SqliteSessionService(str(DB_FILE))

    saved_id = SESSION_FILE.read_text().strip() if SESSION_FILE.exists() else None
    session = None
    if saved_id:
        session = await session_service.get_session(
            app_name=APP_NAME, user_id=USER_ID, session_id=saved_id
        )

    is_resume = session is not None
    if not is_resume:
        session = await session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID
        )
        SESSION_FILE.write_text(session.id)

    runner = Runner(
        agent=root_agent,
        app_name=APP_NAME,
        session_service=session_service,
    )

    new_message = types.Content(
        role="user", parts=[types.Part.from_text(text=prompt)]
    )

    response = ""
    async for event in runner.run_async(
        user_id=USER_ID, session_id=session.id, new_message=new_message
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    response += part.text

    return {
        "is_resume": is_resume,
        "session_id": session.id,
        "prompt": prompt,
        "response": response,
    }


def main(gemini: googleai, prompt: str = "What's the weather in Paris?") -> dict:
    # Script inputs are not env vars — set them explicitly.
    # See "Passing credentials" below.
    os.environ["GOOGLE_API_KEY"] = gemini["api_key"]
    if gemini.get("platform") == "google_vertex_ai":
        os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "true"
    if gemini.get("base_url"):
        os.environ["GOOGLE_GEMINI_BASE_URL"] = gemini["base_url"]
    return asyncio.run(_run(prompt))

Step 2: Run it​

Click Test with a prompt like "What's the weather in Paris?" and your API key. You should see something like:

{
  "is_resume": false,
  "session_id": "51b81888-567e-4f46-9190-dff1e1c4eefd",
  "prompt": "What's the weather in Paris?",
  "response": "The weather in Paris is Sunny with a temperature of 18 degrees Celsius."
}

Now run it again with "And in Tokyo?". The script reads the saved session ID from the volume, resumes the same conversation, and the agent answers based on the prior turn:

{
  "is_resume": true,
  "session_id": "51b81888-567e-4f46-9190-dff1e1c4eefd",
  "prompt": "And in Tokyo?",
  "response": "Tokyo is cloudy at 22°C."
}

How it works​

# sandbox / // sandbox​

Wraps the job process in nsjail. The agent — and any subprocess it spawns (e.g. an MCP server or shell tool) — sees an isolated filesystem and cannot reach the worker's secrets or other jobs.

# volume: adk-state .adk / // volume: adk-state .adk​

Mounts a persistent volume at ./.adk relative to the job's working directory. Files written there survive across runs and are synced to object storage. The script uses two files in the volume:

  • .adk/session-id.txt — the active ADK session ID, so the next run resumes the same conversation.
  • .adk/sessions.db — a SQLite database managed by ADK's DatabaseSessionService, holding the full session history.

If you want a fresh conversation, delete the volume contents (or use a different volume name).

DatabaseSessionService​

ADK ships several session services. InMemorySessionService loses state at the end of each job — useless across Windmill runs. DatabaseSessionService with a SQLite URL inside the volume gives you persistence with zero infrastructure.

The two ports do not use the same SQLite implementation:

  • Python — use SqliteSessionService (in google.adk.sessions.sqlite_session_service). It's a SQLite-native service that stores timestamps as REAL and goes through aiosqlite directly, sidestepping a SQLAlchemy + aiosqlite type-marshalling bug that surfaces on resume as fromisoformat: argument must be str. The generic DatabaseSessionService is fine for postgres/mysql but should not be used with SQLite.
  • TypeScript — DatabaseSessionService works with a sqlite://<path> URL via MikroORM. No extra dependency needed.

For multi-user or multi-tenant setups, scope the volume per workspace or input:

// volume: $workspace-adk-state .adk

See dynamic volume names for the full set of placeholders.

Passing credentials​

Script inputs are language variables — they are not automatically exported as environment variables inside the sandbox. ADK reads its API key from the environment, so the scripts set it explicitly:

  • Python (google-adk) reads GOOGLE_API_KEY.
  • TypeScript (@google/adk) reads GEMINI_API_KEY or GOOGLE_GENAI_API_KEY.

Apply the same pattern for any other credential the agent or its tools need at runtime — for example OPENAI_API_KEY, TAVILY_API_KEY, or an MCP server token.

The script also forwards the resource's optional platform and base_url fields:

  • platform: "google_vertex_ai" flips ADK to Vertex mode by setting GOOGLE_GENAI_USE_VERTEXAI=true. Vertex needs GOOGLE_CLOUD_PROJECT and GOOGLE_CLOUD_LOCATION set on the worker (or wired through additional script inputs).
  • base_url is exported as GOOGLE_GEMINI_BASE_URL for users running behind a proxy or against a non-default endpoint. Most users leave both fields empty.

Extending the agent​

Use a different model​

ADK is model-agnostic. Both ports support LiteLLM (Python) or compatible model wrappers (TypeScript) to swap the underlying provider. To run the same agent on Anthropic Claude in Python, replace the model= line:

from google.adk.models.lite_llm import LiteLlm

root_agent = Agent(
    name="weather_agent",
    model=LiteLlm("anthropic/claude-3-5-sonnet-20241022"),
    ...
)

Then export ANTHROPIC_API_KEY (or the relevant provider key) the same way as GOOGLE_API_KEY. If you already have a Windmill anthropic resource, accept it as a script input:

def main(anthropic: dict, prompt: str = "..."):
    os.environ["ANTHROPIC_API_KEY"] = anthropic["apiKey"]
    return asyncio.run(_run(prompt))

Compose multiple agents​

ADK supports multi-agent systems out of the box — define sub-agents and assign them to a coordinator:

greeter = Agent(name="greeter", model="gemini-2.5-flash", instruction="...", description="...")
researcher = Agent(name="researcher", model="gemini-2.5-flash", instruction="...", description="...", tools=[...])

root_agent = Agent(
    name="coordinator",
    model="gemini-2.5-flash",
    description="Routes user requests to the right specialist.",
    sub_agents=[greeter, researcher],
)

The runner code stays unchanged — ADK handles the routing and tool calls.

Add MCP tools​

ADK can consume any Model Context Protocol server via MCPToolset. Because the sandbox isolates subprocesses, you can safely launch MCP servers inside the same job:

from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters

tools = await MCPToolset.from_server(
    connection_params=StdioServerParameters(command="npx", args=["@modelcontextprotocol/server-filesystem", "."])
)

Pair this with a separate volume for the MCP server's working directory if it needs persistent state.

Use the agent inside a flow​

Drop this script into any flow step and chain it with other Windmill primitives — fetch context from a database, call the agent, then post the response to Slack. The volume keeps the agent's memory consistent across flow runs.

Troubleshooting​

  • #sandbox annotation but nsjail is not available — your worker image does not include nsjail. Use a standard Windmill Docker image or remove the annotation.
  • Volume not persisting — check that workspace object storage is configured under instance settings. Without it, volumes only exist for the duration of a single job.
  • fromisoformat: argument must be str (Python, on resume) — you used DatabaseSessionService against SQLite, which hits a SQLAlchemy + aiosqlite type-marshalling bug. Switch to SqliteSessionService from google.adk.sessions.sqlite_session_service (the snippet above already does this) and delete the old .adk/sessions.db.
  • API key must be provided via constructor or GOOGLE_GENAI_API_KEY or GEMINI_API_KEY (TypeScript) — the JS port reads GEMINI_API_KEY / GOOGLE_GENAI_API_KEY, not GOOGLE_API_KEY. Set both for portability.
AI sandbox
Run AI agents with process isolation and persistent volumes.
Volumes
Persistent file storage for scripts, synced to object storage.