Some agents should run when something happens, not when someone types. ADK calls these ambient agents, and its API server has endpoints for the two Google Cloud event sources: /apps/{app}/trigger/pubsub for Pub/Sub push subscriptions and /apps/{app}/trigger/eventarc for Eventarc’s CloudEvents. Here you turn the docs assistant into a docs reviewer that runs when a page changes, serve it with those endpoints on your machine, and send it the exact request bodies Pub/Sub and Eventarc send. ADK 2.12 refuses to serve the endpoints without an authentication check, so that comes first. These runs were recorded on Gemini gemini-3.8-flash, and no Google Cloud service is called.
When you need this. The docuctl docs live in a repository, and its CI publishes a “docs changed” message for every merged page. A reviewer agent should read each changed page and post a one-sentence summary to the docs team’s log, with nobody watching it run. That needs an endpoint the event source can call, a way to reject anyone else, and a tool that recognizes a repeated delivery, which Pub/Sub does by design.
Recorded example. The archived “Run an agent from event triggers” demonstration retains the earlier shipctl scenario; the current lab uses docuctl and does not claim a new model result.
$ TRIGGER_TOKEN=local-dev-token python main.py
$ python publish.py pubsub tokens --message-id 1001
HTTP 200 {"status": "success"} (5.8 s)
$ python publish.py pubsub tokens --message-id 1001
HTTP 200 {"status": "success"} (6.8 s)
$ python publish.py eventarc rollback --event-id ev-77
HTTP 200 {"status": "success"} (5.3 s)
$ python publish.py pubsub tokens --message-id 1002 --raw-data 'not base64!'
HTTP 400 {"detail": "Invalid base64 message data: Invalid base64-encoded string: number of data characters (9) cannot be 1 more than a multiple of 4"} (0.0 s)
$ TRIGGER_TOKEN=wrong-token python publish.py pubsub tokens --message-id 1004
HTTP 401 {"detail": "Missing or wrong trigger token."} (0.0 s)Lab setup. Download the complete lab ZIP, extract it and open a terminal in adk-triggers/. Keep the bundled code, tests and fixtures together. Create the pinned environment before running the commands below:
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txt1. Write the reviewer agent
The agent gets the event as its user message, reads the changed page and calls post_review. Its output goes to reviews.jsonl, standing in for whatever channel your team reads, because nobody sees an ambient agent’s replies:
"""An ambient docs reviewer: a docs-changed event arrives, the agent reads the page and posts a review."""
import json
import os
from pathlib import Path
from filelock import FileLock
from google.adk import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.tools.tool_context import ToolContext
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")
HERE = Path(__file__).resolve().parent.parent
REVIEWS = Path(os.environ.get("DOCS_REVIEWS", HERE / "reviews.jsonl")) # the output channel nobody watches live
def read_page(page: str) -> dict:
"""Read one page of the docuctl documentation: config, deploy, rollback or tokens."""
if page not in {"config", "deploy", "rollback", "tokens"}:
return {"error": f"No page {page!r}."}
path = HERE / "docs" / f"{page}.md"
return {"page": page, "text": path.read_text(encoding="utf-8")} if path.is_file() else {"error": f"No page {page!r}."}
def post_review(page: str, summary: str, tool_context: ToolContext) -> dict:
"""Post the review of a changed docs page to the docs team's review log.
Args:
page: The page that changed.
summary: One sentence on what a developer must know about the page.
"""
# Trigger endpoints store the delivery (source, message ID) in session state. A redelivered
# message runs in a new session, so dedupe on the message ID where the side effect happens.
delivery = tool_context.state.get("trigger_delivery") or {}
key = f"{delivery['source']}:{delivery['id']}" if delivery.get("source") and delivery.get("id") else None
if not key:
raise ValueError("post_review requires a trigger delivery ID")
# All local processes must use this same lock path. This is not a remote API guarantee.
with FileLock(str(REVIEWS) + ".lock", timeout=10):
lines = REVIEWS.read_text(encoding="utf-8").splitlines() if REVIEWS.exists() else []
if any(json.loads(line)["key"] == key for line in lines):
return {"posted": False, "reason": "duplicate delivery"}
with REVIEWS.open("a", encoding="utf-8") as f:
f.write(json.dumps({"key": key, "page": page, "summary": summary}) + "\n")
return {"posted": True}
root_agent = Agent(
name="docs_reviewer",
model=LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL,
description="Reviews docuctl docs pages when they change.",
instruction="""You receive docs-change events as JSON: {"data": {"page": ..., "commit": ...}, "attributes": {...}}.
Read the changed page with read_page, then call post_review once with a one-sentence summary
of what a developer must know from that page. Then reply with the single word: done.""",
tools=[read_page, post_review],
generate_content_config=types.GenerateContentConfig(
temperature=0.2, thinking_config=types.ThinkingConfig(thinking_level="low")),
)2. Serve the trigger endpoints behind an auth check
The ADK page on ambient agents starts the endpoints with adk api_server --trigger_sources "pubsub". In 2.12 that command fails:
$ adk api_server --trigger_sources pubsub,eventarc --port 8767 .
ValueError: trigger_sources requires trigger_oidc_audience or trigger_auth_verifier to be set.
exit 1The installed package requires either --trigger_oidc_audience, which accepts only Google-signed OIDC tokens for that audience, or a trigger_auth_verifier callable, which only the Python entry point accepts. Pub/Sub and Eventarc attach Google-signed tokens when you configure their push subscriptions with a service account, so a deployed service uses the audience flag. An audience match does not select the allowed sender: also set --trigger_oidc_service_accounts to the push service account emails you trust. The installed verifier checks their verified email claim. A local machine has no such tokens, so main.py passes a verifier that checks a shared bearer token:
"""Serve the docs reviewer with Pub/Sub and Eventarc trigger endpoints on this machine.
Run from this folder: TRIGGER_TOKEN=local-dev-token python main.py
ADK 2.12 refuses trigger endpoints without an authentication check. In Google Cloud that check is
--trigger_oidc_audience (Google-signed OIDC tokens); here it is a shared bearer token for local work.
"""
import hmac
import os
from pathlib import Path
import uvicorn
from fastapi import HTTPException, Request
from google.adk.cli.fast_api import get_fast_api_app
AGENTS_DIR = str(Path(__file__).resolve().parent)
TOKEN = os.environ.get("TRIGGER_TOKEN", "")
def verify_shared_token(request: Request) -> None:
"""Runs before every /apps/{app}/trigger/* request; raising rejects the delivery."""
scheme, _, token = request.headers.get("Authorization", "").partition(" ")
if not TOKEN or scheme.lower() != "bearer" or not hmac.compare_digest(token, TOKEN):
raise HTTPException(status_code=401, detail="Missing or wrong trigger token.")
app = get_fast_api_app(
agents_dir=AGENTS_DIR,
web=False,
session_service_uri=os.environ.get("SESSION_SERVICE_URI", "memory://"),
trigger_sources=["pubsub", "eventarc"],
trigger_auth_verifier=verify_shared_token,
)
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=int(os.environ.get("PORT", "8767")))get_fast_api_app is what adk api_server calls; trigger_sources adds the two routes, and the verifier runs before each of them. hmac.compare_digest avoids a character-by-character early exit during this token comparison. The whole HTTP request has other timing behavior; this is not a security assessment of the server.
Run the server command below from the extracted lab directory, with GOOGLE_API_KEY in its shell. Keep that terminal running and open a second terminal in the same directory for publish.py with the same TRIGGER_TOKEN value:
TRIGGER_TOKEN=local-dev-token python main.pyTrigger endpoints without an auth check. adk api_server --trigger_sources pubsub,eventarc exits 1 with trigger_sources requires trigger_oidc_audience or trigger_auth_verifier to be set. Pass --trigger_oidc_audience in Google Cloud, or a trigger_auth_verifier through get_fast_api_app.
A missing or wrong token. The shared-token check returned HTTP 401; with --trigger_oidc_audience, a request without a token got HTTP 401 Missing or malformed Authorization bearer token. Send the token the server expects; in Google Cloud, configure the push subscription’s service account and audience.
3. Deliver Pub/Sub and Eventarc events
publish.py builds the request bodies the two sources send. A Pub/Sub push request wraps the payload as base64 in message.data, with attributes, a message ID and the subscription name. Eventarc’s default binary content mode puts the CloudEvents attributes in ce-* headers and the Pub/Sub message in the body:
"""Deliver a docs-changed event to the agent's trigger endpoints, shaped as Pub/Sub or Eventarc sends it.
Start the server first: TRIGGER_TOKEN=local-dev-token python main.py
Then: TRIGGER_TOKEN=local-dev-token python publish.py pubsub tokens --message-id 1001
python publish.py eventarc rollback --event-id ev-77
"""
import argparse
import base64
import json
import os
import httpx
BASE_URL = os.environ.get("ADK_URL", "http://127.0.0.1:8767")
APP = "docs_reviewer"
TOKEN = os.environ.get("TRIGGER_TOKEN", "")
def pubsub_body(page: str, message_id: str, raw_data: str | None = None) -> dict:
"""A Pub/Sub push request: the payload is base64 in message.data."""
data = raw_data if raw_data is not None else base64.b64encode(
json.dumps({"page": page, "commit": "4f2a9c1"}).encode()).decode()
return {"message": {"data": data, "attributes": {"repo": "docuctl-docs"}, "messageId": message_id,
"publishTime": "2026-10-21T18:00:00Z"},
"subscription": "projects/docs/subscriptions/docs-changed"}
def eventarc_headers(event_id: str) -> dict:
"""Eventarc binary content mode: CloudEvents attributes travel as ce-* headers."""
return {"ce-id": event_id, "ce-type": "google.cloud.pubsub.topic.v1.messagePublished",
"ce-source": "//pubsub.googleapis.com/projects/docs/topics/docs-changed", "ce-specversion": "1.0"}
def deliver(client: httpx.Client, source: str, page: str, event_id: str, raw_data: str | None = None,
token: str | None = None) -> httpx.Response:
body = pubsub_body(page, event_id, raw_data)
auth = {"Authorization": f"Bearer {TOKEN if token is None else token}"}
if source == "pubsub":
return client.post(f"/apps/{APP}/trigger/pubsub", json=body, headers=auth)
return client.post(f"/apps/{APP}/trigger/eventarc", json={"message": body["message"]},
headers={**auth, **eventarc_headers(event_id)})
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("source", choices=["pubsub", "eventarc"])
parser.add_argument("page")
parser.add_argument("--message-id", "--event-id", dest="event_id", default="1001")
parser.add_argument("--raw-data", help="send this string as message.data instead of base64 JSON")
args = parser.parse_args()
with httpx.Client(base_url=BASE_URL, timeout=120) as client:
r = deliver(client, args.source, args.page, args.event_id, args.raw_data)
print(f"HTTP {r.status_code} {r.text}")
if __name__ == "__main__":
main()TRIGGER_TOKEN=local-dev-token python publish.py pubsub tokens --message-id 1001
TRIGGER_TOKEN=local-dev-token python publish.py pubsub tokens --message-id 1001
TRIGGER_TOKEN=local-dev-token python publish.py eventarc rollback --event-id ev-77The output at the top shows each request held open until the agent finished, about 5.8 s on one machine, then {"status": "success"}. The agent received this user message, built by the endpoint from the decoded payload:
{"data": {"page": "tokens", "commit": "4f2a9c1"}, "attributes": {"repo": "shipctl-docs"}}Malformed base64 data. HTTP 400 Invalid base64 message data, and the agent never runs. Pub/Sub treats a non-success status as a negative acknowledgement and may retry it. Fix the submitted body’s encoding and repeat the local request. For a real Pub/Sub subscription, separately configure and test a dead-letter policy for repeated delivery failures.
A rate-limited model. A 429 RESOURCE_EXHAUSTED error from the model is retried with backoff, and the delivery succeeds if a retry does. Keep ADK_TRIGGER_MAX_CONCURRENT below what your model quota sustains.
A failing agent. Any other agent error returns HTTP 500 Agent processing failed. The local fixture observes that response; it has no source that redelivers it. A deployed source may retry according to its delivery policy, so test idempotent handling and dead-letter monitoring before connecting one.
A page path outside the named docs. The tool rejects ../secret before reading a file. Keep the explicit page allowlist.
Missing delivery identity. The tool refuses to post without a source and message ID. Call it through the trigger route, which supplies trigger_delivery.
$ adk api_server --trigger_sources pubsub --trigger_oidc_audience https://docs-reviewer.example --port 8768 --session_service_uri memory:// .
$ curl -s -X POST -H 'Content-Type: application/json' -d @event.json http://127.0.0.1:8768/apps/docs_reviewer/trigger/pubsub
HTTP 401 {"detail": "Missing or malformed Authorization bearer token."}The 429 and 500 cases come from the offline tests, which make the replay model raise those errors; the live runs did not hit a rate limit.
4. Keep a redelivery from posting twice
Message 1001 was delivered twice, as Pub/Sub does when an acknowledgement is lost. Each delivery ran the agent in a new session, so session state cannot remember the first attempt. What both sessions share is the delivery identity the endpoint stores in state:
{
"trigger_delivery": {
"source": "pubsub",
"id": "1001",
"subscription": "projects/docs/subscriptions/docs-changed",
"publish_time": "2026-10-21T18:00:00Z"
}
}post_review builds its key from source and id and holds a cross-process file lock while it checks and appends to the local log, so the second run found pubsub:1001 and posted nothing:
$ cat reviews.jsonl
{"key": "pubsub:1001", "page": "tokens", "summary": "CI jobs authenticate using deploy tokens stored as SHIPCTL_TOKEN, which are created with specific scopes and expire after 90 days."}
{"key": "eventarc:ev-77", "page": "rollback", "summary": "`shipctl rollback` points traffic to an earlier release without rebuilding, records the rollback as a new release, and does not automatically roll back database migrations."}The agent still ran twice and made model calls twice; the key only stops the side effect. The concurrent-call regression checks that overlapping calls post one review for a key. This local log still has limits: a crash during append can leave an incomplete line, and the key assumes one Pub/Sub subscription or Eventarc source. For multiple sources, include the subscription or CloudEvent source in the key. With a remote tracker, use its documented idempotency mechanism if it has one; a local file lock cannot protect a remote write.
A redelivered message. A second delivery of the same message ran the agent again in a new session. Deduplicate side effects on the trigger_delivery source and ID.
Concurrent local writes. An unlocked lookup and append can post twice. The local log now locks both operations. Keep the same lock path for every local writer; use the remote service’s idempotency mechanism for remote writes.
How a delivery starts the configured agent
A trigger endpoint maps one delivery to one agent run. app_name comes from the URL; the user ID comes from the Pub/Sub subscription or the CloudEvent source, with slashes replaced, so the recorded Pub/Sub sessions belong to user projects--docs--subscriptions--docs-changed. The session ID is a new UUID per delivery. The endpoint decodes message.data, parses it as JSON when it can, and sends {"data": ..., "attributes": ...} as the user message.
The HTTP status is the acknowledgement. The endpoint waits for the run, then returns 200, which tells Pub/Sub or Eventarc the message is done. A rate-limit error from the model (a message with 429 or RESOURCE_EXHAUSTED) is retried inside the endpoint with exponential backoff and jitter, 3 retries by default (ADK_TRIGGER_MAX_RETRIES); any other error returns 500, and the source redelivers later under its own retry policy. A semaphore caps concurrent runs per process at 10 by default (ADK_TRIGGER_MAX_CONCURRENT). It limits simultaneous work, not requests or tokens per minute; your model quota may still be exceeded. Because the request stays open for the whole run, the docs limit trigger work to what finishes within the sources’ 10-minute delivery deadline.
Note: Connecting real sources needs a Google Cloud project: a Pub/Sub push subscription or an Eventarc trigger pointing at the deployed service, with a service account whose OIDC tokens match --trigger_oidc_audience, and usually a dead-letter topic for messages that keep failing. The ADK page deploys with adk deploy cloud_run --trigger_sources="pubsub,eventarc". Those steps were not run for this lesson; see Trigger actions with ambient agents and the module on deploying to Google Cloud.
Test the trigger endpoints offline
test_triggers.py imports main.app, the same app python main.py serves, and calls it with FastAPI’s TestClient. A replay model registered for replay-.* returns the recorded replies, and can raise a chosen error on its first call.
python -m unittest -v test_triggers.pytest_a_429_is_retried_and_then_succeeds (test_triggers.Triggers.test_a_429_is_retried_and_then_succeeds) ... ok
test_a_missing_delivery_id_cannot_post (test_triggers.Triggers.test_a_missing_delivery_id_cannot_post) ... ok
test_a_non_transient_error_returns_500 (test_triggers.Triggers.test_a_non_transient_error_returns_500) ... ok
test_a_page_path_cannot_leave_the_named_docs (test_triggers.Triggers.test_a_page_path_cannot_leave_the_named_docs) ... ok
test_a_pubsub_delivery_runs_the_agent_in_a_new_session (test_triggers.Triggers.test_a_pubsub_delivery_runs_the_agent_in_a_new_session) ... ok
test_a_redelivered_message_does_not_post_twice (test_triggers.Triggers.test_a_redelivered_message_does_not_post_twice) ... ok
test_a_wrong_token_is_rejected_with_401 (test_triggers.Triggers.test_a_wrong_token_is_rejected_with_401) ... ok
test_an_eventarc_binary_mode_event_is_accepted (test_triggers.Triggers.test_an_eventarc_binary_mode_event_is_accepted) ... ok
test_an_oidc_audience_rejects_a_request_without_a_token (test_triggers.Triggers.test_an_oidc_audience_rejects_a_request_without_a_token) ... ok
test_concurrent_deliveries_append_one_review (test_triggers.Triggers.test_concurrent_deliveries_append_one_review) ... ok
test_malformed_base64_is_rejected_with_400 (test_triggers.Triggers.test_malformed_base64_is_rejected_with_400) ... ok
test_the_cli_refuses_triggers_without_auth (test_triggers.Triggers.test_the_cli_refuses_triggers_without_auth) ... ok
----------------------------------------------------------------------
Ran 12 tests in 0.793s
OKAll 12 tests pass.
Read the test and recording scripts
"""Deliver the recorded events to main.py's app offline, with the model replaced by its recording.
Run: python -m unittest -v test_triggers.py
The trigger routes, their auth check, base64 decoding, retry loop and the post_review tool are real.
"""
import json
import logging
import os
import tempfile
import unittest
import warnings
import time
from concurrent.futures import ThreadPoolExecutor
from types import SimpleNamespace
from unittest import mock
from pathlib import Path
warnings.filterwarnings("ignore")
logging.disable(logging.CRITICAL)
os.environ["ADK_DISABLE_LOAD_DOTENV"] = "1" # keep ancestor .env files out of offline tests
os.environ.update(DOCS_MODEL="replay-model", TRIGGER_TOKEN="test-token", ADK_TRIGGER_RETRY_BASE_DELAY="0.01",
DOCS_REVIEWS=str(Path(tempfile.mkdtemp()) / "reviews.jsonl"))
from fastapi.testclient import TestClient # noqa: E402
from google.adk.cli.fast_api import get_fast_api_app # noqa: E402
from google.adk.models.base_llm import BaseLlm # noqa: E402
from google.adk.models.llm_response import LlmResponse # noqa: E402
from google.adk.models.registry import LLMRegistry # noqa: E402
from google.genai import types # noqa: E402
import main # noqa: E402
import publish # noqa: E402
from docs_reviewer import agent as reviewer # noqa: E402
RECORD = json.loads(Path(__file__).with_name("recorded-runs.json").read_text(encoding="utf-8"))
REVIEWS = Path(os.environ["DOCS_REVIEWS"])
class ReplayLlm(BaseLlm):
"""Yields recorded responses in order; FAIL_FIRST raises that many errors first."""
queue: list = []
fail_first: list = []
@classmethod
def supported_models(cls):
return [r"replay-.*"]
async def generate_content_async(self, llm_request, stream=False):
if ReplayLlm.fail_first:
raise ReplayLlm.fail_first.pop(0)
yield LlmResponse(content=types.Content.model_validate(ReplayLlm.queue.pop(0)["content"]))
LLMRegistry.register(ReplayLlm)
# Pin the loaded agent too: unittest discovery can import it before DOCS_MODEL is set.
reviewer.root_agent.model = ReplayLlm(model="replay-model")
# Discovery may also import the package before DOCS_REVIEWS is configured.
# Keep every tool write in the test sandbox, independent of import order.
reviewer.REVIEWS = REVIEWS
def turns(i: int) -> list:
return list(RECORD["deliveries"][i]["session"]["model_turns"])
class Triggers(unittest.TestCase):
def setUp(self):
warnings.simplefilter("ignore")
REVIEWS.unlink(missing_ok=True)
ReplayLlm.fail_first = []
self.http = TestClient(main.app)
def deliver(self, source, page, event_id, raw=None, token="test-token"):
return publish.deliver(self.http, source, page, event_id, raw, token)
def test_concurrent_deliveries_append_one_review(self):
REVIEWS.write_text("", encoding="utf-8")
context = SimpleNamespace(state={"trigger_delivery": {"source": "pubsub", "id": "1001"}})
original = Path.read_text
def slow_read(path, *args, **kwargs):
value = original(path, *args, **kwargs)
if path == REVIEWS:
time.sleep(0.05) # widen the lookup/append race without changing the result
return value
with mock.patch.object(Path, "read_text", slow_read), ThreadPoolExecutor(max_workers=8) as pool:
results = list(pool.map(lambda _: reviewer.post_review("tokens", "Review", context), range(8)))
self.assertEqual(sum(r["posted"] for r in results), 1)
self.assertEqual(len(REVIEWS.read_text().splitlines()), 1)
def test_a_missing_delivery_id_cannot_post(self):
with self.assertRaisesRegex(ValueError, "trigger delivery ID"):
reviewer.post_review("tokens", "Review", SimpleNamespace(state={"trigger_delivery": {"source": "pubsub"}}))
self.assertFalse(REVIEWS.exists())
def test_a_page_path_cannot_leave_the_named_docs(self):
with mock.patch.object(Path, "is_file", return_value=True), mock.patch.object(Path, "read_text", return_value="private"):
self.assertIn("error", reviewer.read_page("../secret"))
def test_a_pubsub_delivery_runs_the_agent_in_a_new_session(self):
ReplayLlm.queue = turns(0)
r = self.deliver("pubsub", "tokens", "1001")
self.assertEqual((r.status_code, r.json()), (200, RECORD["deliveries"][0]["body"]))
self.assertEqual([json.loads(l)["key"] for l in REVIEWS.read_text().splitlines()], ["pubsub:1001"])
self.assertEqual(RECORD["deliveries"][0]["session"]["state"]["trigger_delivery"]["id"], "1001")
def test_a_redelivered_message_does_not_post_twice(self):
ReplayLlm.queue = turns(0) + turns(1)
self.deliver("pubsub", "tokens", "1001")
r = self.deliver("pubsub", "tokens", "1001")
self.assertEqual(r.status_code, 200)
self.assertEqual(len(REVIEWS.read_text().splitlines()), 1)
def test_an_eventarc_binary_mode_event_is_accepted(self):
ReplayLlm.queue = turns(2)
r = self.deliver("eventarc", "rollback", "ev-77")
self.assertEqual(r.status_code, 200)
self.assertEqual(json.loads(REVIEWS.read_text().splitlines()[0])["key"], "eventarc:ev-77")
def test_malformed_base64_is_rejected_with_400(self):
r = self.deliver("pubsub", "tokens", "1002", raw="not base64!")
self.assertEqual((r.status_code, r.json()), (400, RECORD["deliveries"][3]["body"]))
self.assertFalse(REVIEWS.exists())
def test_a_wrong_token_is_rejected_with_401(self):
r = self.deliver("pubsub", "tokens", "1004", token="wrong-token")
self.assertEqual((r.status_code, r.json()), (401, RECORD["deliveries"][4]["body"]))
def test_the_cli_refuses_triggers_without_auth(self):
with self.assertRaisesRegex(ValueError, "trigger_sources requires trigger_oidc_audience"):
get_fast_api_app(agents_dir=main.AGENTS_DIR, web=False, trigger_sources=["pubsub"])
self.assertEqual(RECORD["no_auth"]["exit"], 1)
def test_an_oidc_audience_rejects_a_request_without_a_token(self):
app = get_fast_api_app(agents_dir=main.AGENTS_DIR, web=False, session_service_uri="memory://",
trigger_sources=["pubsub"], trigger_oidc_audience="https://docs-reviewer.example")
r = TestClient(app).post("/apps/docs_reviewer/trigger/pubsub", json=publish.pubsub_body("tokens", "1003"))
self.assertEqual((r.status_code, r.json()), (401, RECORD["oidc_missing_token"]["body"]))
def test_a_429_is_retried_and_then_succeeds(self):
ReplayLlm.queue = turns(0)
ReplayLlm.fail_first = [RuntimeError("429 RESOURCE_EXHAUSTED (injected)")]
r = self.deliver("pubsub", "tokens", "1001")
self.assertEqual(r.status_code, 200)
self.assertEqual(len(REVIEWS.read_text().splitlines()), 1)
def test_a_non_transient_error_returns_500(self):
ReplayLlm.fail_first = [RuntimeError("model unavailable (injected)")]
r = self.deliver("pubsub", "tokens", "1005")
self.assertEqual(r.status_code, 500)
self.assertIn("Agent processing failed", r.json()["detail"])
if __name__ == "__main__":
unittest.main()"""Record docs-changed events delivered to the local trigger endpoints of `adk api_server`.
Run from this folder with GOOGLE_API_KEY set: python record.py
Starts two servers: one with trigger endpoints on port 8767, one that also requires an OIDC token
on port 8768. No Google Cloud service is called: publish.py sends what Pub/Sub and Eventarc would.
Writes run-output.txt and recorded-runs.json; reviews.jsonl is temporary.
"""
import json
import os
import subprocess
import sys
import tempfile
import time
from pathlib import Path
import httpx
import publish
HERE = Path(__file__).resolve().parent
ADK = str(Path(sys.executable).with_name("adk"))
REVIEWS = Path(tempfile.mkdtemp()) / "reviews.jsonl"
ENV = {**os.environ, "GOOGLE_GENAI_USE_ENTERPRISE": "0", "GOOGLE_GENAI_USE_VERTEXAI": "0", "PYTHONWARNINGS": "ignore",
"DOCS_REVIEWS": str(REVIEWS), "TRIGGER_TOKEN": "local-dev-token"}
publish.TOKEN = "local-dev-token"
NO_AUTH = ["api_server", "--trigger_sources", "pubsub,eventarc", "--port", "8767", "."]
OIDC = ["api_server", "--trigger_sources", "pubsub", "--trigger_oidc_audience", "https://docs-reviewer.example",
"--port", "8768", "--session_service_uri", "memory://", "."]
def start(args: list[str], port: int) -> subprocess.Popen:
proc = subprocess.Popen(args, cwd=HERE, env=ENV, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
for _ in range(60):
try:
httpx.get(f"http://127.0.0.1:{port}/list-apps")
return proc
except httpx.ConnectError:
time.sleep(0.5)
raise RuntimeError(f"server on {port} did not start")
def reviews() -> list:
return [json.loads(l) for l in REVIEWS.read_text().splitlines()] if REVIEWS.exists() else []
USERS = {"pubsub": "projects--docs--subscriptions--docs-changed",
"eventarc": "pubsub.googleapis.com--projects--docs--topics--docs-changed"}
def new_session(http: httpx.Client, source: str, seen: set) -> dict | None:
"""The session the trigger endpoint created for the last delivery, with its events."""
listed = http.get(f"/apps/docs_reviewer/users/{USERS[source]}/sessions").json()
fresh = [s["id"] for s in listed if s["id"] not in seen]
if not fresh:
return None
seen.add(fresh[0])
return http.get(f"/apps/docs_reviewer/users/{USERS[source]}/sessions/{fresh[0]}").json()
def model_turns(session: dict | None) -> list:
turns = []
for e in (session or {}).get("events", []):
content = e.get("content") or {}
if content.get("role") == "model":
turns.append({"content": {"role": "model", "parts": [{k: v for k, v in p.items() if k != "thoughtSignature"}
for p in content["parts"]]}})
return turns
def main() -> None:
rec = {"model": "gemini-3.8-flash", "oidc_server": "adk " + " ".join(OIDC), "deliveries": []}
proc = subprocess.run([ADK, *NO_AUTH], cwd=HERE, env=ENV, capture_output=True, text=True, timeout=60)
rec["no_auth"] = {"command": "adk " + " ".join(NO_AUTH), "error": proc.stderr.strip().splitlines()[-1], "exit": proc.returncode}
steps = [("pubsub", "tokens", "1001", None, None), ("pubsub", "tokens", "1001", None, None),
("eventarc", "rollback", "ev-77", None, None), ("pubsub", "tokens", "1002", "not base64!", None),
("pubsub", "tokens", "1004", None, "wrong-token")]
server = start([sys.executable, "main.py"], 8767)
try:
seen: set = set()
with httpx.Client(base_url="http://127.0.0.1:8767", timeout=180) as http:
for source, page, event_id, raw, token in steps:
t0 = time.perf_counter()
r = publish.deliver(http, source, page, event_id, raw, token)
rec["deliveries"].append({"command": (f"TRIGGER_TOKEN={token} " if token else "")
+ f"python publish.py {source} {page} --{'message' if source == 'pubsub' else 'event'}-id {event_id}"
+ (f" --raw-data {raw!r}" if raw else ""),
"status": r.status_code, "body": r.json(), "seconds": round(time.perf_counter() - t0, 1),
"reviews": reviews()})
session = new_session(http, source, seen)
rec["deliveries"][-1]["session"] = {"user_id": session["userId"], "state": session["state"],
"first_message": session["events"][0]["content"]["parts"][0]["text"],
"model_turns": model_turns(session)} if session else None
finally:
server.terminate()
server.wait(timeout=20)
server = start([ADK, *OIDC], 8768)
try:
event = REVIEWS.with_name("event.json")
event.write_text(json.dumps(publish.pubsub_body("tokens", "1003")), encoding="utf-8")
curl = subprocess.run(["curl", "-s", "-w", "\nHTTP %{http_code}", "-X", "POST", "-H", "Content-Type: application/json",
"-d", f"@{event}", "http://127.0.0.1:8768/apps/docs_reviewer/trigger/pubsub"],
capture_output=True, text=True, timeout=60)
body, status = curl.stdout.rsplit("\nHTTP ", 1)
rec["oidc_missing_token"] = {"status": int(status), "body": json.loads(body), "reviews": len(reviews())}
finally:
server.terminate()
server.wait(timeout=20)
out = [f"$ {rec['no_auth']['command']}", rec["no_auth"]["error"], f"exit {rec['no_auth']['exit']}", "",
"$ TRIGGER_TOKEN=local-dev-token python main.py"]
for d in rec["deliveries"]:
out += [f"$ {d['command']}", f"HTTP {d['status']} {json.dumps(d['body'])} ({d['seconds']} s)"]
out += ["$ cat reviews.jsonl", *[json.dumps(r) for r in reviews()], "",
f"$ {rec['oidc_server']}", "$ curl -s -X POST -H 'Content-Type: application/json' -d @event.json http://127.0.0.1:8768/apps/docs_reviewer/trigger/pubsub",
f"HTTP {rec['oidc_missing_token']['status']} {json.dumps(rec['oidc_missing_token']['body'])}"]
(HERE / "run-output.txt").write_text("\n".join(out) + "\n", encoding="utf-8")
(HERE / "recorded-runs.json").write_text(json.dumps(rec, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
print("\n".join(out))
if __name__ == "__main__":
main()- Serve an agent over HTTP and drive it from the CLI: the
/runendpoint, for event sources outside Google Cloud such as a webhook from your Git host.
Check your understanding
Read the explanation
The delivery still starts a new agent session and may make model calls. The key and lock deduplicate the local side effect.
References
Official sources checked October 21, 2026, against google-adk 2.12.0. The auth requirement, retry and concurrency defaults come from the installed google.adk.cli.api_server and google.adk.cli.trigger_routes modules.