Reader feedback lives in PostgreSQL. The docs team wants to ask which pages received unhelpful votes and what readers said. MCP Toolbox exposes two reviewed SQL queries as named tools, and ADK’s ToolboxToolset makes those tools available to the assistant. The model supplies parameter values; PostgreSQL executes the statements as a read-only role. The activity uses an isolated local database seeded from the bundled fixtures.
When you need this. When an agent needs answers from a production database, such as a docs assistant reading feedback or a release bot checking deploy history, we give it a fixed set of reviewed queries instead of a SQL prompt. The queries live in one file the database owners review, and any MCP client, not only this agent, can use the same tools.
Toolbox is a separate server between the agent and the database.
Prepare the database lab
Download the complete lab ZIP, extract it, and open a terminal in adk-mcp-toolbox/. The commands below use that directory and its pinned dependencies. The [toolbox] extra of google-adk installs toolbox-adk, the Toolbox client for ADK; the lock pins 1.4.0. You also need the PostgreSQL 17 binaries (set PG_BIN to their directory) and the Toolbox 0.21.0 toolbox binary on PATH (or TOOLBOX_BIN); the Toolbox introduction lists the install options. To run Gemma locally, set DOCS_MODEL=ollama_chat/gemma4:e2b.
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txtlab.py creates a PostgreSQL 17 cluster in a new temporary directory, on a free port with an explicit check against 5432, loads schema.sql, and starts toolbox in front of it. Ctrl-C stops both and deletes the directory. Its connection settings target that new local cluster. Toolbox listens on 127.0.0.1 only; once it runs as a service, pass credentials= to ToolboxToolset, where toolbox_adk.CredentialStrategy offers workload identity, user identity and API keys, as the toolbox-adk guide describes.
Read lab.py
"""A throwaway PostgreSQL 17 cluster plus an MCP Toolbox server in front of it.
Run: python lab.py # starts both, prints the Toolbox URL, stops and deletes both on Ctrl-C
Set PG_BIN if the PostgreSQL 17 binaries are not in /opt/homebrew/opt/postgresql@17/bin,
and TOOLBOX_BIN if `toolbox` (MCP Toolbox for Databases 0.21.0) is not on PATH.
The cluster lives in a new temporary directory, listens on 127.0.0.1 only and never uses port 5432.
"""
import os
import secrets
import shutil
import socket
import subprocess
import sys
import tempfile
import time
import urllib.request
from pathlib import Path
import psycopg
from psycopg import sql
HERE = Path(__file__).resolve().parent
PG_BIN = Path(os.environ.get("PG_BIN", "/opt/homebrew/opt/postgresql@17/bin"))
TOOLBOX = os.environ.get("TOOLBOX_BIN", shutil.which("toolbox") or "toolbox")
ENV = {**os.environ, "LC_ALL": "C"}
def free_port() -> int:
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
port = probe.getsockname()[1]
assert port != 5432, "never touch the default PostgreSQL port"
return port
def config_error(tools_file: Path) -> str:
"""Start toolbox on a tools file it should refuse and return the first line of its error."""
result = subprocess.run([TOOLBOX, "--tools-file", str(tools_file), "--port", str(free_port())],
capture_output=True, text=True, timeout=20, env={**os.environ, "PG_PORT": "1", "PG_PASSWORD": "x"})
lines = [line for line in (result.stdout + result.stderr).splitlines() if " ERROR " in line]
message = lines[0].split(" ERROR ", 1)[1].split("\\n", 1)[0] if lines else f"exit {result.returncode}"
return message.replace(str(tools_file), tools_file.name)
class Lab:
"""create() starts PostgreSQL and Toolbox; destroy() stops both and deletes the data directory."""
def __init__(self, tools_file: Path = HERE / "tools.yaml"):
self.tools_file = tools_file
self.root = Path(tempfile.mkdtemp(prefix="adk-toolbox-lab-"))
self.pg_port, self.toolbox_port = free_port(), free_port()
self.admin_password, self.reader_password = secrets.token_urlsafe(18), secrets.token_urlsafe(18)
self.toolbox = None
@property
def url(self) -> str:
return f"http://127.0.0.1:{self.toolbox_port}"
def admin(self, database: str = "docs_analytics"):
return psycopg.connect(host="127.0.0.1", port=self.pg_port, dbname=database,
user="lab_admin", password=self.admin_password, autocommit=True)
def create(self) -> "Lab":
try:
pwfile = self.root / "admin-password"
pwfile.write_text(self.admin_password + "\n")
subprocess.run([PG_BIN / "initdb", "-D", self.root / "data", "-U", "lab_admin", "--auth=scram-sha-256",
f"--pwfile={pwfile}", "-E", "UTF8", "--locale=C"], check=True, capture_output=True, env=ENV)
options = f"-p {self.pg_port} -c listen_addresses=127.0.0.1 -c unix_socket_directories=''"
subprocess.run([PG_BIN / "pg_ctl", "-D", self.root / "data", "-o", options, "-l", self.root / "pg.log",
"-w", "start"], check=True, capture_output=True, env=ENV)
with self.admin("postgres") as conn:
conn.execute(sql.SQL("CREATE ROLE toolbox_reader LOGIN PASSWORD {}").format(sql.Literal(self.reader_password)))
conn.execute("CREATE DATABASE docs_analytics")
with self.admin() as conn:
conn.execute((HERE / "schema.sql").read_text())
conn.execute((HERE / "grants.sql").read_text())
self.start_toolbox()
except BaseException:
self.destroy()
raise
return self
def start_toolbox(self) -> None:
env = {**os.environ, "PG_PORT": str(self.pg_port), "PG_PASSWORD": self.reader_password}
self.toolbox = subprocess.Popen(
[TOOLBOX, "--tools-file", str(self.tools_file), "--port", str(self.toolbox_port), "--disable-reload"],
env=env, stdout=open(self.root / "toolbox.log", "w"), stderr=subprocess.STDOUT)
for _ in range(100):
try:
urllib.request.urlopen(self.url + "/", timeout=1)
return
except OSError:
if self.toolbox.poll() is not None:
raise RuntimeError("toolbox exited:\n" + (self.root / "toolbox.log").read_text())
time.sleep(0.1)
raise RuntimeError("toolbox did not start")
def stop_toolbox(self) -> None:
if self.toolbox and self.toolbox.poll() is None:
self.toolbox.terminate()
self.toolbox.wait(timeout=10)
def destroy(self) -> None:
self.stop_toolbox()
if (self.root / "data" / "postmaster.pid").exists():
subprocess.run([PG_BIN / "pg_ctl", "-D", self.root / "data", "-m", "fast", "-w", "stop"],
check=True, capture_output=True, env=ENV)
shutil.rmtree(self.root, ignore_errors=True)
def __enter__(self) -> "Lab":
return self.create()
def __exit__(self, *exc) -> None:
self.destroy()
if __name__ == "__main__":
with Lab() as lab:
print(f"PostgreSQL on 127.0.0.1:{lab.pg_port}; MCP Toolbox at {lab.url}. Ctrl-C stops both.", flush=True)
try:
lab.toolbox.wait()
except KeyboardInterrupt:
pass
sys.exit(0)Define the tools in tools.yaml
Each tool is one SQL statement with typed parameters, and a toolset is a named list of tools. The model gets a tool’s name, description and parameters. Toolbox executes the SQL from this file; the model supplies argument values.
# MCP Toolbox 0.21 configuration: one PostgreSQL source, parameterized SQL tools, two toolsets.
sources:
docs-db:
kind: postgres
host: 127.0.0.1
port: ${PG_PORT}
database: docs_analytics
user: toolbox_reader
password: ${PG_PASSWORD}
tools:
find-pages:
kind: postgres-sql
source: docs-db
description: Find documentation pages of a team whose title contains the given text. Returns slug, title and last update date.
parameters:
- name: team
type: string
description: The team that owns the pages.
- name: title_contains
type: string
description: Text to look for in the page title, such as "rollback".
statement: |
SELECT slug, title, updated_on FROM pages
WHERE team = $1 AND title ILIKE '%' || $2 || '%'
ORDER BY slug LIMIT 20;
page-feedback:
kind: postgres-sql
source: docs-db
description: Count helpful and unhelpful votes per page of a team since a date, with the written comments, worst pages first.
parameters:
- name: team
type: string
description: The team that owns the pages.
- name: since
type: string
description: The first day to count, as YYYY-MM-DD.
- name: limit
type: integer
description: How many pages to return, at most 10.
statement: |
SELECT p.slug, count(*) FILTER (WHERE NOT f.helpful) AS unhelpful, count(*) FILTER (WHERE f.helpful) AS helpful,
array_remove(array_agg(f.comment ORDER BY f.created_on), NULL) AS comments
FROM pages p JOIN feedback f ON f.page_slug = p.slug
WHERE p.team = $1 AND f.created_on >= $2::date
GROUP BY p.slug ORDER BY unhelpful DESC, p.slug LIMIT LEAST($3, 10);
delete-feedback:
kind: postgres-sql
source: docs-db
description: Delete all feedback for a page.
parameters:
- name: page_slug
type: string
description: The page whose feedback to delete.
statement: DELETE FROM feedback WHERE page_slug = $1;
toolsets:
docs-reader:
- find-pages
- page-feedback
docs-admin:
- delete-feedbackToolbox sends $1, $2 and $3 to PostgreSQL as bind parameters, in the order the parameters are listed, so a value such as x'; DROP TABLE pages; -- stays a string: test_sql_in_an_argument_stays_a_value sends it and checks that the search returns no rows and the table is intact. LIMIT LEAST($3, 10) caps the number of page rows at ten for a nonnegative limit. Aggregated comments in a row can still be large; this is not a byte limit. The password comes from ${PG_PASSWORD}, an environment variable that lab.py sets, so the file can be reviewed and committed without secrets.
delete-feedback exists so the failure cases can show it, and it sits in a separate docs-admin toolset. The assistant loads docs-reader only.
The database role is the last boundary, and it does not depend on the YAML being right:
-- The role Toolbox logs in as can read both tables and nothing else.
REVOKE ALL ON SCHEMA public FROM PUBLIC;
GRANT USAGE ON SCHEMA public TO toolbox_reader;
GRANT SELECT ON pages, feedback TO toolbox_reader;Note: the current MCP Toolbox configuration docs describe a newer per-object format, one YAML document per object with kind: source and type: postgres. Toolbox 0.21.0, the version installed here, rejects that format, as the bundled failure test checks. This lesson uses the sources:, tools: and toolsets: maps that 0.21.0 reads; match the format to the version you run.
A write tool loaded by mistake. With docs-admin loaded, the recorded model calls delete-feedback; PostgreSQL returns permission denied (42501) and leaves the rows unchanged. Keep the read-only role even when toolset selection is correct.
A tools file in the newer format. Toolbox 0.21.0 exits with unknown field and serves nothing. Use the maps-format file supported by the pinned binary; the particular field named in the error can vary, so the test checks refusal rather than one message.
Keep the first terminal in this lab directory. After reviewing the YAML and database grants, start the temporary services:
python lab.pyKeep this terminal running. Open a second terminal in the same directory and activate .venv for the agent commands below.
Connect the agent with ToolboxToolset
"""The docs assistant reading the docs analytics database through MCP Toolbox for Databases."""
import asyncio
import os
import sys
from google.adk import Agent, Runner
from google.adk.models.lite_llm import LiteLlm
from google.adk.sessions import InMemorySessionService
from google.adk.tools.toolbox_toolset import ToolboxToolset
from google.genai import types
# Gemini through ADK's native client; set DOCS_MODEL=ollama_chat/gemma4:e2b to run Gemma locally.
MODEL = os.environ.get("DOCS_MODEL", "gemini-3.8-flash")
TOOLBOX_URL = os.environ.get("TOOLBOX_URL", "http://127.0.0.1:5000")
TEAM = "platform-docs" # demo value; a service must derive this from an authorized identity
def docs_db(url: str = TOOLBOX_URL, toolset: str = "docs-reader", bind_team: bool = True) -> ToolboxToolset:
"""Only the docs-reader toolset; team is supplied here and omitted from the model-facing declaration."""
return ToolboxToolset(server_url=url, toolset_name=toolset, bound_params={"team": TEAM} if bind_team else None)
def build_agent(model=None, toolset: ToolboxToolset | None = None) -> Agent:
return Agent(
name="docs_assistant",
model=model or (LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL),
instruction="You help the platform-docs team decide which pages to fix. Answer from the docs analytics "
"database tools only, and quote reader comments exactly. Today is 2026-10-21. "
"Say plainly when you cannot do something with your tools.",
tools=[toolset or docs_db()],
generate_content_config=types.GenerateContentConfig(
temperature=0.2, thinking_config=types.ThinkingConfig(thinking_level="low")),
)
async def ask(text: str, agent: Agent | None = None) -> list:
sessions = InMemorySessionService()
runner = Runner(app_name="docs", agent=agent or build_agent(), session_service=sessions)
try:
session = await sessions.create_session(app_name="docs", user_id="dev")
message = types.Content(role="user", parts=[types.Part(text=text)])
return [e async for e in runner.run_async(user_id="dev", session_id=session.id, new_message=message)]
finally:
await runner.close()
def show(events: list) -> str:
lines = []
for event in events:
for call in event.get_function_calls():
lines.append(f"call {call.name} {call.args}")
for result in event.get_function_responses():
lines.append(f"result {result.name} {result.response}")
if event.is_final_response() and event.author != "user" and event.content:
lines.append("answer " + "".join(p.text or "" for p in event.content.parts if not p.thought).strip())
return "\n".join(lines)
async def list_tools(url: str = TOOLBOX_URL, toolset: str = "docs-reader", bind_team: bool = True) -> list[str]:
tools = docs_db(url, toolset, bind_team)
try:
return [tool.name for tool in await tools.get_tools()]
finally:
await tools.close()
if __name__ == "__main__":
if sys.argv[1:] == ["--tools"]:
print(asyncio.run(list_tools()))
else:
print(show(asyncio.run(ask(" ".join(sys.argv[1:])))))ToolboxToolset(server_url=..., toolset_name="docs-reader") loads one toolset from the server the first time the agent needs its tools. bound_params={"team": "platform-docs"} fills the team parameter in the client, on every call, and removes the parameter from the tool declaration. This example hardcodes the team and has no login. A real service must derive it from an authenticated identity; the value may still appear in the instruction or tool results. Binding is an application control, though: anyone who can call Toolbox directly can pass any team, so enforce tenant isolation in the database as well, with row-level security or per-tenant roles.
Set GOOGLE_API_KEY in the second terminal. Replace PORT with the numeric port printed by lab.py; the complete URL must match that output:
source .venv/bin/activate
export TOOLBOX_URL='http://127.0.0.1:PORT'
python agent.py --tools
python agent.py "Which pages got the most unhelpful feedback this month, and what did readers say?"
python agent.py "Delete the feedback on the rollback page."Follow the calls for the feedback question: page-feedback retrieves votes and comments, then find-pages can resolve page titles. Check the answer against those rows. The bundled search matches substrings: searching for rollback misses the title Roll back a release. An empty search result is a reason to refine the query, not to invent a page. For the deletion question, the assistant should report that no write tool is available.
A Toolbox server that is down. ADK logs the load error and runs the agent without database tools; the recorded answer is empty. Call get_tools() at startup and fail startup if it raises, before accepting user questions.
A bound parameter that no tool uses. The client refuses the whole toolset and ADK continues without it. Bind only parameters declared by its tools, and exercise startup discovery in a test.
Call python agent.py --tools before asking the model a question. This startup discovery check raises when Toolbox cannot be reached or the bound parameters are invalid. Without that check, the agent loop can continue without the expected database tools. The offline suite also loads the separate write tool and checks that PostgreSQL refuses it under the reader role.
Follow tool declarations and bound SQL parameters
Three processes are involved. The agent process runs ADK and the Toolbox client; toolbox is a Go server that holds a connection pool to PostgreSQL and serves each toolset over MCP on HTTP, at /mcp/docs-reader for this one; PostgreSQL runs the statements as toolbox_reader. ToolboxToolset turns each tool from the server into an ADK tool, with a declaration built from the YAML parameters minus the bound ones:
Inspect the saved test fixture
{
"description": "Count helpful and unhelpful votes per page of a team since a date, with the written comments, worst pages first.\n\nArgs:\n limit (int): How many pages to return, at most 10.\n since (str): The first day to count, as YYYY-MM-DD.",
"name": "page-feedback",
"parameters": {
"properties": {
"limit": {
"description": "How many pages to return, at most 10.",
"type": "INTEGER"
},
"since": {
"description": "The first day to count, as YYYY-MM-DD.",
"type": "STRING"
}
},
"required": [
"limit",
"since"
],
"type": "OBJECT"
}
}The result comes back as a JSON string in result, which the model parses. The design decision is where the SQL lives. A generic SQL tool, such as Toolbox’s postgres-execute-sql tool kind, accepts model-written statements within the database role’s permissions. Named tools expose the SQL reviewed in tools.yaml; supporting another query requires a change to that file. The read-only role still controls which tables and operations those statements can access.
Original recording. The saved turns use the earlier shipctl scenario. Replay tests start the local services with the preserved inputs in historical-shipctl-source/; current lesson files use docuctl. No new model response is claimed.
Test against a real database
test_agent.py starts a fresh lab, replays the recorded model turns, and runs Toolbox, PostgreSQL, ToolboxToolset and the agent loop for real.
PG_BIN=/path/to/postgresql/bin python -m unittest -v test_agent.pyRead the offline test
"""Replay the recorded Gemini turns against a real PostgreSQL 17 cluster behind a real MCP Toolbox server.
Run: PG_BIN=/path/to/postgresql@17/bin python -m unittest -v test_agent.py (needs `toolbox` 0.21.0 on PATH)
The model is replaced by its recorded turns; Toolbox, the database, ToolboxToolset and the agent loop are real.
"""
import asyncio
import json
import logging
import re
import unittest
import warnings
from pathlib import Path
warnings.filterwarnings("ignore")
logging.disable(logging.CRITICAL)
from google.adk.models.base_llm import BaseLlm # noqa: E402
from google.adk.models.llm_response import LlmResponse # noqa: E402
from google.genai import types # noqa: E402
import agent # noqa: E402
from lab import HERE, Lab, config_error, free_port # noqa: E402
RECORD = json.loads(Path(__file__).with_name("recorded-runs.json").read_text(encoding="utf-8"))
LAB = None
def setUpModule():
global LAB
warnings.simplefilter("ignore") # unittest resets warning filters before the tests run
import lab
with __import__("unittest.mock",fromlist=["patch"]).patch.object(lab,"HERE",HERE / "historical-shipctl-source"):
LAB = Lab().create()
def tearDownModule():
LAB.destroy()
class ReplayLlm(BaseLlm):
turns: list
requests: list = []
async def generate_content_async(self, llm_request, stream=False):
self.requests.append(llm_request)
yield LlmResponse(content=types.Content.model_validate(self.turns.pop(0)))
def replay(turns: list, toolset) -> tuple[str, ReplayLlm]:
model = ReplayLlm(model="replay", turns=list(turns), requests=[])
return agent.show(asyncio.run(agent.ask("replayed question", agent.build_agent(model, toolset)))), model
def offered(model: ReplayLlm) -> dict:
return {d.name: d for t in model.requests[0].config.tools or [] for d in t.function_declarations}
def feedback_rows() -> int:
with LAB.admin() as conn:
return conn.execute("SELECT count(*) FROM feedback").fetchone()[0]
class ToolboxToolsetTests(unittest.TestCase):
def test_the_reader_toolset_lists_only_read_tools(self):
self.assertEqual(asyncio.run(agent.list_tools(LAB.url)), RECORD["discovery"])
def test_the_feedback_question_replays_against_the_database(self):
run = RECORD["runs"]["feedback"]
transcript, _ = replay(run["model_turns"], agent.docs_db(LAB.url))
self.assertEqual(transcript, run["transcript"])
def test_team_is_bound_and_never_offered_to_the_model(self):
_, model = replay(RECORD["runs"]["delete"]["model_turns"], agent.docs_db(LAB.url))
tools = offered(model)
self.assertEqual(sorted(tools), ["find-pages", "page-feedback"])
self.assertTrue(all("team" not in d.parameters.properties for d in tools.values()))
self.assertTrue(all("team" not in d["parameters"]["properties"] for d in RECORD["runs"]["feedback"]["declarations"].values()))
def test_the_read_only_role_refuses_a_write(self):
run = RECORD["failures"]["read_only_role"]
transcript, _ = replay(run["model_turns"], agent.docs_db(LAB.url, "docs-admin", bind_team=False))
self.assertEqual(transcript, run["transcript"])
self.assertIn("permission denied for table feedback (SQLSTATE 42501)", transcript)
self.assertEqual(feedback_rows(), RECORD["feedback_rows_after"])
def test_a_bound_parameter_no_tool_uses_fails_the_whole_toolset(self):
with self.assertRaises(ValueError) as caught:
asyncio.run(agent.list_tools(LAB.url, "docs-admin"))
self.assertIn("unused bound parameters could not be applied to any tool: team", str(caught.exception))
run = RECORD["failures"]["unused_bound_param"]
_, model = replay(run["model_turns"], agent.docs_db(LAB.url, "docs-admin"))
self.assertEqual(offered(model), {})
self.assertIn("will run without the tools from toolset ToolboxToolset", run["errors"][0])
def test_a_toolbox_server_that_is_down_leaves_the_agent_without_tools(self):
down = f"http://127.0.0.1:{free_port()}"
_, model = replay([{"role": "model", "parts": [{"text": ""}]}], agent.docs_db(down))
self.assertEqual(offered(model), {})
with self.assertRaises(Exception) as caught:
asyncio.run(agent.list_tools(down))
self.assertIn("Cannot connect to host", str(caught.exception))
recorded = RECORD["failures"]["server_down"]
self.assertEqual(re.sub(r"\d{4,5}", "PORT", recorded["errors"][0]).split(" which failed to load: ")[1].split(" ssl")[0],
"Cannot connect to host 127.0.0.1:PORT")
self.assertEqual(recorded["transcript"], "answer ")
def test_sql_in_an_argument_stays_a_value(self):
from toolbox_core import ToolboxClient
async def search():
async with ToolboxClient(LAB.url) as toolbox:
tools = await toolbox.load_toolset("docs-reader", bound_params={"team": agent.TEAM})
find = next(t for t in tools if t._name == "find-pages")
return await find(title_contains="x'; DROP TABLE pages; --")
self.assertEqual(asyncio.run(search()), "null")
with LAB.admin() as conn:
self.assertEqual(conn.execute("SELECT count(*) FROM pages").fetchone()[0], 4)
def test_the_per_object_config_format_is_refused_by_toolbox_0_21(self):
error = config_error(HERE / "tools-v2-format.yaml")
prefix = 'unable to parse tool file at \\"tools-v2-format.yaml\\": ['
self.assertTrue(error.startswith('"' + prefix), error)
self.assertIn("unknown field", error)
self.assertTrue(RECORD["config_error"].startswith('"' + prefix))
if __name__ == "__main__":
unittest.main()Check your understanding
Read the explanation
Use both reviewed query tools and database permissions. Neither tool selection nor parameter binding is a substitute for database authorization.
- Query BigQuery with BigQueryToolset: apply identity and access controls to analytical queries.
References
Official sources checked October 21, 2026, against google-adk 2.12.0, toolbox-adk 1.4.0 and MCP Toolbox 0.21.0.
- ADK: MCP Toolbox for Databases
- MCP Toolbox: introduction
- MCP Toolbox: configuration
- MCP Toolbox: PostgreSQL source
- toolbox-adk package guide
Previously on the blog: Google ADK + MCP Toolbox for Databases (PostgreSQL) covers the ADK 1.x version of this setup.