When a developer asks the docs assistant to check several manual pages, Gemini can return one response with a function call per page. Google ADK starts all of those calls together, but whether they actually overlap depends on how each tool is written. Against a docs host that takes 1 s per page, the same 5 calls took 1 s with an async tool and 5.05 s with a blocking one. This is one recorded live run on Gemini gemini-3.8-flash:
Docuctl is the demonstration deployment CLI described by this lab’s bundled manual; the services here use local sample data.
When you need this. The docs assistant checks several manual pages before answering a developer. When the model requests those pages together, awaiting the HTTP calls lets network waits overlap; it does not reduce model latency.
Recorded example. The archived “Make tools parallel-ready” demonstration retains the earlier shipctl scenario; the current lab uses docuctl and does not claim a new model result.
> Check that the rollback, migrations, hooks, approvals and canary pages are reachable.
model turns: [5, 0] function calls per turn
call fetch_page {'page': 'rollback'}
call fetch_page {'page': 'migrations'}
call fetch_page {'page': 'hooks'}
call fetch_page {'page': 'approvals'}
call fetch_page {'page': 'canary'}
result {"page": "rollback", "ok": true, "title": "Roll back a release", "words": 92}
result {"page": "migrations", "ok": true, "title": "Database migrations", "words": 244}
result {"page": "hooks", "ok": true, "title": "Deploy hooks", "words": 227}
result {"page": "approvals", "ok": true, "title": "Approvals", "words": 209}
result {"page": "canary", "ok": false, "status": 404}
server {'requests': 5, 'span_seconds': 1.0, 'max_in_flight': 5}
answer Here is the status of the requested pages: - **rollback**: Reachable ("Roll back a release") - **migrations**: Reachable ("Database migrations") - **hooks**: Reachable ("Deploy hooks") - **approvals**: Reachable ("Approvals") - **canary**: Not reachable (404)1. Start a slow docs host
Download the complete lab ZIP, extract it and use adk-parallel-tools/ as the project directory. Keep the docs files beside the server.
unzip make-tools-parallel-ready.zip
cd adk-parallel-tools
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txt
cp docs_checker/.env.example docs_checker/.envEdit docs_checker/.env and fill GOOGLE_API_KEY privately. GOOGLE_GENAI_USE_ENTERPRISE=0 selects the Gemini API. An already installed local model can instead be selected with DOCS_MODEL=ollama_chat/gemma4:e2b; no command here downloads a model. Starting the docs service needs no model key. If this terminal already exports a different backend or model, unset those values deliberately; native ADK does not override them from .env.
uvicorn docs_server:app --host 127.0.0.1 --port 8770docs_server.py serves the docuctl manual with a fixed delay per page and records when each request started and finished, so overlap is measured on the server side.
"""A slow local docs host: every page takes a second to serve, like a busy wiki or a remote API.
Run: uvicorn docs_server:app --port 8770
Each request records when it started and finished, so overlap between tool calls is measurable.
"""
import asyncio
import time
from pathlib import Path
from fastapi import FastAPI, HTTPException
HERE = Path(__file__).resolve().parent
DELAY_SECONDS = 1.0
app = FastAPI(title="docs host")
SERVED: list[dict] = []
@app.get("/pages/{page}")
async def page(page: str):
path = HERE / "docs" / f"{page}.md"
start = time.perf_counter()
await asyncio.sleep(DELAY_SECONDS) # the server is slow; the client should not wait for one page at a time
SERVED.append({"page": page, "start": start, "end": time.perf_counter()})
if not path.exists():
raise HTTPException(status_code=404, detail=f"no page {page}")
return {"page": page, "text": path.read_text(encoding="utf-8")}2. Write the tool three ways
"""The docs assistant checks several manual pages at once against a slow docs host.
ADK starts every function call from one model response together. Whether they overlap
depends on the tool: an async tool that awaits its I/O does; a blocking one holds the event loop.
"""
import argparse
import asyncio
import os
import sys
import httpx
from google.adk import Agent, Runner
from google.adk.agents.run_config import RunConfig, ToolThreadPoolConfig
from google.adk.models.lite_llm import LiteLlm
from google.adk.sessions import InMemorySessionService
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")
DOCS_URL = os.environ.get("DOCS_URL", "http://127.0.0.1:8770")
def summary(page: str, response: httpx.Response) -> dict:
if response.status_code != 200:
return {"page": page, "ok": False, "status": response.status_code}
text = response.json()["text"]
return {"page": page, "ok": True, "title": text.splitlines()[0].lstrip("# "), "words": len(text.split())}
async def fetch_page(page: str) -> dict:
"""Fetch one page of the docuctl manual, such as rollback or hooks, and report whether it is reachable.
Call this once per page; several calls can run at the same time."""
async with httpx.AsyncClient(timeout=10) as client:
return summary(page, await client.get(f"{DOCS_URL}/pages/{page}")) # await: the loop runs other calls
def fetch_page_blocking(page: str) -> dict:
"""Fetch one page of the docuctl manual, such as rollback or hooks, and report whether it is reachable."""
return summary(page, httpx.get(f"{DOCS_URL}/pages/{page}", timeout=10)) # blocks its thread
async def fetch_page_async_but_blocking(page: str) -> dict:
"""Fetch one page of the docuctl manual, such as rollback or hooks, and report whether it is reachable."""
return summary(page, httpx.get(f"{DOCS_URL}/pages/{page}", timeout=10)) # async def, but nothing is awaited
INSTRUCTION = """You check the docuctl manual for broken pages.
When asked about several pages, call fetch_page once for each page, all in the same response.
Then list each page as reachable or not, with its title. Keep answers under 80 words."""
def make_agent(tool=fetch_page) -> Agent:
instruction = INSTRUCTION.replace("fetch_page", tool.__name__)
return Agent(
name="docs_assistant",
model=LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL,
instruction=instruction,
tools=[tool],
generate_content_config=types.GenerateContentConfig(
temperature=0.2, thinking_config=types.ThinkingConfig(thinking_level="low")),
)
# Sync tools run on worker threads instead of the event loop. Async tools are not affected.
THREAD_POOL = RunConfig(tool_thread_pool_config=ToolThreadPoolConfig(max_workers=4))
async def ask(agent: Agent, text: str, run_config: RunConfig | None = None) -> list:
runner = Runner(app_name="docs", agent=agent, session_service=InMemorySessionService(),
auto_create_session=True)
message = types.Content(role="user", parts=[types.Part(text=text)])
events = [e async for e in runner.run_async(user_id="dana", session_id="s1", new_message=message,
run_config=run_config or RunConfig())]
await runner.close()
return events
def parse_options(argv=None):
parser = argparse.ArgumentParser(description="Check manual pages using a chosen tool and optional sync-tool worker pool.")
parser.add_argument("question", help="question to send to the configured model")
parser.add_argument("--tool", choices=["async", "sync", "blocking-async"], default="async")
parser.add_argument("--workers", type=int, help="max_workers for sync tools; not an overall request limit")
args = parser.parse_args(argv)
if args.workers is not None and args.workers < 1:
parser.error("--workers must be at least 1")
return args
def selected_run(args):
tool = {"async": fetch_page, "sync": fetch_page_blocking,
"blocking-async": fetch_page_async_but_blocking}[args.tool]
config = RunConfig(tool_thread_pool_config=ToolThreadPoolConfig(max_workers=args.workers)) if args.workers else RunConfig()
return make_agent(tool), config
async def main(question: str, agent=None, run_config=None) -> None:
for event in await ask(agent or make_agent(), question, run_config):
for call in event.get_function_calls():
print("call ", call.name, call.args)
if event.is_final_response() and event.content:
print("answer", "".join(p.text or "" for p in event.content.parts if not p.thought))
if __name__ == "__main__":
options = parse_options()
agent, config = selected_run(options)
asyncio.run(main(options.question, agent, config))The three fetch tools return the same result; they differ only in how they wait:
fetch_pageisasync defand awaitshttpx.AsyncClient.get. While one request waits for the server, the event loop runs the others.fetch_page_blockingis a plain function that callshttpx.get. It holds whatever thread runs it until the response arrives.fetch_page_async_but_blockingisasync defbut calls the blockinghttpx.get. It looks async to ADK and to a reviewer, and it blocks the loop anyway.
Two instructions encourage the model to ask for every page in one response. The instruction says so, and the tool description says that several calls can run at once. A model that calls one page per turn pays a full model round trip per page, and no tool change can fix that.
THREAD_POOL is a RunConfig whose ToolThreadPoolConfig sets max_workers to 4. With it, ADK runs sync function tools on worker threads instead of on the event loop. Async tools still run on the loop.
A blocking sync tool. The calls ran one at a time: 1 in flight, 5.05 s. Make the tool async def with an async client, or set ToolThreadPoolConfig.
An async def that blocks. Even with the thread pool: 1 in flight, 5.05 s. Await an async client, or wrap the blocking call in asyncio.to_thread.
3. Replay one response against each tool
First inspect one actual agent turn. In a second terminal, activate the same environment and open the native CLI:
cd /path/to/unpacked/adk-parallel-tools
source .venv/bin/activate
adk run docs_checkerAt [user]:, ask Check rollback, hooks, tokens, deploy and canary. Which pages are reachable?. Look for one fetch_page call per requested page, results including the missing canary page, and a final answer. The model may request a different batch. Exit, then use adk web --host 127.0.0.1 --port 8769 . to inspect the same docs_checker agent and its tool events in the Dev UI.
"""Native ADK entry point for the async tool; the thread-pool experiment uses RunConfig in assistant.py."""
from assistant import make_agent
root_agent = make_agent()The native entry point deliberately uses only the async tool. The sync-tool comparison needs a per-invocation RunConfig; this CLI does not invent a native adk run --workers option. For that experiment, use the Python Runner with explicit tool and worker choices. It does not load docs_checker/.env, so set the model key privately in this terminal first:
export GOOGLE_GENAI_USE_ENTERPRISE=0
# Set GOOGLE_API_KEY privately in this terminal.
python assistant.py --tool sync --workers 4 "Check rollback, hooks, tokens, deploy and canary."--tool async selects the awaiting tool; --tool sync --workers 4 moves the blocking function to four workers; --tool blocking-async --workers 4 demonstrates why a blocking coroutine still stalls the event loop. Each command makes a fresh model request, so different tool batches are not a controlled timing comparison. Use the recorded replay below for that comparison. Stop both local services with Ctrl-C when finished.
The original record.py run asked the question once and then replayed those model turns against each tool, so only the tool changed between measured runs:
# the same model turns replayed; the docs host takes 1.0s per page
async tool span=1.0s in_flight=5 wall=1.06s
sync tool span=5.05s in_flight=1 wall=5.07s
sync tool + thread pool span=2.02s in_flight=4 wall=2.09s
async def that blocks + thread pool span=5.05s in_flight=1 wall=5.07sspan is the time from the first page request to the last reply, measured by the server, and in_flight is the most requests it had open at once:
- The async tool had all
5requests open together, and the batch took about one page’s delay. - The blocking sync tool ran them one after another.
- The thread pool ran the sync tool
4at a time, its worker count. With5calls, the last one waited for a free worker:2.02s. - The async def that blocks got nothing from the pool, because async tools never use it outside live mode.
More calls than workers. 5 calls on 4 workers ran in 2 rounds. Size max_workers to the calls one response can make, and to what the downstream service can take.
fetch_page_async_but_blocking is declared async, and nothing in ADK warns that it blocks. A timing test like the ones below catches it.
Parallel calls also change what the downstream service sees. The 5 calls in one response were 5 requests at once. An asyncio.Semaphore around the request limits simultaneous calls. It does not enforce requests per second; an API with that quota also needs pacing or a rate limiter.
How FunctionTool schedules sync and async calls
When a model response contains several function calls, ADK creates one task per call and gathers their results into a single event, in call order. The live run’s function responses came back as one list, the canary page’s 404 included:
{"page":"rollback","ok":true,"title":"Roll back a release","words":92}
{"page":"migrations","ok":true,"title":"Database migrations","words":244}
{"page":"hooks","ok":true,"title":"Deploy hooks","words":227}
{"page":"approvals","ok":true,"title":"Approvals","words":209}
{"page":"canary","ok":false,"status":404}A task can only give up the event loop at an await. fetch_page awaits the network, so all its tasks are waiting on the server at the same moment. A blocking call inside any tool, sync or async, keeps the loop busy until it returns, and the next task cannot start. The thread pool moves sync tool functions off the loop with run_in_executor, so each blocking call holds a worker thread instead. Invocations on one event loop share a pool when they use the same max_workers. A different worker count creates another pool, so this setting is not a service-wide concurrency limit.
On GIL-enabled CPython, the global interpreter lock limits parallel execution of Python code. Blocking I/O, such as a socket read, releases the lock, so these threads can overlap while waiting. A thread pool does not provide parallel execution of pure Python computation in that runtime; use a process pool or a separate service for that work.
Test the timing without a model
test_assistant.py replays the recorded response against each tool and a shorter server delay, and checks the overlap the server measured.
python -m unittest -v test_assistant.py test_native_entrypoint.pytest_a_missing_page_is_a_result_not_an_exception (test_assistant.ParallelTools.test_a_missing_page_is_a_result_not_an_exception) ... /Users/arjunprabhulal/aiml-lanes/envs/venv-agents/lib/python3.12/site-packages/google/adk/models/llm_request.py:306: UserWarning: [EXPERIMENTAL] feature FeatureName.JSON_SCHEMA_FOR_FUNC_DECL is enabled.
declaration = tool._get_declaration()
ok
test_an_async_def_that_blocks_gets_no_help_from_the_pool (test_assistant.ParallelTools.test_an_async_def_that_blocks_gets_no_help_from_the_pool) ... ok
test_async_tool_calls_overlap (test_assistant.ParallelTools.test_async_tool_calls_overlap) ... ok
test_sync_tool_calls_run_one_at_a_time (test_assistant.ParallelTools.test_sync_tool_calls_run_one_at_a_time) ... ok
test_the_model_asked_for_every_page_in_one_response (test_assistant.ParallelTools.test_the_model_asked_for_every_page_in_one_response) ... ok
test_the_thread_pool_runs_at_most_max_workers_sync_calls (test_assistant.ParallelTools.test_the_thread_pool_runs_at_most_max_workers_sync_calls) ... ok
----------------------------------------------------------------------
Ran 6 tests in 4.892s
OKThe saved output reports 6 replay tests. They compare overlap counts and timing ratios; heavy load can still make timing assertions fail. test_native_entrypoint.py also tests native async-tool loading, the selected worker count reaching RunConfig, invalid arguments and recorder opt-in.
Read the test and recording scripts
"""Replay the recorded parallel function calls against each tool variant and a real slow server, offline.
Run: python -m unittest -v test_assistant.py
"""
import asyncio
import json
import logging
import os
import threading
import time
import unittest
import warnings
from pathlib import Path
warnings.filterwarnings("ignore")
logging.disable(logging.CRITICAL)
os.environ["DOCS_URL"] = "http://127.0.0.1:8771"
import uvicorn # noqa: E402
import assistant # noqa: E402
import docs_server # noqa: E402
from record import QUESTION, ReplayLlm, overlap, renamed # noqa: E402
RECORD = json.loads(Path(__file__).with_name("recorded-runs.json").read_text(encoding="utf-8"))
TURNS = RECORD["live"]["model_turns"]
CALLS = RECORD["live"]["calls_per_model_turn"][0]
DELAY = docs_server.DELAY_SECONDS = 0.3 # a shorter delay keeps the test fast; the ratios are the same
def run(tool, run_config=None):
docs_server.SERVED.clear()
agent = assistant.make_agent(tool).clone(update={"model": ReplayLlm(model="replay", turns=renamed(TURNS, tool.__name__))})
events = asyncio.run(assistant.ask(agent, QUESTION, run_config))
return events, overlap(docs_server.SERVED)
class ParallelTools(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.server = uvicorn.Server(uvicorn.Config(docs_server.app, host="127.0.0.1", port=8771, log_level="warning"))
threading.Thread(target=cls.server.run, daemon=True).start()
while not cls.server.started:
time.sleep(0.05)
@classmethod
def tearDownClass(cls):
cls.server.should_exit = True
def test_the_model_asked_for_every_page_in_one_response(self):
self.assertEqual(CALLS, 5)
self.assertEqual(RECORD["live"]["server"]["max_in_flight"], CALLS)
def test_async_tool_calls_overlap(self):
_, server = run(assistant.fetch_page)
self.assertEqual(server["max_in_flight"], CALLS)
self.assertLess(server["span_seconds"], 2 * DELAY)
def test_sync_tool_calls_run_one_at_a_time(self):
_, server = run(assistant.fetch_page_blocking)
self.assertEqual(server["max_in_flight"], 1)
self.assertGreaterEqual(server["span_seconds"], CALLS * DELAY * 0.95)
def test_the_thread_pool_runs_at_most_max_workers_sync_calls(self):
_, server = run(assistant.fetch_page_blocking, assistant.THREAD_POOL)
workers = assistant.THREAD_POOL.tool_thread_pool_config.max_workers
self.assertEqual(server["max_in_flight"], min(workers, CALLS))
self.assertGreaterEqual(server["span_seconds"], 2 * DELAY * 0.95) # five calls on four workers: two rounds
def test_an_async_def_that_blocks_gets_no_help_from_the_pool(self):
_, server = run(assistant.fetch_page_async_but_blocking, assistant.THREAD_POOL)
self.assertEqual(server["max_in_flight"], 1)
def test_a_missing_page_is_a_result_not_an_exception(self):
events, _ = run(assistant.fetch_page)
results = [r.response for e in events for r in e.get_function_responses()]
self.assertIn({"page": "canary", "ok": False, "status": 404}, results)
self.assertEqual(len(results), CALLS)
if __name__ == "__main__":
unittest.main()python record.py --record explicitly permits a new live model turn followed by controlled replays and overwrites saved files. It is optional authoring work, not setup. Without --record, the script refuses before starting a service or calling a model. Stop the manually started docs host before deliberately recording in a separate copy.
"""Record one parallel tool call on Gemini, then replay it against four tool variants.
Run from this folder with GOOGLE_API_KEY set: python record.py --record
Starts docs_server on 127.0.0.1:8770. Writes recorded-runs.json and run-output.txt.
"""
import asyncio
import json
import threading
import time
import warnings
from pathlib import Path
warnings.filterwarnings("ignore")
import uvicorn # noqa: E402
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 assistant # noqa: E402
import docs_server # noqa: E402
HERE = Path(__file__).resolve().parent
QUESTION = "Check that the rollback, migrations, hooks, approvals and canary pages are reachable."
VARIANTS = [ # (label, tool, run_config)
("async tool", assistant.fetch_page, None),
("sync tool", assistant.fetch_page_blocking, None),
("sync tool + thread pool", assistant.fetch_page_blocking, assistant.THREAD_POOL),
("async def that blocks + thread pool", assistant.fetch_page_async_but_blocking, assistant.THREAD_POOL),
]
class ReplayLlm(BaseLlm):
turns: list
async def generate_content_async(self, llm_request, stream=False):
yield LlmResponse(content=types.Content.model_validate(self.turns.pop(0)))
def serve() -> uvicorn.Server:
server = uvicorn.Server(uvicorn.Config(docs_server.app, host="127.0.0.1", port=8770, log_level="warning"))
threading.Thread(target=server.run, daemon=True).start()
while not server.started:
time.sleep(0.05)
return server
def overlap(served: list[dict]) -> dict:
"""How long the page fetches took end to end, and how many were in flight at once."""
span = max(s["end"] for s in served) - min(s["start"] for s in served)
peak = max(sum(1 for o in served if o["start"] <= s["start"] < o["end"]) for s in served)
return {"requests": len(served), "span_seconds": round(span, 2), "max_in_flight": peak}
def renamed(turns: list, name: str) -> list:
out = json.loads(json.dumps(turns))
for turn in out:
for part in turn.get("parts", []):
if "function_call" in part:
part["function_call"]["name"] = name
return out
async def live() -> dict:
turns = []
def keep(callback_context, llm_response):
if llm_response.content:
content = llm_response.content.model_dump(mode="json", exclude_none=True)
for part in content.get("parts", []):
part.pop("thought_signature", None)
turns.append(content)
docs_server.SERVED.clear()
events = await assistant.ask(assistant.make_agent().clone(update={"after_model_callback": keep}), QUESTION)
calls = [{"name": c.name, "args": c.args} for e in events for c in e.get_function_calls()]
answer = "".join(p.text or "" for p in events[-1].content.parts if not p.thought).strip()
return {"calls": calls, "calls_per_model_turn": [sum("function_call" in p for p in t.get("parts", [])) for t in turns],
"results": [r.response for e in events for r in e.get_function_responses()],
"server": overlap(docs_server.SERVED), "model_turns": turns, "answer": answer}
async def replayed(turns: list, label: str, tool, run_config) -> dict:
docs_server.SERVED.clear()
agent = assistant.make_agent(tool).clone(update={"model": ReplayLlm(model="replay", turns=renamed(turns, tool.__name__))})
start = time.perf_counter()
await assistant.ask(agent, QUESTION, run_config)
return {"variant": label, "tool": tool.__name__, "thread_pool": run_config is not None,
"wall_seconds": round(time.perf_counter() - start, 2), **overlap(docs_server.SERVED)}
async def main() -> None:
server = serve()
run = await live()
matrix = [await replayed(run["model_turns"], *v) for v in VARIANTS]
server.should_exit = True
record = {"model": assistant.MODEL, "server_delay_seconds": docs_server.DELAY_SECONDS, "question": QUESTION,
"live": run, "matrix": matrix}
(HERE / "recorded-runs.json").write_text(json.dumps(record, indent=2) + "\n", encoding="utf-8")
lines = [f"> {QUESTION}", f"model turns: {run['calls_per_model_turn']} function calls per turn"]
lines += [f"call {c['name']} {c['args']}" for c in run["calls"]]
lines += [f"result {json.dumps(r)}" for r in run["results"]]
lines += [f"server {run['server']}", "answer " + run["answer"].replace("\n", " "), "",
f"# the same model turns replayed; the docs host takes {docs_server.DELAY_SECONDS}s per page"]
lines += [f"{m['variant']:<38} span={m['span_seconds']}s in_flight={m['max_in_flight']} wall={m['wall_seconds']}s"
for m in matrix]
(HERE / "run-output.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
print("\n".join(lines))
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Optional live-model recording; overwrites saved run files.")
parser.add_argument("--record", action="store_true", help="explicitly allow live model calls and record replacement")
args = parser.parse_args()
if not args.record:
parser.error("recording is opt-in: use --record; normal lesson commands do not need this script")
asyncio.run(main())- Expose the docs assistant as an MCP server: let other agents and MCP hosts call the whole assistant as one tool.
Check your understanding
Read the explanation
The pool moves synchronous function tools to workers. Declaring async does not make a blocking call yield.
References
Checked October 22, 2026, against adk.dev and google-adk 2.12.0.