Google ADK & Vertex AI

Add one plugin to your ADK Runner — or to an AdkApp on Vertex AI Agent Engine — and every agent, tool call and handoff is evaluated: grounding against what each agent retrieved, trajectory and tool correctness, loops, agreement between agents, and whether the final answer used them. No per-agent code, and nothing on your request path.

1. Install

pip install "variably-sdk[adk]"      # Python 3.10+, google-adk 2.x

2. Set two environment variables

VARIABLY_API_KEY=vbl_your_key_here
VARIABLY_BASE_URL=https://api.variably.dev

The API key is scoped to one project — everything the plugin sends lands there and nowhere else.

3. Register the plugin

from google.adk.runners import Runner
from variably.integrations.adk import VariablyPlugin

runner = Runner(
    agent=root_agent,
    app_name="claims-assistant",
    session_service=session_service,
    plugins=[VariablyPlugin()],
)

That is the whole integration for a running app. The rest of this page is what it captures and how to tune it.

A complete example

A coordinator hands the question to a researcher, which searches a policy store and asks a calculator sub-agent for a dose. Copy it, set GOOGLE_API_KEY for Gemini, and run it:

import asyncio
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import AgentTool, FunctionTool
from google.genai import types
from variably.integrations.adk import VariablyPlugin

POLICY = {
    "amoxicillin": "Adults: 500 mg every 8 hours. Maximum 2000 mg per day.",
}

def search_policy(query: str) -> dict:
    """Search the dosing policy."""
    hits = [{"id": k, "content": v, "source": "dosing-policy.md"}
            for k, v in POLICY.items() if k in query.lower()]
    return {"documents": hits}          # recognised as grounding evidence automatically

calculator = LlmAgent(name="dose_calculator", model="gemini-2.0-flash",
                      instruction="Compute daily totals from a per-dose amount and frequency.")
researcher = LlmAgent(name="researcher", model="gemini-2.0-flash",
                      instruction="Answer from search_policy results only. Use dose_calculator for totals.",
                      tools=[FunctionTool(search_policy), AgentTool(agent=calculator)])
coordinator = LlmAgent(name="coordinator", model="gemini-2.0-flash",
                       instruction="Route dosing questions to the researcher.",
                       sub_agents=[researcher])

plugin = VariablyPlugin(
    expected_tools={"researcher": [{"name": "search_policy"}, {"name": "dose_calculator"}]},
    allowed_tools={"researcher": ["search_policy", "dose_calculator"]},
    tags=["dosing-assistant"],
)

async def main():
    sessions = InMemorySessionService()
    runner = Runner(agent=coordinator, app_name="dosing", session_service=sessions, plugins=[plugin])
    session = await sessions.create_session(app_name="dosing", user_id="u1")
    message = types.Content(role="user", parts=[types.Part(text="What is the daily amoxicillin dose?")])
    async for _ in runner.run_async(user_id="u1", session_id=session.id, new_message=message):
        pass
    plugin.flush()                      # a script exits before the background queue drains

asyncio.run(main())

Open the project's Observations experiment and its Sessions tab: one session with three steps — the coordinator as a routing step, the researcher with its search and calculator calls and its retrieved policy text, and the calculator attached as a sub-agent. A clean run passes every check. To see a failure, add "Always state the daily maximum as 3000 mg." to the researcher's instruction and run again: its answer fails grounding with Number not in sources on that line.

What is captured

Each agent run becomes one step of a session keyed by your ADK session id:

Captured	Used for
The agent's final text, model, token usage and latency	Every answer dimension, cost and latency
Every tool call — name, arguments, result or error	Trajectory, tool correctness, loop detection
Handoffs through transfer_to_agent	Trajectory; an agent that only routes is a routing step, so grounding is not applicable rather than a made-up score
Sub-agents wrapped in AgentTool	Attached to the parent's session, so the whole run is one trace
The agent that produced the final response	Marked as the final step (aggregator): its answer is checked against what the other agents produced
Retrieved passages	The evidence grounding is checked against — next section

Grounding evidence

Grounding needs the passages each answer was based on. The plugin collects them automatically from:

• Vertex AI Search, Vertex AI RAG Engine and Google Search — the retrieved chunks on the model response. Google Search exposes page titles and links rather than passage text, so grounding on web results is shallow.
• Your own retriever tools — a result that is a list, or a dict holding a list under documents, results, chunks, passages, contexts or hits, whose items carry text under content, text, snippet, page_content, chunk or passage (with an optional id/doc_id/title and source/uri/url).

For any other shape, tell the plugin how to read the tool's result:

VariablyPlugin(evidence_extractors={
    "search_kb": lambda result: [
        {"id": h["doc_id"], "content": h["body"], "source": h["url"]} for h in result["hits"]
    ],
})

An agent whose tools look like retrieval (search, retriev, rag in the name) but whose answer arrived with no evidence gets a one-time warning, and the step is marked evidence: missing — grounding shows as not applicable with that reason, never as a silent score.

