DEV Community: Koichi

Building an Agent Evaluation Pipeline with Google ADK

Koichi — Mon, 11 May 2026 10:36:31 +0000

Intro

If you've built an agent, you've probably noticed that "it works" is surprisingly hard to prove. LLMs are probabilistic, so the deterministic pass/fail model of unit tests doesn't quite apply. This post walks through why agents need their own evaluation story, and how to build one with ADK in three steps: prepare test cases → prepare criteria → run the evaluation.

The sample code is in this repository:

Koichi73 / adk-tool-trajectory-eval-sample

sample of using tool_trajectory_avg_score in ADK.

adk-tool-trajectory-eval-sample

A minimal sample for evaluating an ADK agent with the tool_trajectory_avg_score criterion.

Layout

.
├── sample_agent/         # the agent under evaluation
│   └── agent.py
└── tests/
    ├── eval_config.json  # evaluation criteria
    ├── evalsets/         # test cases
    └── test_sample_agent.py

Setup

uv sync
cp .env.example .env  # fill in the values

Run the evaluation

uv run pytest

View on GitHub

Why evaluate agents?

In traditional software, tests return a clean pass/fail, which gives you both quality gates and regression detection for free. Agents don't behave that way — the same input can produce different outputs, so deterministic pass/fail alone can't capture quality.

If you ignore that gap and keep shipping, agent quality tends to plateau, and you have no clear signal for where to improve next. In Your AI Product Needs Evals, Hamel Husain calls out three symptoms he saw in products built without an eval system:

Addressing one failure mode led to the emergence of others, resembling a game of whack-a-mole.
There was limited visibility into the AI system’s effectiveness across tasks beyond vibe checks.
Prompts expanded into long and unwieldy forms, attempting to cover numerous edge cases and examples.

He sums up the root cause:

I’ve found that unsuccessful products almost always share a common root cause: a failure to create robust evaluation systems.

The flip side: evals are the foundation that lets you run a quality-improvement loop. Without them, you don't even know which direction to improve in.

What do you actually measure?

LLM evaluation focuses on the quality of the final answer. Agent evaluation has to go further and look at the decision-making process itself. The ADK docs split agent evaluation into two axes:

Trajectory evaluation: analyze the steps the agent took to reach a solution. Which tools did it call, with which arguments, in what order?
Final response evaluation: judge the quality, relevance, and accuracy of the final answer.

Imagine the agent is asked about the weather and gets the answer right — but along the way it called three unrelated tools. The output is right, but the agent's behavior clearly isn't. Trajectory evaluation is how you catch this.

The ADK evaluation pipeline

With ADK, you can break the evaluation pipeline into three steps:

Prepare test cases
Prepare evaluation criteria
Run the evaluation

We'll go through each in turn.

1. Prepare test cases

ADK supports two file formats for writing test cases:

Test file (.test.json)
Eval set file (.evalset.json)

The official docs frame the former as a format for single, simple sessions (a form of unit testing) and the latter as ideal for multi-turn conversations. In practice, the two share a schema, and you can perfectly well write simple tests in an eval set file. I'll use eval set files throughout this post.

An eval set file looks like this:

{
  "eval_set_id": "basic_eval",
  "name": "Basic Agent Evaluation",
  "description": "Sample evaluation set.",
  "eval_cases": [
    {
      "eval_id": "weather_query",
      "conversation": [
        {
          "user_content": {
            "parts": [{"text": "What's the weather in Tokyo?"}]
          },
          "final_response": {
            "parts": [{"text": "Today is sunny in Tokyo."}]
          }
        }
      ],
      "session_input": {
        "app_name": "app",
        "user_id": "eval_user",
        "state": {}
      }
    }
  ]
}

Each case in eval_cases describes a test case through fields such as the user input (user_content) and the expected output (final_response). session_input.state lets you seed the initial state for the run — useful when the agent depends on something like a user profile being present from the start.

2. Prepare evaluation criteria

ADK ships with 11 built-in criteria (and you can define your own) — see the criteria docs for the full list. I'll cover the four most representative ones below.

Criterion	What it checks	Needs ground truth	Uses a judge LLM
`tool_trajectory_avg_score`	Whether the tool-call trajectory matches the expected one	Yes	No
`response_match_score`	ROUGE-1 similarity between the final response and ground truth	Yes	No
`final_response_match_v2`	Semantic match between the final response and ground truth	Yes	Yes
`rubric_based_final_response_quality_v1`	Quality of the final response against a rubric	No	Yes

tool_trajectory_avg_score

Mechanically checks whether the tool-call trajectory (tool name, arguments, order) matches a predefined ground truth. Each call gets a 1.0 (match) or 0.0 (no match), and the final score is the average.

You can choose how matches are determined with match_type. The default is EXACT.

