> ## Documentation Index
> Fetch the complete documentation index at: https://docs.superserve.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Google ADK

> Use Superserve sandboxes as the execution runtime for agents built with Google's Agent Development Kit.

[Agent Development Kit](https://adk.dev/) (ADK) is Google's framework for building agents with tools, sessions, and streaming events. Back the agent's tools with `sandbox.commands.run` and `sandbox.files`, and keep each conversation's sandbox ID in the ADK session state. Every command the model runs, and every file it writes, then stays in an isolated VM instead of on your server. The model API key stays in your app; the sandbox never sees it.

## Setup

```bash theme={"theme":{"light":"github-light","dark":"github-dark-default"}}
pip install superserve google-adk "litellm[mcp]>=1.84,<1.106.3"
export SUPERSERVE_API_KEY=ss_live_...
export ANTHROPIC_API_KEY=sk-ant-...
```

ADK needs Python 3.11 or later. The examples on this page use Claude through ADK's [LiteLLM connector](https://adk.dev/agents/models/litellm/), which reads `ANTHROPIC_API_KEY` from the environment. The LiteLLM version range is the one that ADK declares for its `extensions` extra. Install the `mcp` extra: without it, LiteLLM 1.106 fails every request that includes tools with `litellm.APIConnectionError: No module named 'mcp'`.

The examples were tested with `google-adk` 2.13.0, `litellm` 1.106.2, and `superserve` 0.11.5. See [API keys](https://docs.superserve.ai/api-key.md) to create a Superserve API key.

> [!NOTE]
>
> ADK uses Gemini models by default. To use Gemini, pass a model name such as `model="gemini-3.5-flash"` instead of the `LiteLlm` object, and set `GOOGLE_API_KEY` instead of `ANTHROPIC_API_KEY`. You do not need LiteLLM for Gemini. The examples on this page were not tested with Gemini. See [Models](https://adk.dev/agents/models/) in the ADK docs.

## Wire the sandbox as ADK tools

An ADK tool is a Python function. The highlighted lines define a `bash` tool backed by the sandbox, and store the sandbox ID in the session state. Everything else is standard ADK usage.

```python {11-12,15-19,26,33,38,54} theme={"theme":{"light":"github-light","dark":"github-dark-default"}}
import asyncio
from typing import Any

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import InMemoryRunner
from google.adk.tools.tool_context import ToolContext
from google.genai import types
from superserve import AsyncSandbox

# Sandbox handles by ID. The session state holds only the ID.
sandboxes: dict[str, AsyncSandbox] = {}


async def bash(command: str, tool_context: ToolContext) -> dict[str, Any]:
    """Run a shell command in the sandbox."""
    sandbox = sandboxes[tool_context.state["sandbox_id"]]
    result = await sandbox.commands.run(command, timeout_seconds=60)
    return result.model_dump()


agent = Agent(
    name="sandbox_agent",
    model=LiteLlm(model="anthropic/claude-sonnet-5-5"),
    instruction="Use the bash tool to inspect the sandbox before answering.",
    tools=[bash],
)
runner = InMemoryRunner(agent=agent, app_name="sandbox_app")


async def main() -> None:
    sandbox = await AsyncSandbox.create(name="agent-runtime")
    sandboxes[sandbox.id] = sandbox
    try:
        session = await runner.session_service.create_session(
            app_name="sandbox_app",
            user_id="user-1",
            state={"sandbox_id": sandbox.id},
        )
        message = types.Content(
            role="user",
            parts=[
                types.Part(
                    text="List the files in /tmp and tell me how many there are."
                )
            ],
        )
        async for event in runner.run_async(
            user_id="user-1", session_id=session.id, new_message=message
        ):
            if event.is_final_response() and event.content and event.content.parts:
                print(event.content.parts[0].text)
    finally:
        del sandboxes[sandbox.id]
        await sandbox.kill()


asyncio.run(main())
```

ADK builds the tool declaration from the function signature and docstring. It passes `tool_context` to every tool that declares it, and does not show that parameter to the model. See [Context injection](https://adk.dev/tools-custom/function-tools/#context-injection) in the ADK docs.

The session state holds the sandbox ID, not the sandbox object, so one agent defined at module level can work with a different sandbox in every session. The `sandboxes` dictionary maps each ID to its sandbox handle in this process. Do not put the handle in the session state: session services copy the state, and `DatabaseSessionService` stores it as JSON. See [State](https://adk.dev/sessions/state/) in the ADK docs.

ADK prefers tools that return a dictionary. `result.model_dump()` gives the model `stdout`, `stderr`, `exit_code`, and `truncated`, so the model sees failed commands as well as their output.

## File tools

Add file tools next to `bash` to let the model write a script, run it, and read the result.

```python {22-37,47,74} theme={"theme":{"light":"github-light","dark":"github-dark-default"}}
import asyncio
from typing import Any

from google.adk.agents import Agent, RunConfig
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import InMemoryRunner
from google.adk.tools.tool_context import ToolContext
from google.genai import types
from superserve import AsyncSandbox, NotFoundError

# Sandbox handles by ID. The session state holds only the ID.
sandboxes: dict[str, AsyncSandbox] = {}


async def bash(command: str, tool_context: ToolContext) -> dict[str, Any]:
    """Run a shell command in the sandbox."""
    sandbox = sandboxes[tool_context.state["sandbox_id"]]
    result = await sandbox.commands.run(command, timeout_seconds=60)
    return result.model_dump()


async def write_file(
    path: str, content: str, tool_context: ToolContext
) -> dict[str, Any]:
    """Write a text file in the sandbox. Use an absolute path."""
    sandbox = sandboxes[tool_context.state["sandbox_id"]]
    await sandbox.files.write(path, content)
    return {"status": "success", "path": path, "characters": len(content)}


async def read_file(path: str, tool_context: ToolContext) -> dict[str, Any]:
    """Read a text file from the sandbox. Use an absolute path."""
    sandbox = sandboxes[tool_context.state["sandbox_id"]]
    try:
        return {"status": "success", "content": await sandbox.files.read_text(path)}
    except NotFoundError:
        return {"status": "error", "error_message": f"{path} does not exist."}


agent = Agent(
    name="code_runner",
    model=LiteLlm(model="anthropic/claude-sonnet-5-5"),
    instruction=(
        "Save code under /home/user and run it with the bash tool. "
        "Use python3, not python."
    ),
    tools=[bash, write_file, read_file],
)
runner = InMemoryRunner(agent=agent, app_name="code_runner_app")


async def main() -> None:
    sandbox = await AsyncSandbox.create(name="code-runner")
    sandboxes[sandbox.id] = sandbox
    try:
        session = await runner.session_service.create_session(
            app_name="code_runner_app",
            user_id="user-1",
            state={"sandbox_id": sandbox.id},
        )
        message = types.Content(
            role="user",
            parts=[
                types.Part(
                    text="Write a Python script that prints the first 10 prime numbers, "
                    "run it, and show me the output."
                )
            ],
        )
        async for event in runner.run_async(
            user_id="user-1",
            session_id=session.id,
            new_message=message,
            run_config=RunConfig(max_llm_calls=10),
        ):
            for call in event.get_function_calls():
                print(f"> {call.name} {call.args}")
            if event.is_final_response() and event.content and event.content.parts:
                print(event.content.parts[0].text)
    finally:
        del sandboxes[sandbox.id]
        await sandbox.kill()


asyncio.run(main())
```

* `read_file` returns a missing file as an error result with `status` and `error_message`, and the model can correct the path and try again. An exception that a tool does not catch stops the run, and `run_async` raises it.
* `RunConfig(max_llm_calls=10)` caps the model calls in the run (the default is 500). Past the limit, the run raises `LlmCallsLimitExceededError`. See [Runtime Config](https://adk.dev/runtime/runconfig/) in the ADK docs.

Output from a run (trimmed; the model's wording varies):

````text theme={"theme":{"light":"github-light","dark":"github-dark-default"}}
> write_file {'path': '/home/user/primes.py', 'content': 'def is_prime(n):\n    if n < 2:\n ...'}
> bash {'command': 'python3 /home/user/primes.py'}
I saved the script as `/home/user/primes.py` and ran it. It finds primes by trial division and stops once it has 10.

Output:
```
First 10 prime numbers:
2
3
5
7
11
13
17
19
23
29
```
````

A run that asked the model to read `/home/user/config.json` first got the error result, then created the file and read it back:

```text theme={"theme":{"light":"github-light","dark":"github-dark-default"}}
> read_file {'path': '/home/user/config.json'}
< read_file {'status': 'error', 'error_message': '/home/user/config.json does not exist.'}
> write_file {'path': '/home/user/config.json', 'content': '{"debug": true}'}
< write_file {'status': 'success', 'path': '/home/user/config.json', 'characters': 15}
> read_file {'path': '/home/user/config.json'}
< read_file {'status': 'success', 'content': '{"debug": true}'}
`/home/user/config.json` didn't exist, so I created it with `{"debug": true}`. I then read it back, and the contents are `{"debug": true}`.
```

`sandbox.files` paths must be absolute. See [Read & write files](https://docs.superserve.ai/filesystem/read-write.md) for binary content and directory downloads.

## Streaming events and tool calls

`runner.run_async()` yields ADK events. With `StreamingMode.SSE`, the model's text arrives in partial events as it is generated. Tool calls and tool results arrive in events that are not partial.

```python {5,53-63} theme={"theme":{"light":"github-light","dark":"github-dark-default"}}
import asyncio
from typing import Any

from google.adk.agents import Agent
from google.adk.agents.run_config import RunConfig, StreamingMode
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import InMemoryRunner
from google.adk.tools.tool_context import ToolContext
from google.genai import types
from superserve import AsyncSandbox

# Sandbox handles by ID. The session state holds only the ID.
sandboxes: dict[str, AsyncSandbox] = {}


async def bash(command: str, tool_context: ToolContext) -> dict[str, Any]:
    """Run a shell command in the sandbox."""
    sandbox = sandboxes[tool_context.state["sandbox_id"]]
    result = await sandbox.commands.run(command, timeout_seconds=60)
    return result.model_dump()


agent = Agent(
    name="sandbox_agent",
    model=LiteLlm(model="anthropic/claude-sonnet-5-5"),
    instruction="Use the bash tool to inspect the sandbox before answering.",
    tools=[bash],
)
runner = InMemoryRunner(agent=agent, app_name="sandbox_app")


async def main() -> None:
    sandbox = await AsyncSandbox.create(name="agent-runtime")
    sandboxes[sandbox.id] = sandbox
    try:
        session = await runner.session_service.create_session(
            app_name="sandbox_app",
            user_id="user-1",
            state={"sandbox_id": sandbox.id},
        )
        message = types.Content(
            role="user",
            parts=[
                types.Part(
                    text="Which Linux distribution and Python version does this machine run?"
                )
            ],
        )
        async for event in runner.run_async(
            user_id="user-1",
            session_id=session.id,
            new_message=message,
            run_config=RunConfig(streaming_mode=StreamingMode.SSE),
        ):
            if event.partial and event.content and event.content.parts:
                for part in event.content.parts:
                    if part.text:
                        print(part.text, end="", flush=True)
                continue
            for call in event.get_function_calls():
                print(f"\n> {call.name} {call.args}")
            for response in event.get_function_responses():
                print(f"< {response.response}")
        print()
    finally:
        del sandboxes[sandbox.id]
        await sandbox.kill()


asyncio.run(main())
```

The event that ends a streamed text response is not partial and repeats the full text, so the loop prints text only from partial events. Output from a run (trimmed; the model's wording varies):

```text theme={"theme":{"light":"github-light","dark":"github-dark-default"}}
> bash {'command': 'cat /etc/os-release; python3 --version; python --version'}

> bash {'command': 'uname -a'}
< {'stdout': 'PRETTY_NAME="Ubuntu 24.04.5 LTS"\n...Python 3.12.3\n', 'stderr': '/bin/sh: 1: python: not found\n', 'exit_code': 127, 'truncated': False}
< {'stdout': 'Linux sandbox-c7cbc8e3 6.1.158 #1 SMP PREEMPT_DYNAMIC Wed Oct 21 22:41:16 UTC 2026 x86_64 x86_64 x86_64 GNU/Linux\n', 'stderr': '', 'exit_code': 0, 'truncated': False}
The machine runs **Ubuntu 24.04.5 LTS (Noble Numbat)** and has **Python 3.12.3**.

Python is only available as `python3`. There is no `python` command. The kernel is Linux 6.1.158 on x86_64.
```

See [Events](https://adk.dev/events/) in the ADK docs, and [Streaming](https://docs.superserve.ai/commands/streaming.md) to stream a single command's output.

## One sandbox per session

To serve many conversations, create one sandbox for each ADK session, and keep its ID in the session state. On each turn, resume the sandbox, run the agent, and pause the sandbox afterwards. Files and installed packages stay available, and you do not pay for an idle VM between turns.

```python {16,21,37-47,60-69,74,84-86,92-97} theme={"theme":{"light":"github-light","dark":"github-dark-default"}}
import asyncio
from typing import Any

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import InMemoryRunner
from google.adk.sessions import Session
from google.adk.tools.tool_context import ToolContext
from google.genai import types
from superserve import AsyncSandbox

APP_NAME = "chat_app"


# One sandbox handle for each chat, by ID. The session state holds only the ID.
sandboxes: dict[str, AsyncSandbox] = {}


async def bash(command: str, tool_context: ToolContext) -> dict[str, Any]:
    """Run a shell command in the sandbox."""
    sandbox = sandboxes[tool_context.state["sandbox_id"]]
    result = await sandbox.commands.run(command, timeout_seconds=60)
    return result.model_dump()


agent = Agent(
    name="chat_agent",
    model=LiteLlm(model="anthropic/claude-sonnet-5-5"),
    instruction="Complete requests by running shell commands in the sandbox.",
    tools=[bash],
)
runner = InMemoryRunner(agent=agent, app_name=APP_NAME)
session_service = runner.session_service


async def start_chat(user_id: str) -> str:
    sandbox = await AsyncSandbox.create(name="chat-session")
    try:
        session = await session_service.create_session(
            app_name=APP_NAME, user_id=user_id, state={"sandbox_id": sandbox.id}
        )
        await sandbox.pause()
    except BaseException:
        # Delete the sandbox if setup fails, so it does not keep running.
        await sandbox.kill()
        raise
    sandboxes[sandbox.id] = sandbox
    return session.id


async def get_chat(user_id: str, session_id: str) -> Session:
    session = await session_service.get_session(
        app_name=APP_NAME, user_id=user_id, session_id=session_id
    )
    if session is None:
        raise LookupError(f"No session {session_id}")
    return session


async def resume_sandbox(sandbox_id: str) -> AsyncSandbox:
    sandbox = sandboxes.get(sandbox_id)
    if sandbox is None:
        # No handle in this process, for example after a restart.
        # connect() resumes the sandbox if it is paused.
        sandbox = await AsyncSandbox.connect(sandbox_id)
        sandboxes[sandbox_id] = sandbox
    else:
        await sandbox.resume()
    return sandbox


async def send(user_id: str, session_id: str, text: str) -> str:
    session = await get_chat(user_id, session_id)
    sandbox = await resume_sandbox(session.state["sandbox_id"])
    try:
        reply = ""
        message = types.Content(role="user", parts=[types.Part(text=text)])
        async for event in runner.run_async(
            user_id=user_id, session_id=session_id, new_message=message
        ):
            if event.is_final_response() and event.content and event.content.parts:
                reply = event.content.parts[0].text or ""
        return reply
    finally:
        # Pause between turns. Files and memory are kept until the next resume.
        await sandbox.pause()


async def end_chat(user_id: str, session_id: str) -> None:
    session = await get_chat(user_id, session_id)
    sandbox_id = session.state["sandbox_id"]
    sandbox = sandboxes.pop(sandbox_id, None)
    if sandbox is None:
        await AsyncSandbox.kill_by_id(sandbox_id)
    else:
        # kill() also closes the HTTP connection of the handle.
        await sandbox.kill()
    await session_service.delete_session(
        app_name=APP_NAME, user_id=user_id, session_id=session_id
    )


async def main() -> None:
    session_id = await start_chat("user-1")
    try:
        await send(
            "user-1", session_id, "Create /home/user/todo.txt with three short items."
        )
        print(
            await send(
                "user-1", session_id, "Add a fourth item to that file, then print it."
            )
        )
    finally:
        await end_chat("user-1", session_id)


asyncio.run(main())
```

The session keeps the conversation history and the sandbox ID. The second turn understands "that file" because of the session history, and finds the file because it lives in the sandbox.

`start_chat` keeps the handle from `create()` in `sandboxes`, and every turn of the chat uses that one handle. A handle does not resume the sandbox by itself after `pause()`, so `resume_sandbox` calls `resume()` before each turn. If the process has no handle for the chat, for example after a restart, `AsyncSandbox.connect()` makes one and resumes the sandbox. `end_chat` calls `kill()` on the handle, which deletes the sandbox and closes the HTTP connection of the handle. See [Lifecycle](https://docs.superserve.ai/sandbox/lifecycle.md) and [Connect](https://docs.superserve.ai/sandbox/connect.md).

If `create_session()` or `pause()` fails, `start_chat` deletes the sandbox, so no sandbox runs without a session. Run one turn at a time for each session. If two turns overlap, the first turn to finish pauses the sandbox while the other turn still uses it.

> [!NOTE]
>
> `InMemoryRunner` keeps sessions in memory, so they are lost when the process stops. To keep sessions across restarts, install `google-adk[db]` and pass a `DatabaseSessionService`, for example `DatabaseSessionService(db_url="sqlite+aiosqlite:///sessions.db")`, as `session_service` to `Runner`. The sandbox ID is stored with the rest of the session state. See [Session](https://adk.dev/sessions/session/#databasesessionservice) in the ADK docs.

## Good to know

* Each `AsyncSandbox` handle opens its own HTTP connection, which stays open until `kill()` or until Python frees the handle. `pause()` does not close it. Keep one handle for each sandbox, share it with the tools, and do not call `connect()` in each tool call or each turn.
* The model can call only the tools you define. Keep `kill()`, `pause()`, and `update()` in your application code, not in a tool.
* To limit what model-written code can reach, create the sandbox with [network rules](https://docs.superserve.ai/sandbox/networking.md).
* Pass `timeout_seconds` to `commands.run`. Without it, the SDK request times out after 30 seconds, so the tool call fails on longer commands such as package installs.
* The default sandbox has `python3` but no `python`. Tell the agent in its `instruction`, or use a [template](https://docs.superserve.ai/templates/overview.md) with the tools you need.
* As a backstop for crashes, set `timeout_seconds` and `auto_delete_seconds` when you create the sandbox. See [Auto-delete](https://docs.superserve.ai/sandbox/lifecycle.md#auto-delete).

## Resources

> [card] [ADK docs](https://adk.dev/)
>
> Agents, tools, sessions, and events in Google's Agent Development Kit.

> [card] [Run commands](https://docs.superserve.ai/commands/overview.md)
>
> Options for `commands.run`, including environment variables and working directory.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.