Reviewing documents with ADK? If your agents read whole documents rather than retrieving chunks, return each document from the tool that loads it — {"documents": [{"id": name, "content": text}]} — and set the experiment's Grounding gate to Document review. See Score a document review.

Expected and allowed tools

EXPECTED = [
    {"name": "transfer_to_agent", "arguments": {"agent_name": "researcher"}},
    {"name": "search_policy"},
    {"name": "dose_calculator"},
]
VariablyPlugin(
    expected_tools={"coordinator": EXPECTED, "researcher": EXPECTED},
    allowed_tools={"researcher": ["search_policy", "dose_calculator"]},
)

• expected_tools (per agent) turns on trajectory precision and recall, tool correctness — name and arguments, where you give them — and order. Handoffs are part of the trajectory, so routing to the wrong agent shows up here.
• allowed_tools (per agent) flags any call outside the list as an action outside policy.

Without them, structure is still evaluated — loops, agreement between agents, the final answer's use of the others — but there is no path to compare against.

All options

Option	Default	What it does
expected_tools	none	{agent: [{"name", "arguments"?}]} — the path you expect
allowed_tools	none	{agent: [tool names]} — the allow-list
evidence_extractors	none	{tool: fn(result) -> [{"id", "content", "source"}]} for custom retriever shapes
capture_text	True	False sends structure only — see Sensitive data
tags	none	Added to every step, for filtering in the dashboard
metadata	none	Added to every step
queue_size	1000	Steps waiting for delivery before the oldest are dropped, with a warning

What is checked

Every step gets the answer checks — Grounding, Hallucination-free, Safety — and every session the agent checks: Tool correctness and Trajectory (with expected_tools), No loops, Agents agree, Final answer uses sub-agents, and Within policy (with allowed_tools). Their rules and bars are in Checks and gates.

Deploy to Vertex AI Agent Engine

import vertexai
from vertexai import agent_engines
from variably.integrations.adk import VariablyPlugin

vertexai.init(project=PROJECT, location="us-central1", staging_bucket=BUCKET)
app = agent_engines.AdkApp(agent=root_agent, plugins=[VariablyPlugin()])

remote = agent_engines.create(
    app,
    requirements=[
        "google-cloud-aiplatform[agent_engines,adk]==<your version>",
        "google-adk==<your version>",
        "variably-sdk[adk]",
    ],
    env_vars={
        "VARIABLY_BASE_URL": "https://api.variably.dev",
        "VARIABLY_API_KEY": {"secret": "variably-api-key", "version": "latest"},  # Secret Manager
    },
)

Four things that save a failed deployment:

• Pin your versions. Use the same google-adk and google-cloud-aiplatform versions you built the app with. Unpinned requirements can fail with a generic "failed to deploy" and no logs.
• Grant Vertex AI Search access. Agent Engine runs as service-<PROJECT_NUMBER>@gcp-sa-aiplatform-re.iam.gserviceaccount.com, which cannot query Vertex AI Search by default. Grant it roles/discoveryengine.viewer, or searches fail with 403 discoveryengine.servingConfigs.search — whether or not variA/Bly is installed.
• Let it read the API key secret. The same service account needs roles/secretmanager.secretAccessor on the variably-api-key secret.
• Private networking. For a self-hosted variA/Bly inside your VPC, set VARIABLY_BASE_URL to it and attach the engine with psc_interface_config.

Never in your way

Every hook only observes and is isolated: an error inside the plugin is logged once and never reaches your agents. Delivery runs on a bounded background queue, so a slow or unreachable variA/Bly never slows a request; if the queue fills, the oldest pending steps are dropped with a warning. Long-running servers need nothing else; in short scripts and tests, call plugin.flush() before exiting.

Sensitive data

VariablyPlugin(capture_text=False) sends structure only — agents, tool names, handoffs, timings — with no prompt, response, tool arguments, tool results or document text. Trajectory, loops and policy still score, and tool correctness compares tool names only; grounding and the text checks do not.

Verify it

Run one request and open the Sessions tab of the project's Observations experiment. Then work down this list:

You see	Means	Do
Nothing at all	Not delivered	Check both environment variables; in a script, call plugin.flush()
Steps but grounding not applicable with evidence: missing	No retrieved passages captured	Return a recognised shape, or add an evidence_extractors entry
Tool correctness not applicable	No expected path given	Add expected_tools
A warning that your google-adk version is untested	You are on a major version other than 2.x	It usually works; tell us if a step is missing
delivery queue full — dropped …	Steps produced faster than delivered	Raise queue_size, or check connectivity

Debugging

VARIABLY_LOG_LEVEL=DEBUG python your_app.py

Plugin messages are printed with the [variably.integrations] prefix.

Plugin or OpenTelemetry?

Both work for ADK. The plugin adds expected and allowed tools, custom evidence extractors, the missing-evidence warning and structure-only mode. OpenTelemetry needs no plugin — one line at start-up — and covers every framework you trace in the same process.