DEV Community: Koichi

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

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)