When you need this. A model can draft a database migration, but a person should read the SQL before it changes the database. We will pause a workflow with the exact draft attached, collect an explicit decision and continue only along the approved branch.
The example uses a small SQLite search index for the documentation assistant. An agent drafts the SQL. Python checks a narrow statement allowlist, RequestInput pauses the run for review, and the reply selects either apply_migration or record_rejection.
A pause ends the current call to run_async. The session retains the pending request, so a subsequent call using the same session ID can resume it. Our client keeps that session in memory in the same process.
Prepare the approval project
You need Python 3.12 or later and a Gemini API key. Use Download code, extract the ZIP, and open its adk-human-approval-lesson directory. Keep the Python files and fixtures together. All commands in this lesson run from that directory.
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txtOn Windows PowerShell, activate with .venv\Scripts\Activate.ps1.
The Python client reads exported environment variables. The bundled .env.example documents them; it is not loaded by the client. Set GOOGLE_API_KEY privately, then select the Gemini API in this terminal:
unset GOOGLE_GENAI_USE_VERTEXAI DOCS_MODEL
export GOOGLE_GENAI_USE_ENTERPRISE=0Sending a request to the model uses your provider account. The offline checks later in the lesson need no key.
The exercise creates a new local database file. Use an unused filename; the client refuses an existing file before contacting the model. The allowlist accepts only a small set of SQL prefixes and is not a SQL sandbox. Keep this activity on the sample database.
The requested change is a small addition to the sample pages table:
Add a nullable section column to pages so search results can show the section heading, and index it for filtering by section.Write the gate
Define the draft and decision schemas, then join the draft, check, pause and action nodes:
"""Draft a search-index migration with Gemini, pause for a reviewer, then apply or drop it."""
import hashlib
import re
import sqlite3
from contextlib import closing
from typing import Literal
from google.adk import Agent, Event, Workflow
from google.adk.events import RequestInput
from google.genai import types
from pydantic import BaseModel
ALLOWED = re.compile(r"^(ALTER TABLE pages ADD COLUMN|CREATE INDEX) ", re.I)
class Migration(BaseModel):
summary: str
up_sql: list[str]
class Decision(BaseModel):
verdict: Literal["approve", "reject"]
reviewer: str
def create_index(path: str) -> None:
"""The docs search index the migration changes."""
with sqlite3.connect(path) as db:
db.execute("CREATE TABLE IF NOT EXISTS pages (url TEXT PRIMARY KEY, title TEXT, body TEXT)")
db.execute("CREATE TABLE IF NOT EXISTS schema_migrations (id TEXT PRIMARY KEY)")
def check_migration(node_input: Migration):
"""Refuse SQL outside the allowlist before a person is asked to review it."""
refused = [sql for sql in node_input.up_sql if not ALLOWED.match(sql.strip())]
if refused:
return Event(message=f"Refused before review: {refused}") # no route: the run ends
return Event(route="review", output=node_input, state={"migration": node_input.model_dump()})
def request_approval(node_input: Migration):
yield RequestInput(message=f"Apply this migration? {node_input.summary}",
payload=node_input.model_dump(), response_schema=Decision)
def read_decision(node_input: dict):
decision = Decision.model_validate(node_input) # the resume checks fields, not enum values
return Event(route=decision.verdict, output=decision)
def apply_migration(node_input: Decision, migration: dict, db_path: str):
"""Apply once: the change and its migration id commit together, or neither does."""
migration_id = hashlib.sha256("\n".join(migration["up_sql"]).encode()).hexdigest()[:12]
# isolation_level=None stops sqlite3 from opening (and skipping) transactions for us:
# by default it does not open one before DDL, so each ALTER would commit on its own.
with closing(sqlite3.connect(db_path, isolation_level=None)) as db:
db.execute("BEGIN IMMEDIATE") # take the write lock before the check
try:
applied = db.execute("SELECT 1 FROM schema_migrations WHERE id = ?",
(migration_id,)).fetchone()
if not applied:
for sql in migration["up_sql"]:
db.execute(sql)
db.execute("INSERT INTO schema_migrations (id) VALUES (?)", (migration_id,))
db.execute("COMMIT")
except BaseException:
db.execute("ROLLBACK")
raise
if applied:
return Event(message=f"Migration {migration_id} was already applied; nothing to do.")
return Event(message=f"Applied migration {migration_id}, approved by {node_input.reviewer}.")
def record_rejection(node_input: Decision):
return Event(message=f"Migration rejected by {node_input.reviewer}; the index is unchanged.")
def build_workflow(model=None) -> Workflow:
draft = Agent(
name="draft_migration",
model=model or "gemini-3.8-flash", # or LiteLlm(model="ollama_chat/gemma4:e2b")
generate_content_config=types.GenerateContentConfig(max_output_tokens=1024),
output_schema=Migration,
instruction=(
"You write SQLite migrations for the docs search index, a table "
"pages(url TEXT PRIMARY KEY, title TEXT, body TEXT). Return one SQL statement "
"per list item, without a trailing semicolon, and a one-sentence summary."
),
)
return Workflow(
name="migration_gate",
edges=[
("START", draft, check_migration),
(check_migration, {"review": request_approval}),
(request_approval, read_decision),
(read_decision, {"approve": apply_migration, "reject": record_rejection}),
],
)Migration carries a summary and SQL statements. Decision permits only approve or reject and includes a reviewer label. check_migration returns a route only for a permitted draft; a refused draft ends before review.
request_approval yields RequestInput with a message, the draft payload and Decision as its response schema. read_decision validates the returned object again with Pydantic before routing on its verdict. The resume failures explained later show why that explicit validation matters.
apply_migration uses BEGIN IMMEDIATE to take the SQLite write lock before checking whether the SQL hash is in schema_migrations. It applies the statements and writes the hash in one transaction. A failed statement rolls back both; a repeated approval of the same SQL finds the hash and does nothing.
The hash identifies the exact SQL text. It does not detect two differently written migrations that make the same change.
SQL outside the allowlist. A draft with a statement outside the allowed prefixes is refused before review, so no approval prompt appears.
The same migration approved twice. The second approval finds the recorded hash and reports that it was already applied. Repeating the same SQL does not repeat the change.
Two approvals at the same time. BEGIN IMMEDIATE makes the second writer wait. After the first transaction commits, the second sees the hash and makes no change.
A statement fails mid-migration. If a later statement fails, the explicit transaction rolls back the earlier change and leaves no migration row. Without a shared transaction, a partial schema change could remain.
Pause and resume a run
Run the client with a new database filename:
python approval_client.py --reviewer learner --db learner-index.dbRead the printed paused_events, then the review payload. Find summary, up_sql and the adk_request_input call’s ID. Read every statement before typing exactly approve or reject. An invalid entry stays at the prompt without sending a reply.
Approval should add the requested section column; rejection should leave the sample columns url, title, body. Check resumed_events and columns_after rather than assuming the draft was accepted by the allowlist. The model can produce a different draft or one the code refuses before review.
--reviewer learner records a local label. It does not authenticate a person.
"""One real Workflow pause and an interactive review against a NEW local SQLite file.
The in-memory paused session stays in this process. No .env loading or recorder batch.
"""
import argparse
import asyncio
import json
import os
import sqlite3
from pathlib import Path
from google.adk.runners import InMemoryRunner
from harness import reply_to, run
from migration_gate import Decision, build_workflow, create_index
REQUEST = Path(__file__).with_name("migration-request.txt").read_text().strip()
def parse_options(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--request", default=REQUEST)
parser.add_argument("--reviewer", required=True, help="local review label, not authenticated identity")
parser.add_argument("--db", type=Path, default=Path("learner-index.db"))
args = parser.parse_args(argv)
if not args.request.strip() or not args.reviewer.strip(): parser.error("request and reviewer need text")
return args
async def review_payload(payload):
print("Review this actual draft before deciding:")
print(json.dumps(payload, ensure_ascii=False, indent=2))
while True:
verdict = (await asyncio.to_thread(input, "Type approve or reject (Ctrl-C cancels): ")).strip()
if verdict in ("approve", "reject"): return verdict
print("No decision sent: type exactly approve or reject.")
async def run_one(options, model=None, review=review_payload):
# Exclusive creation refuses any existing database BEFORE a model request or SQL change.
with options.db.open("xb"): pass
create_index(str(options.db.resolve()))
workflow = build_workflow(model=model or os.environ.get("DOCS_MODEL") or None)
runner = InMemoryRunner(agent=workflow, app_name="docs_assistant")
try:
paused, session_id = await run(workflow, options.request, runner=runner,
state={"db_path": str(options.db.resolve())})
print(json.dumps({"session_id": session_id, "paused_events": paused}, ensure_ascii=False, indent=2))
calls = [c for e in paused for c in e.get("calls", []) if c["name"] == "adk_request_input"]
resumed = []
if calls:
if len(calls) != 1: raise ValueError("expected one pending review")
call = calls[0]
verdict = await review(call["args"]["payload"])
decision = Decision(verdict=verdict, reviewer=options.reviewer)
resumed, _ = await run(workflow, reply_to(call["id"], decision.model_dump()),
runner=runner, session_id=session_id)
with sqlite3.connect(options.db) as db:
columns = [row[1] for row in db.execute("PRAGMA table_info(pages)")]
result = {"resumed_events": resumed, "columns_after": columns}
print(json.dumps(result, ensure_ascii=False, indent=2))
return result
finally:
await runner.close()
if __name__ == "__main__":
asyncio.run(run_one(parse_options()))The client passes db_path into session state and resumes the same session with a function response. It keeps ownership of the database and the pending review together, which is why we use a Python client for this lesson.
Ctrl+C or EOF before a decision closes the client without applying the pending SQL. Closing the process loses this in-memory session. Use a new --db filename if you repeat the exercise; keep or delete only the sample file you created after the client stops.
Test the gate without a model
python -m unittest -v test_migration_gate.py test_client.pyThe tests supply saved model drafts or explicit fakes, then run the real pause, resume and SQLite paths. The resume test checks that answering a review continues the existing draft instead of generating another one.
Try the rejection path with a separate new sample database. Before you reply, predict which nodes should run and which columns should remain. Compare your prediction with the resumed events and schema.
Read the explanation
ADK matches the function response to a pending request in the existing session history.
Resume a pending RequestInput
This excerpt shows the pause event’s function call and the response schema:
{
"author": "migration_gate",
"node_info": {
"path": "migration_gate@1/request_approval@1"
},
"calls": [
{
"id": "4a223920-5ab3-4a51-b01f-9df7db87284f",
"name": "adk_request_input",
"args": {
"interruptId": "4a223920-5ab3-4a51-b01f-9df7db87284f",
"payload": {
"summary": "Add a nullable section column to the pages table and create an index on it to support filtering by section.",
"up_sql": [
"ALTER TABLE pages ADD COLUMN section TEXT",
"CREATE INDEX idx_pages_section ON pages(section)"
]
},
"message": "Apply this migration? Add a nullable section column to the pages table and create an index on it to support filtering by section.",
"response_schema": {
"properties": {
"verdict": {
"enum": [
"approve",
"reject"
],
"title": "Verdict",
"type": "string"
},
"reviewer": {
"title": "Reviewer",
"type": "string"
}
},
"required": [
"verdict",
"reviewer"
],
"title": "Decision",
"type": "object"
}
}
}
]
}ADK represents the request as a call named adk_request_input, with its ID in long_running_tool_ids. The client sends a user message containing a FunctionResponse with the same ID and name. reply_to in the bundled harness.py builds that message.
On resume, ADK reads the session history, skips finished nodes and continues after the pause with the reply as input. The draft agent is already finished, so it has no reason to run again.
Reply to an unknown interrupt ID. ADK raises ValueError: Function call not found for function response ids. The original pending review remains unanswered.
Reply missing a field. The resume raises WorkflowDataError when a required field such as reviewer is absent.
Reply with an unlisted verdict. In the pinned SDK, a string such as "yes" can pass the resume’s schema handling. Decision.model_validate then rejects it before the graph routes to an action. Keep this application-level validation.
Secure the approval reply
A service built around this example must authenticate reviewers and check their permission before forwarding the response to run_async. The request ID connects a reply to a pending review; possession of an ID is not authorization.
Use a persistent session service if approval must survive a process restart. Keep the session ID and interrupt ID with the review task. The current InMemoryRunner deliberately keeps the exercise within one process.
Catching BaseException around a pause. Let pause exceptions propagate through dynamic parent nodes. ADK uses NodeInterruptedError, a BaseException, to unwind execution during a child pause; swallowing it can let application code continue before a reviewer answers.
Check the review boundary
Before accepting an approval result, find the pending call ID, the explicit reviewer verdict and the resulting schema. A generated draft, a paused run and an applied migration are three different states.
Next, use dynamic workflows when Python needs to decide how many child nodes to start.
References
The pause, resume and transaction paths use the pinned google-adk 2.12.0 SDK and Python’s sqlite3 module.
- ADK human input: requests and replies.
- ADK 2.0 overview: dynamic pauses and exceptions.
- ADK session state: state lifetime.
- Python sqlite3 transaction control: explicit transactions.