EXACT: the actual trajectory must match exactly — no extra calls, no missing calls.
IN_ORDER: expected calls must appear in the same relative order; additional calls may be interleaved.
ANY_ORDER: all expected calls must appear in any order; additional calls may be interleaved.

response_match_score

Computes the ROUGE-1 similarity between the agent's final response and the ground truth. The score ranges from 0.0 to 1.0; closer to 1.0 is better.

ROUGE-1 tokenizes both strings and computes an F1 score from word overlap. Take this example:

Ground truth: the cat sat on the mat
Output: the cat sat on a mat

Each splits into 6 tokens, and 5 tokens (the, cat, sat, on, mat) overlap. Precision and recall are both 5/6, giving an F1 of about 0.83.

final_response_match_v2

Uses a judge LLM (LLM-as-a-Judge) to evaluate semantic similarity between the agent's final response and the ground truth. Where response_match_score works at the token level, this one asks "are they saying the same thing?" — so it's robust to phrasing differences.

For example, if the ground truth is "It's sunny in Tokyo today." and the output is "Today's Tokyo weather is clear.", the judge will treat them as equivalent. The score ranges from 0.0 to 1.0, and you can set both the pass threshold and the judge model in eval_config.json (covered below).

A note on cost and latency

final_response_match_v2 calls a judge LLM internally, so every evaluation incurs an API cost, and latency is higher than response_match_score. Be mindful of call volume and cost when wiring this into CI.

rubric_based_final_response_quality_v1

The three criteria above all require a predefined "correct answer". This rubric-based criterion is different: instead of defining the answer itself, you define the properties the answer should have (the rubric). The judge LLM rates each rubric as satisfied (1.0) or not (0.0), and the score is the average.

This is a good fit for open-ended tasks where there's no single right answer — summarization, suggestions, conversational responses, and so on. Rubrics look like this:

{
  "criteria": {
    "rubric_based_final_response_quality_v1": {
      "rubrics": [
        {
          "rubric_id": "conciseness",
          "rubric_content": {
            "text_property": "The response is concise and avoids unnecessary preamble or repetition."
          }
        },
        {
          "rubric_id": "intent_inference",
          "rubric_content": {
            "text_property": "Even for ambiguous questions, the response infers and addresses the user's actual intent."
          }
        }
      ]
    }
  }
}

Writing the criteria

I'd recommend putting criteria in eval_config.json (you can also pass them from code, but I won't cover that here). The format:

{
  "criteria": {
    "response_match_score": 0.8
  }
}

The value is the pass threshold. Any test case scoring below it is treated as a failure.

3. Run the evaluation

ADK gives you three ways to run evaluations:

Web UI (adk web)
Programmatic, via pytest
CLI (adk eval)

This post focuses on option 2 — it's easy to put in CI, and the flow is explicit in code. For the others, see the official docs.

Writing evals with pytest

ADK exposes the AgentEvaluator class for this. The basic shape:

from google.adk.evaluation.agent_evaluator import AgentEvaluator
from google.adk.evaluation.eval_config import EvalConfig
from google.adk.evaluation.eval_set import EvalSet
import pytest


@pytest.mark.asyncio
async def test_sample_agent():
    with open("tests/evalsets/basic.evalset.json", "r") as f:
        eval_set = EvalSet.model_validate_json(f.read())

    with open("tests/eval_config.json", "r") as f:
        eval_config = EvalConfig.model_validate_json(f.read())

    await AgentEvaluator.evaluate_eval_set(
        agent_module="sample_agent",
        eval_set=eval_set,
        eval_config=eval_config,
    )

agent_module takes the Python module name (e.g. sample_agent) of the agent you want to evaluate. Load the eval set and config, pass them to evaluate_eval_set, and it runs every case you defined.

AgentEvaluator.evaluate also exists and handles the loading for you, so the code above can be made even shorter. I'm using evaluate_eval_set here on purpose because it keeps the flow visible.

Summary

Agents are probabilistic, so traditional unit and integration tests can't fully guarantee quality. Building an eval setup early gives you the foundation to actually run a quality-improvement loop.
ADK structures the evaluation pipeline into three steps: prepare test cases, prepare criteria, run the evaluation.
There are 11 built-in criteria, ranging from trajectory checks (tool_trajectory_avg_score) to semantic match (final_response_match_v2) to rubric-based judgments (rubric_based_final_response_quality_v1) — pick whichever fits the task.
Because it plugs into pytest, it's easy to put on CI.

A reasonable starting point is to take a small agent, write 2-3 test cases against it, and grow from there as you need more coverage.

I've put a minimal tool_trajectory_avg_score setup in this repo — uv sync → uv run pytest and you're ready to go:

Koichi73 / adk-tool-trajectory-eval-sample

sample of using tool_trajectory_avg_score in ADK.

adk-tool-trajectory-eval-sample

A minimal sample for evaluating an ADK agent with the tool_trajectory_avg_score criterion.

Layout

