Google ADK

Agent Development Kit (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

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, 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 to create a Superserve API key.

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 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.

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 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 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.

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 in the ADK docs.

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

> 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:

> 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 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.

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):

> 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 in the ADK docs, and Streaming 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.

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 and Connect. 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.

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 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.
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 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.

Resources

ADK docs

Agents, tools, sessions, and events in Google’s Agent Development Kit.

Run commands

Options for commands.run, including environment variables and working directory.

Use cases

Integrations

Setup

Wire the sandbox as ADK tools

File tools

Streaming events and tool calls

One sandbox per session

Good to know

Resources

ADK docs

Run commands

​Setup

​Wire the sandbox as ADK tools

​File tools

​Streaming events and tool calls

​One sandbox per session

​Good to know

​Resources

ADK docs

Run commands

Setup

Wire the sandbox as ADK tools

File tools

Streaming events and tool calls

One sandbox per session

Good to know

Resources