.
├── sample_agent/         # the agent under evaluation
│   └── agent.py
└── tests/
    ├── eval_config.json  # evaluation criteria
    ├── evalsets/         # test cases
    └── test_sample_agent.py

Setup

uv sync
cp .env.example .env  # fill in the values

Run the evaluation

uv run pytest

View on GitHub

That's it — hope this gives you a useful starting point for your own agent evals!

[ADK] Passing Session State to Remote Agents Over Stateless A2A

Koichi — Tue, 14 Apr 2026 11:41:44 +0000

Intro

ADK ships with a mechanism for calling remote agents over the A2A protocol. It's handy, but A2A is stateless by design, so the client-side state doesn't get carried over to the server side. That's intentional — A2A is built to keep distributed agents loosely coupled.

For ADK developers, though, state is an everyday concept — a shared memory used to pass information between agents. It's a little jarring when something you rely on with local sub-agents just... stops working the moment the agent goes remote.

This post tackles exactly that tension: A2A is stateless, but I still want my state to flow through to the remote agent within the ADK world. I'll walk through a minimal sample that uses the metadata field of A2A requests as a transport channel to inject client-side state into the remote agent's state.

The sample code is in this repository:

Koichi73 / a2a-share-state

A minimal example of sharing session state between ADK agents over the A2A protocol.

ADK A2A Share State

A minimal example of sharing session state between ADK agents over the A2A protocol.

ADK's A2A communication is stateless by design — session state is not shared across process boundaries. This sample demonstrates a pattern to bridge that gap by forwarding client-side session state through A2A request metadata, making it available on the server side for instruction template resolution.

Blog: https://dev.to/koichi73/adk-passing-session-state-to-remote-agents-over-stateless-a2a-bfj

Prerequisites

Python 3.11+
uv
Google Cloud project with Vertex AI API enabled
Application Default Credentials configured (gcloud auth application-default login)

Setup

cp .env.example .env

Fill in .env:

GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1

Install dependencies:

uv sync

Running

Terminal 1 — start the A2A server:

bash start_a2a_server.sh

The server starts at http://localhost:8001.

Terminal 2 — run the client agent:

uv run adk run client
# or use the web UI
uv run adk web

The root agent calls the remote greet_agent and receives a…

View on GitHub

Goal

The goal is to make user_name, which we put into the client-side state, resolvable from {user_name} in the remote agent's instruction.

Concretely, we set state on the client like this:

# Client side
callback_context.state["user_name"] = "Alice"

And we want the remote agent to resolve the following instruction and respond with Hello Alice! How are you doing today?:

# Server side
greet_agent = Agent(
    instruction="""
    Please greet according to the format.
    Hello {user_name}! How are you doing today?
    """,
    ...
)

Architecture

Here's how the pieces fit together. A RequestInterceptor on the client pushes the state into A2A metadata, and a before_agent_callback on the server writes it back into the session state.

There are two key points:

Client side: in RequestInterceptor.before_request, copy ctx.session.state into parameters.request_metadata
Server side: in before_agent_callback, write run_config.custom_metadata['a2a_metadata'] back into callback_context.state

The part where the incoming metadata gets expanded into RunConfig.custom_metadata['a2a_metadata'] on the server is handled automatically by ADK's request_converter, so there's nothing for us to do there.

Implementation

Directory layout

share_state/
├── client/
│   ├── __init__.py
│   └── agent.py          # root_agent + RemoteA2aAgent + RequestInterceptor
├── server/
│   └── remote_agent.py   # greet_agent + before_agent_callback
├── start_a2a_server.sh   # launches the server with uvicorn
├── pyproject.toml
└── .env

Set Vertex AI env vars in .env:

GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1

Include ADK's A2A extras in pyproject.toml:

[project]
dependencies = [
    "google-adk[a2a]>=1.27.4",
    "python-dotenv>=1.2.2",
]

Server side: expand received metadata into state

The server's job is to copy values from the A2A request's metadata back into callback_context.state.

server/remote_agent.py:

from google.adk.a2a.utils.agent_to_a2a import to_a2a
from google.adk.agents import Agent
from google.adk.agents.callback_context import CallbackContext


def _inject_metadata_into_state(callback_context: CallbackContext) -> None:
    """Expand A2A metadata into the session state."""
    run_config = callback_context.run_config
    for key, value in (run_config.custom_metadata.get("a2a_metadata") or {}).items():
        callback_context.state[key] = value

ADK automatically repackages the received MessageSendParams.metadata into RunConfig.custom_metadata['a2a_metadata']. This happens inside google.adk.a2a.converters.request_converter, so no extra plumbing is needed on our end.

Writes to callback_context.state are reflected in the live session.state immediately. before_agent_callback runs before the LLM flow expands the instruction template, so {user_name} is resolved against the value we just wrote.

Next, pass the callback to before_agent_callback and turn the agent into an A2A server with to_a2a():

greet_agent = Agent(
    model="gemini-3-flash-preview",
    name="greet_agent",
    description="Greet according to the format.",
    instruction="""
    Please greet according to the format.
    Hello {user_name}! How are you doing today? 
    """,
    before_agent_callback=_inject_metadata_into_state,
)


a2a_app = to_a2a(greet_agent, port=8001)

Launch the server with uvicorn:

# start_a2a_server.sh
set -a
source .env
set +a

uv run uvicorn server.remote_agent:a2a_app --host localhost --port 8001

Client side: move state into A2A metadata

On the client, we prepare a RequestInterceptor that copies state into request_metadata just before sending the A2A request.

client/agent.py:

from typing import Any
from a2a.types import Message as A2AMessage
from google.adk.a2a.agent.config import (
    A2aRemoteAgentConfig,
    ParametersConfig,
    RequestInterceptor,
)
from google.adk.agents import Agent
from google.adk.agents.callback_context import CallbackContext
from google.adk.agents.invocation_context import InvocationContext
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.adk.tools.agent_tool import AgentTool

async def _forward_state_as_a2a_metadata(
    ctx: InvocationContext,
    a2a_request: A2AMessage,
    parameters: ParametersConfig,
) -> tuple[A2AMessage, ParametersConfig]:
    """before_request: Include all session state keys in the A2A metadata."""
    payload: dict[str, Any] = dict(ctx.session.state)
    if payload:
        parameters.request_metadata = {
            **(parameters.request_metadata or {}),
            **payload,
        }
    return a2a_request, parameters

Whatever you put into parameters.request_metadata is forwarded to a2a.client.send_message(request_metadata=...), and ultimately sent over JSON-RPC as MessageSendParams(message=..., metadata=request_metadata).

Since this is a minimal sample, I'm just dumping the entire ctx.session.state into the metadata without any filtering. In production, I'd recommend whitelisting only the keys you actually want to send.

Next, register the interceptor on the RemoteA2aAgent:

greet_agent = RemoteA2aAgent(
    name="greet_agent",
    description="Greet according to the format.",
    agent_card="http://localhost:8001/.well-known/agent-card.json",
    config=A2aRemoteAgentConfig(
        request_interceptors=[
            RequestInterceptor(before_request=_forward_state_as_a2a_metadata),
        ]
    ),
)

RemoteA2aAgent accepts a list of RequestInterceptors via A2aRemoteAgentConfig, and calls the hooks around each A2A request.

Finally, define root_agent and seed a fixed value into state for the demo:

def _seed_state(callback_context: CallbackContext) -> None:
    """For demo purposes: set a fixed value to the state."""
    callback_context.state.setdefault("user_name", "Alice")


root_agent = Agent(
    name="root_agent",
    model="gemini-3-flash-preview",
    instruction="Please use the tool to greet the user.",
    tools=[AgentTool(greet_agent)],
    before_agent_callback=_seed_state,
)

root_agent invokes the remote agent as a tool via AgentTool(greet_agent). AgentTool.run_async passes the parent's tool_context.state.to_dict() as-is to the sub-runner's new session, so user_name="Alice" is visible to the interceptor via the sub-invocation's ctx.session.state.

Trying it out

Start the A2A server and the client in separate terminals:

# Terminal 1: A2A server
bash start_a2a_server.sh

# Terminal 2: client agent
uv run adk web

From the ADK Web UI, send any message to root_agent. It will call the remote greet_agent and return a response like:

Hello Alice! How are you doing today?

{user_name} has been resolved with "Alice" from the client-side state — proof that the state made it to the server via A2A metadata.

Summary

ADK's A2A communication is stateless by design, so the client-side state doesn't reach the server as-is
You can use the A2A request's metadata field as a transport channel to inject client-side state into the remote agent's instruction template
All you need is a RequestInterceptor.before_request on the client to push state into metadata, and a before_agent_callback on the server to write metadata back into state
ADK automatically expands the incoming metadata into RunConfig.custom_metadata['a2a_metadata'], so the plumbing on each side fits in just a few lines

Sample code is here:

Koichi73 / a2a-share-state

A minimal example of sharing session state between ADK agents over the A2A protocol.

ADK A2A Share State

A minimal example of sharing session state between ADK agents over the A2A protocol.

Blog: https://dev.to/koichi73/adk-passing-session-state-to-remote-agents-over-stateless-a2a-bfj

Prerequisites

Python 3.11+
uv
Google Cloud project with Vertex AI API enabled
Application Default Credentials configured (gcloud auth application-default login)

Setup

cp .env.example .env

Fill in .env:

GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1

Install dependencies:

uv sync

Running

Terminal 1 — start the A2A server:

bash start_a2a_server.sh

The server starts at http://localhost:8001.

Terminal 2 — run the client agent:

uv run adk run client
# or use the web UI
uv run adk web

The root agent calls the remote greet_agent and receives a…

View on GitHub

That's it — hope this helps if you've been wrestling with state and remote agents in ADK!

Injecting Initial State into ADK Web UI with ASGI Middleware

Koichi — Tue, 24 Mar 2026 10:28:58 +0000

If you're using Google's ADK Web UI for agent development, you've probably run into this: every time you create a new session, you have to manually set up the state your agent needs. Here's a clean way to automate that with a preset file and ASGI middleware.

The Problem

ADK's Web UI is great for testing agents, but injecting initial state into a session is a bit of a hassle.

The current options are:

Add state via the 3-dot menu in the UI → You have to manually enter values every time you create a session
Use before_agent_callback with a first-run check → Extra code to write and maintain

Instead, I built a setup where you define initial state in a preset.yaml file, and a custom ASGI middleware automatically merges it into the session when you click "New Session" in the Web UI.

How It Works

When you click "New Session" in the Web UI, it internally calls POST /apps/{app_name}/users/{user_id}/sessions.

The middleware intercepts this request, reads preset.yaml, and merges its initial_state values into the request body's state before passing it to the application.

The key point: you don't need to touch your agent code at all — just add the middleware at server startup.

Setup

Directory Structure

.
├── .venv
├── hello_agent
│   ├── __init__.py
│   ├── agent.py        # Agent definition
│   └── preset.yaml     # Preset config
└── preset_server.py    # Startup script

Target Agent

from google.adk.agents.llm_agent import Agent

root_agent = Agent(
    model='gemini-2.5-flash',
    name='root_agent',
    instruction="""Greet the user in the following format:
    Hello, it's currently {current_time} and the weather is {weather}.
    """,
)

Without initial state, chatting with this agent in the Web UI will fail because current_time and weather don't exist in the session state.

Dependencies

You'll need pyyaml in addition to ADK's dependencies:

pip install pyyaml

Preset File (preset.yaml)

Define your initial state in YAML format:

initial_state:
  current_time: "10:30 AM"
  weather: "sunny"

Nested values are supported too.

Place the file in the same directory as agent.py, or inside an .adk/ directory. If both exist, .adk/preset.yaml takes priority.

Startup Script Implementation

The startup script has three main parts.

1. Loading preset.yaml

PRESET_FILENAME = "preset.yaml"

def find_preset_file(agent_dir: Path) -> Optional[Path]:
    """Search for preset.yaml in the agent directory."""
    candidates = [
        agent_dir / ".adk" / PRESET_FILENAME,
        agent_dir / PRESET_FILENAME,
    ]
    for p in candidates:
        if p.is_file():
            return p
    return None

def load_preset(path: Path) -> dict:
    with open(path, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    return {
        "state": data.get("initial_state", {}),
    }

It checks .adk/preset.yaml first, then falls back to preset.yaml, and returns the initial_state as a dictionary.

2. The Middleware (PresetMiddleware)

This ASGI middleware detects session creation requests and rewrites the body:

SESSION_PATH = re.compile(r"^/apps/([^/]+)/users/[^/]+/sessions$")

class PresetMiddleware:
    def __init__(self, app, agents_dir: Path):
        self.app = app
        self.agents_dir = agents_dir

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        path = scope.get("path", "")
        method = scope.get("method", "")
        m = SESSION_PATH.match(path)

        if method == "POST" and m:
            app_name = m.group(1)
            preset = self._load_preset(app_name)
            if preset and preset.get("state"):
                # Read request body
                body_parts = []
                while True:
                    message = await receive()
                    body_parts.append(message.get("body", b""))
                    if not message.get("more_body", False):
                        break
                body = b"".join(body_parts)

                # Merge state (preset as base, request values override)
                data = json.loads(body) if body else {}
                merged = {**preset["state"], **(data.get("state") or {})}
                data["state"] = merged
                new_body = json.dumps(data).encode()

                # Update Content-Length and forward
                # ... (header rewriting and receive replacement)
                await self.app(scope, new_receive, send)
                return

        await self.app(scope, receive, send)

Here's what it does:

Only targets POST /apps/{app_name}/users/{user_id}/sessions requests
Reads the request body and merges initial_state from preset.yaml into state
Updates Content-Length with the new body size and passes it to the application

Since request-side state takes priority, any values you manually set in the UI won't be overwritten by the preset.

3. CLI (adk web compatible)

The script supports all the same options as adk web, plus a --preset / --no-preset flag to toggle the middleware:

app = get_fast_api_app(
    agents_dir=agents_dir,
    # ... same options as adk web
)

if preset:
    app.add_middleware(PresetMiddleware, agents_dir=agents_dir_path)

4. Full Startup Script

Full startup script

from __future__ import annotations
import sys
import json
import logging
import os
import re
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Optional

import click
import uvicorn
from fastapi import FastAPI

from google.adk.cli.fast_api import get_fast_api_app
from google.adk.cli.utils import logs

try:
    import yaml
except ImportError:
    print("PyYAML is required. Install with: pip install pyyaml")
    sys.exit(1)

logger = logging.getLogger(__name__)

LOG_LEVELS = click.Choice(
    ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
    case_sensitive=False,
)

SESSION_PATH = re.compile(r"^/apps/([^/]+)/users/[^/]+/sessions$")


# ---------------------------------------------------------------------------
# Preset file handling
# ---------------------------------------------------------------------------
PRESET_FILENAME = "preset.yaml"

def find_preset_file(agent_dir: Path) -> Optional[Path]:
    """Search for preset.yaml in .adk/ directory of the agent."""
    candidates = [
        agent_dir / ".adk" / PRESET_FILENAME,
        agent_dir / PRESET_FILENAME,
    ]
    for p in candidates:
        if p.is_file():
            return p
    return None

def load_preset(path: Path) -> dict:
    with open(path, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    return {
        "state": data.get("initial_state", {}),
    }


# ---------------------------------------------------------------------------
# PresetMiddleware (ASGI)
# ---------------------------------------------------------------------------

class PresetMiddleware:
    """ASGI middleware that injects preset state into POST /apps/{app_name}/users/{user_id}/sessions."""

    def __init__(self, app, agents_dir: Path):
        self.app = app
        self.agents_dir = agents_dir

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        path = scope.get("path", "")
        method = scope.get("method", "")
        m = SESSION_PATH.match(path)

        if method == "POST" and m:
            app_name = m.group(1)
            preset = self._load_preset(app_name)
            if preset and preset.get("state"):
                body_parts = []
                while True:
                    message = await receive()
                    body_parts.append(message.get("body", b""))
                    if not message.get("more_body", False):
                        break
                body = b"".join(body_parts)

                data = json.loads(body) if body else {}
                merged = {**preset["state"], **(data.get("state") or {})}
                data["state"] = merged
                new_body = json.dumps(data).encode()

                new_headers = []
                seen_ct = False
                seen_cl = False
                for key, value in scope.get("headers", []):
                    lower_key = key.lower()
                    if lower_key == b"content-length":
                        new_headers.append((b"content-length", str(len(new_body)).encode()))
                        seen_cl = True
                    elif lower_key == b"content-type":
                        new_headers.append((b"content-type", b"application/json"))
                        seen_ct = True
                    else:
                        new_headers.append((key, value))
                if not seen_ct:
                    new_headers.append((b"content-type", b"application/json"))
                if not seen_cl:
                    new_headers.append((b"content-length", str(len(new_body)).encode()))
                scope["headers"] = new_headers

                body_sent = False

                async def new_receive():
                    nonlocal body_sent
                    if not body_sent:
                        body_sent = True
                        return {"type": "http.request", "body": new_body, "more_body": False}
                    return await receive()

                logger.info(
                    "PresetMiddleware: injected state for app=%s keys=%s",
                    app_name,
                    list(preset["state"].keys()),
                )
                await self.app(scope, new_receive, send)
                return

        await self.app(scope, receive, send)

    def _load_preset(self, app_name: str) -> Optional[dict]:
        agent_dir = self.agents_dir / app_name
        preset_path = find_preset_file(agent_dir)
        if preset_path is None:
            return None
        try:
            return load_preset(preset_path)
        except Exception as e:
            logger.warning("PresetMiddleware: failed to load preset for %s: %s", app_name, e)
            return None


# ---------------------------------------------------------------------------
# CLI (adk web compatible)
# ---------------------------------------------------------------------------

@click.command()
@click.argument(
    "agents_dir",
    type=click.Path(exists=True, dir_okay=True, file_okay=False, resolve_path=True),
    default=os.getcwd,
)
@click.option("--host", type=str, default="127.0.0.1", show_default=True, help="The binding host of the server.")
@click.option("--port", type=int, default=8000, help="The port of the server.")
@click.option("--allow_origins", multiple=True, help="Origins to allow for CORS.")
@click.option("-v", "--verbose", is_flag=True, default=False, help="Enable verbose (DEBUG) logging.")
@click.option("--log_level", type=LOG_LEVELS, default="INFO", help="Set the logging level.")
@click.option("--trace_to_cloud", is_flag=True, default=False, help="Enable cloud trace for telemetry.")
@click.option("--otel_to_cloud", is_flag=True, default=False, help="Write OTel data to Google Cloud.")
@click.option("--reload/--no-reload", default=True, help="Enable auto reload for server.")
@click.option("--a2a", is_flag=True, default=False, help="Enable A2A endpoint.")
@click.option("--reload_agents", is_flag=True, default=False, help="Enable live reload for agents changes.")
@click.option("--eval_storage_uri", type=str, default=None, help="Evals storage URI (e.g. gs://bucket).")
@click.option("--extra_plugins", multiple=True, help="Extra plugin classes or instances.")
@click.option("--url_prefix", type=str, default=None, help="URL path prefix for reverse proxy.")
@click.option("--session_service_uri", default=None, help="Session service URI.")
@click.option("--artifact_service_uri", type=str, default=None, help="Artifact service URI.")
@click.option("--memory_service_uri", type=str, default=None, help="Memory service URI.")
@click.option("--use_local_storage/--no_use_local_storage", default=True, show_default=True, help="Use local .adk storage.")
@click.option("--logo-text", type=str, default=None, help="Logo text in web UI.")
@click.option("--logo-image-url", type=str, default=None, help="Logo image URL in web UI.")
# --- preset-specific option ---
@click.option("--preset/--no-preset", default=True, show_default=True, help="Enable PresetMiddleware for auto state injection.")
def cli_preset_web(
    agents_dir: str,
    host: str,
    port: int,
    allow_origins: Optional[tuple[str, ...]],
    verbose: bool,
    log_level: str,
    trace_to_cloud: bool,
    otel_to_cloud: bool,
    reload: bool,
    a2a: bool,
    reload_agents: bool,
    eval_storage_uri: Optional[str],
    extra_plugins: Optional[tuple[str, ...]],
    url_prefix: Optional[str],
    session_service_uri: Optional[str],
    artifact_service_uri: Optional[str],
    memory_service_uri: Optional[str],
    use_local_storage: bool,
    logo_text: Optional[str],
    logo_image_url: Optional[str],
    preset: bool,
):
    """Starts a FastAPI server with Web UI for agents (with preset support).

    AGENTS_DIR: The directory of agents, where each subdirectory is a single
    agent, containing at least __init__.py and agent.py files.

    Example:

      python preset_server.py path/to/agents_dir
      python preset_server.py . --port 8080 --no-preset
    """
    if verbose and log_level == "INFO":
        log_level = "DEBUG"

    logs.setup_adk_logger(getattr(logging, log_level.upper()))

    agents_dir_path = Path(agents_dir).resolve()

    @asynccontextmanager
    async def _lifespan(app: FastAPI):
        preset_msg = " + PresetMiddleware" if preset else ""
        click.secho(
            f"""
+-----------------------------------------------------------------------------+
| ADK Web Server started{preset_msg:<53s}|
|                                                                             |
| For local testing, access at http://{host}:{port}.{" " * (29 - len(str(port)))}|
+-----------------------------------------------------------------------------+
""",
            fg="green",
        )
        yield
        click.secho(
            """
+-----------------------------------------------------------------------------+
| ADK Web Server shutting down...                                             |
+-----------------------------------------------------------------------------+
""",
            fg="green",
        )

    app = get_fast_api_app(
        agents_dir=agents_dir,
        session_service_uri=session_service_uri,
        artifact_service_uri=artifact_service_uri,
        memory_service_uri=memory_service_uri,
        use_local_storage=use_local_storage,
        eval_storage_uri=eval_storage_uri,
        allow_origins=list(allow_origins) if allow_origins else None,
        web=True,
        trace_to_cloud=trace_to_cloud,
        otel_to_cloud=otel_to_cloud,
        lifespan=_lifespan,
        a2a=a2a,
        host=host,
        port=port,
        url_prefix=url_prefix,
        reload_agents=reload_agents,
        extra_plugins=list(extra_plugins) if extra_plugins else None,
        logo_text=logo_text,
        logo_image_url=logo_image_url,
    )

    if preset:
        app.add_middleware(PresetMiddleware, agents_dir=agents_dir_path)
        click.secho(
            "PresetMiddleware: auto-injecting preset state into POST /apps/*/users/*/sessions",
            fg="cyan",
        )

    config = uvicorn.Config(
        app,
        host=host,
        port=port,
        reload=reload,
    )
    server = uvicorn.Server(config)
    server.run()


if __name__ == "__main__":
    cli_preset_web()

Testing It Out

Run the startup script:

python preset_server.py .

If the middleware is active, you'll see this in the startup log:

PresetMiddleware: auto-injecting preset state into POST /apps/*/users/*/sessions

Click "New Session" in the Web UI, and the state gets injected automatically:

Ask the agent for a greeting, and you'll see the values from preset.yaml reflected in the response:

Wrap-Up

Define initial_state in preset.yaml and it gets auto-injected on session creation
Since it's implemented as ASGI middleware, no changes to your agent code are needed
Fully compatible with all adk web options

That's it — no more copy-pasting state values every time you start a new session.

How to Run Gemini 3 on Agent Engine with ADK

Koichi — Thu, 19 Mar 2026 05:14:31 +0000

If you're trying to deploy an ADK agent with Gemini 3 to Agent Engine and getting a 404 error, here's the fix.

TL;DR

Two ways to make it work:

1. `adk deploy agent_engine` command

Pass a .env file containing GOOGLE_CLOUD_LOCATION=global via the --env_file flag.

uv run adk deploy agent_engine \
        --project=$PROJECT_ID \
        --region=$LOCATION_ID \
        --env_file=$ENV_FILE \
        --display_name="Greeting Agent" \
        sample_agent

2. Python script

Add the following to the agent_engines.create parameters:

remote_agent = agent_engines.create(
    agent_engine=root_agent,
    display_name="Greeting Agent",
    requirements=[...],
    env_vars={
        "GOOGLE_CLOUD_LOCATION": "global",  # ← set the global region
    },
)

Sample code

Koichi73 / adk-gemini-3-in-agent-engine

How to run ADK agents, including Gemini 3, on Agent Engine.

Run Gemini 3 Series on ADK + Agent Engine

A sample project for deploying ADK agents with Gemini 3 series models (e.g., gemini-3-flash-preview) to Agent Engine.

To use Gemini 3 series on Agent Engine, you need to pass the environment variable GOOGLE_CLOUD_LOCATION=global. This repository demonstrates two ways to do that.

Prerequisites

Python 3.11+
uv
Google Cloud project with Vertex AI API enabled
GCS bucket for staging

Setup

cp .env.example .env
# Edit .env and set GOOGLE_CLOUD_PROJECT, BUCKET_NAME

uv sync

Deployment

Option 1: Shell script (`adk deploy` CLI)

Pass the path to a .env file containing GOOGLE_CLOUD_LOCATION=global via the --env_file option of adk deploy agent_engine.

bash deploy.sh

# Relevant part of deploy.sh
uv run adk deploy agent_engine \
        --project=$PROJECT_ID \
        --region=$LOCATION_ID \
        --env_file=$ENV_FILE \    # ← path to .env file containing GOOGLE_CLOUD_LOCATION=global
        --display_name="Greeting Agent" \
        sample_agent

Option 2: Python script

Pass…

View on GitHub

Why does this happen?

Gemini 3 models (as of March 2026) are only available in the global region
Agent Engine deployments don't let you specify global as the region directly
Without this workaround, ADK agents deployed to Agent Engine will hit a 404 error — the model can't be found

Result

Agent definition:

from google.adk.agents.llm_agent import Agent

MODEL = "gemini-3-flash-preview"

root_agent = Agent(
    model=MODEL,
    name='root_agent',
    description="greeting",
    instruction="Please greet me cheerfully.",
)

Agent Engine playground:

That's it — hope this saves you some debugging time!

DEV Community: Koichi

Building an Agent Evaluation Pipeline with Google ADK

Intro

Koichi73 / adk-tool-trajectory-eval-sample

sample of using tool_trajectory_avg_score in ADK.

adk-tool-trajectory-eval-sample

Layout

Setup

Run the evaluation

Why evaluate agents?

What do you actually measure?

The ADK evaluation pipeline

1. Prepare test cases

2. Prepare evaluation criteria

tool_trajectory_avg_score

response_match_score

final_response_match_v2

rubric_based_final_response_quality_v1

Writing the criteria

3. Run the evaluation

Writing evals with pytest

Summary

Koichi73 / adk-tool-trajectory-eval-sample

sample of using tool_trajectory_avg_score in ADK.

adk-tool-trajectory-eval-sample

Layout

Setup

Run the evaluation

[ADK] Passing Session State to Remote Agents Over Stateless A2A

Intro

Koichi73 / a2a-share-state

A minimal example of sharing session state between ADK agents over the A2A protocol.

ADK A2A Share State

Prerequisites

Setup

Running

Goal

Architecture

Implementation

Directory layout

Server side: expand received metadata into state

Client side: move state into A2A metadata

Trying it out

Summary

Koichi73 / a2a-share-state

A minimal example of sharing session state between ADK agents over the A2A protocol.

ADK A2A Share State

Prerequisites

Setup

Running

Injecting Initial State into ADK Web UI with ASGI Middleware

The Problem

How It Works

Setup

Directory Structure

Target Agent

Dependencies

Preset File (preset.yaml)

Startup Script Implementation

1. Loading preset.yaml

2. The Middleware (PresetMiddleware)

3. CLI (adk web compatible)

4. Full Startup Script

Testing It Out

Wrap-Up

How to Run Gemini 3 on Agent Engine with ADK

TL;DR

1. adk deploy agent_engine command

2. Python script

Sample code

Koichi73 / adk-gemini-3-in-agent-engine

How to run ADK agents, including Gemini 3, on Agent Engine.

Run Gemini 3 Series on ADK + Agent Engine

Prerequisites

Setup

Deployment

Option 1: Shell script (adk deploy CLI)

Option 2: Python script

Why does this happen?

Result

1. `adk deploy agent_engine` command

Option 1: Shell script (`adk deploy` CLI)