adk run is a terminal session for one developer. A docs portal, chat bot or scheduled job can call an agent through an HTTP endpoint. ADK ships that server: adk api_server turns a folder of agents into a FastAPI app with session endpoints, /run and /run_sse. Here you serve the docuctl docs assistant with it, write client.py, a small httpx client that creates a session and runs a turn both ways, and then script the same agent from the CLI with a replay file and JSONL output. The examples below show recorded runs on Gemini gemini-3.8-flash, not a reliability estimate.
Docuctl is the demonstration deployment CLI described by this lab’s bundled manual; the services here use local sample data.
When you need this. Say your team’s internal docs portal gets a “Ask the docs” box. The portal’s backend sends each question to the docs assistant and shows the answer and the page it read. That backend needs a stable HTTP interface with sessions, streaming and clear errors, and your CI needs a way to run fixed questions against the agent without a browser. This lesson gives you both.
Recorded example. The archived “Serve an agent over HTTP and drive it from the CLI” demonstration retains the earlier shipctl scenario; the current lab uses docuctl and does not claim a new model result.
$ adk api_server --port 8765 --session_service_uri sqlite:///sessions.db --allow_origins http://localhost:3000 .
$ python client.py 'How do I undo the last production deploy?'
apps: ['docs_assistant']
docs_assistant final call read_page({"page": "rollback"})
docs_assistant final result read_page
docs_assistant final 'To undo the last production deploy, run:'
answer: To undo the last production deploy, run:
`shipctl rollback --env production`
Without `--to`, `shipctl rollback` moves the environment back to the previous release. Note that database migrations are not rolled back automatically; if the release ran a migration, you must write a new migration to reverse it.These runtime and deployment lessons share a docs-assistant scenario, not one package that gains every feature in sequence. Each download contains the configuration and recorded output for its own experiment. Combining session, trigger, tracing and hosting changes requires a separate integration check.
1. Serve the agent folder
Download the complete lab ZIP, not just client.py. Extract it and keep the terminal in adk-api-server/: the relative docs paths, SQLite file and agent discovery all depend on that project boundary. The folder contains an importable docs_assistant package with one root_agent and a read_page tool.
unzip serve-an-agent-over-http-and-drive-it-from-the-cli.zip
cd adk-api-server
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txt
cp docs_assistant/.env.example docs_assistant/.envEdit docs_assistant/.env and set GOOGLE_API_KEY to your Gemini API key. This selects the Gemini API, not a cloud project. Keep the file private. If your terminal exports another backend or model, unset those values deliberately before using this Gemini setup; ADK does not replace existing environment variables with .env values. For an already installed local model, export DOCS_MODEL=ollama_chat/gemma4:e2b in the server terminal; the agent uses LiteLlm for that value. No model is downloaded by these commands.
"""The docs assistant with one tool, so its event stream has calls, results and an answer."""
import os
from pathlib import Path
from google.adk import Agent
from google.adk.models.lite_llm import LiteLlm
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 = Path(__file__).resolve().parent.parent / "docs"
PAGES = sorted(p.stem for p in DOCS.glob("*.md"))
def read_page(page: str) -> dict:
"""Read one page of the docuctl documentation.
Args:
page: The page name: config, deploy, rollback or tokens.
"""
if page not in PAGES:
return {"error": f"No page {page!r}. Pages: {', '.join(PAGES)}."}
return {"page": page, "text": (DOCS / f"{page}.md").read_text(encoding="utf-8")}
root_agent = Agent(
name="docs_assistant",
model=LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL,
description="Answers developer questions about docuctl from its documentation.",
instruction="""You answer questions from developers about docuctl, the team's deploy CLI.
Read the relevant page with read_page before you answer, and answer only from it.
Copy commands exactly, in backticks. Keep the answer under 100 words.""",
tools=[read_page],
generate_content_config=types.GenerateContentConfig(
temperature=0.2,
max_output_tokens=1024,
thinking_config=types.ThinkingConfig(thinking_level="low"),
),
)Keep this first terminal open. . means the project that contains the docs_assistant/ agent folder:
adk api_server --host 127.0.0.1 --port 8765 \
--session_service_uri sqlite:///sessions.db \
--allow_origins http://localhost:3000 .Starting the server does not ask the model a question. The HTTP and terminal activities below do. Stop the server with Ctrl-C when you finish.
Three flags decide how the server behaves in front of real traffic. --session_service_uri sqlite:///sessions.db keeps sessions in a SQLite file, so a restart does not lose conversations; without it the server uses local .adk storage or memory. --allow_origins lists the browser origins allowed to call it, which matters as soon as a web page calls the server directly. --port picks the port. adk web starts the same server plus the Dev UI; the help text for both says the endpoints are unauthenticated, so put them behind your own gateway or identity proxy before anything outside your machine can reach them.
A browser origin not allowed. A request with Origin: https://docs-portal.example gets HTTP 403 Forbidden: origin not allowed before it reaches a route. Add the portal’s origin to --allow_origins. The preflight for an allowed origin returned access-control-allow-origin: http://localhost:3000.
2. Create a session and run a turn with /run
Open a second terminal, replace /path/to/unpacked with your extraction directory, and activate the same environment. client.py asks one question, creates a session and prints the actual events; --stream uses the streaming endpoint. Each command invokes the selected model.
cd /path/to/unpacked/adk-api-server
source .venv/bin/activate
python client.py "How do I undo the last production deploy?"
python client.py --stream "Which page explains tokens?"Look for a read_page call, its result and the final answer. The trace below is historical output; your model response and timings can differ.
"""Drive the docs assistant through the ADK API server over HTTP.
Start the server from this folder first:
adk api_server --port 8765 --session_service_uri sqlite:///sessions.db .
Then: python client.py "How do I undo the last production deploy?"
"""
import argparse
import json
import os
import uuid
import httpx
BASE_URL = os.environ.get("ADK_URL", "http://127.0.0.1:8765")
APP, USER = "docs_assistant", "dev-1"
def create_session(client: httpx.Client, session_id: str, state: dict | None = None) -> dict:
"""POST /apps/{app}/users/{user}/sessions with the session ID in the body."""
r = client.post(f"/apps/{APP}/users/{USER}/sessions", json={"session_id": session_id, "state": state or {}})
r.raise_for_status()
return r.json()
def message(text: str) -> dict:
return {"role": "user", "parts": [{"text": text}]}
def run(client: httpx.Client, session_id: str, text: str) -> list[dict]:
"""POST /run: the server runs the whole turn and returns every event at once."""
body = {"app_name": APP, "user_id": USER, "session_id": session_id, "new_message": message(text)}
r = client.post("/run", json=body)
r.raise_for_status()
return r.json()
def run_sse(client: httpx.Client, session_id: str, text: str, streaming: bool = True):
"""POST /run_sse: one server-sent event per ADK event, as the runner yields it."""
body = {"app_name": APP, "user_id": USER, "session_id": session_id,
"new_message": message(text), "streaming": streaming}
with client.stream("POST", "/run_sse", json=body) as r:
r.raise_for_status()
for line in r.iter_lines():
if line.startswith("data: "):
yield json.loads(line[len("data: "):])
def describe(event: dict) -> str:
"""One line per event: author, partial or final, and what the parts carry."""
parts = (event.get("content") or {}).get("parts") or []
what = []
for p in parts:
if "functionCall" in p:
what.append(f"call {p['functionCall']['name']}({json.dumps(p['functionCall']['args'])})")
elif "functionResponse" in p:
what.append(f"result {p['functionResponse']['name']}")
elif p.get("text") and not p.get("thought"):
what.append(repr(p["text"][:40]))
if not what:
what.append("(no text)")
kind = "partial" if event.get("partial") else "final"
return f"{event['author']:<15} {kind:<8} {' + '.join(what)}"
def answer(events: list[dict]) -> str:
"""The text of the agent's last complete event."""
for event in reversed(events):
if event["author"] != "user" and not event.get("partial"):
texts = [p["text"] for p in (event.get("content") or {}).get("parts", []) if p.get("text") and not p.get("thought")]
if texts:
return "".join(texts)
return ""
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("question")
parser.add_argument("--stream", action="store_true", help="use /run_sse with streaming")
args = parser.parse_args()
with httpx.Client(base_url=BASE_URL, timeout=120) as client:
print("apps:", client.get("/list-apps").json())
sid = f"s-{uuid.uuid4().hex[:8]}"
create_session(client, sid, {"team": "payments"})
events = list(run_sse(client, sid, args.question)) if args.stream else run(client, sid, args.question)
for event in events:
print(describe(event))
print("answer:", answer(events))
if __name__ == "__main__":
main()create_session posts to /apps/{app}/users/{user}/sessions with the session ID and initial state in the body. The server returns the stored session, and the recorded state is {"team":"payments"}. run posts a RunAgentRequest: app name, user ID, session ID and the new message as a types.Content in JSON. /run blocks until the turn ends and returns every event as one list, the same Event objects a Runner yields in Python, serialized in camelCase. The output above shows 3 events: the model’s read_page call, the tool result and the answer.
An unknown session ID. /run returns HTTP 404 with Session not found: no-such-session; nothing runs. Create the session first, or start the server with --auto_create_session when clients pick their own IDs.
State nested under the deprecated path. Posting {"state": {...}} to the older /sessions/{session_id} path stores {"state":{"team":"payments"}}: that path reads the whole body as the state. ADK 2.12 marks it deprecated. Post session_id and state to /apps/{app}/users/{user}/sessions, as create_session does.
$ curl -X POST /run with session_id no-such-session
HTTP 404 {"detail": "Session not found: no-such-session"}
$ curl -X POST /run -H 'Origin: https://docs-portal.example'
HTTP 403 Forbidden: origin not allowed
$ curl -X POST /apps/docs_assistant/users/dev-1/sessions/s-old -d '{"state": {"team": "payments"}}'
HTTP 200 state: {"state": {"team": "payments"}}The origin check stops a browser page on another site from using the server; it does not authenticate anyone. A client that sends no Origin header, such as client.py, passes it.
3. Stream the turn with /run_sse
A UI should show text as it arrives. /run_sse takes the same body plus "streaming": true and answers with server-sent events: one data: {...} line per event. run_sse reads them with client.stream and iter_lines.
$ python client.py --stream 'How do I undo the last production deploy?'
docs_assistant partial call read_page({"page": "rollback"})
docs_assistant partial (no text)
docs_assistant final call read_page({"page": "rollback"})
docs_assistant final result read_page
docs_assistant partial 'To undo the last production deploy and r'
docs_assistant partial '`\n\nWithout the `--to` flag, `shipctl rol'
docs_assistant partial ' migrations are not rolled back automati'
docs_assistant partial (no text)
docs_assistant final 'To undo the last production deploy and r'
yielded 9 events; the session stored 4The partial events are chunks of model output, and some carry no visible text. The session keeps only complete events, so a client renders partials as they arrive and replaces them with the final event, the same rule you used with Runner.run_async.
4. Drive the agent from the CLI
Two adk run flags make the CLI useful in scripts. --replay FILE takes initial state and a list of queries, creates a fresh session and runs each query in order. It suits a smoke test after a deploy, because the questions live in version control:
{
"state": {"team": "payments"},
"queries": [
"How do I roll back production to release r41?",
"Does that also undo the database migration?"
]
}$ adk run --in_memory --replay replay.json docs_assistant
[user]: How do I roll back production to release r41?
[docs_assistant]: To roll back production to release r41, run:
`shipctl rollback --env production --to r41`
This points traffic to the specified release without rebuilding. Note that database migrations are not rolled back automatically.
[user]: Does that also undo the database migration?
[docs_assistant]: No, database migrations are not rolled back. If the release you are leaving ran a migration, you must write a new migration that reverses it.--jsonl prints one JSON event per line instead of the [author]: text view, so a script can check which tools ran. This is the first record of the recorded run, the function call event, with its signature and text trimmed:
{
"author": "docs_assistant",
"session_id": "6a0dc749-f1c8-41ea-b470-4e353ed8e778",
"node_path": "docs_assistant@1",
"id": "037796ae-5804-40e6-831a-66dfcf060a65",
"modelVersion": "gemini-3.8-flash",
"content": {
"parts": [
{
"functionCall": {
"id": "call_454943",
"args": {
"page": "tokens"
},
"name": "read_page"
}
}
]
},
"finishReason": "STOP",
"usageMetadata": {
"candidatesTokenCount": 16,
"promptTokenCount": 172,
"promptTokensDetails": [
{
"modality": "TEXT",
"tokenCount": 172
}
],
"thoughtsTokenCount": 78,
"totalTokenCount": 266
},
"invocationId": "e-c35d73cf-ce8a-4df7-95dc-9e783b8a6d14",
"nodeInfo": {
"path": "docs_assistant@1"
},
"longRunningToolIds": [],
"timestamp": 1792628369.447583
}--in_memory keeps these runs out of the local session store. --save_session with --session_id, and --resume FILE, save a session to JSON and reopen it later; those are for a developer at the terminal, not for scripts.
Stop the API server before reusing its port. From the first terminal you can inspect the same package with ADK Web instead:
adk web --host 127.0.0.1 --port 8765 \
--session_service_uri sqlite:///sessions.db .Open http://127.0.0.1:8765, select docs_assistant, start a session and send one question. Inspect its events and state after the response. This is a developer interface, not authentication for a deployed service.
How the server loads agents and streams events
adk api_server calls get_fast_api_app(agents_dir=..., web=False, ...) and serves the FastAPI app with uvicorn. Each folder in agents_dir with a root_agent is an app; GET /list-apps returned ["docs_assistant"]. On the first request for an app, the server loads the agent and builds one Runner for it with the configured session service. Every request then names its session, so many clients share one runner while each conversation keeps its own history and state in sessions.db.
/run and /run_sse differ only in delivery. /run collects the events and returns them when the turn ends. /run_sse checks the session first, so a missing session still gets a clean 404 instead of a broken stream, then writes each event as the runner yields it. In the recorded stream the first event arrived at 1.98 s and the last at 4.23 s, on one machine: most of the wait is the two model calls, and streaming changes when the client sees text, not how long the turn takes.
Test the server without a model
test_server.py builds the same app with get_fast_api_app and calls it through FastAPI’s TestClient. The model is a BaseLlm subclass registered with LLMRegistry for names matching replay-.*; setting DOCS_MODEL=replay-model makes agent.py resolve to it, so the server, runner, session service and tool all run for real and only the model replies come from the recording.
python -m unittest -v test_server.py test_recording_opt_in.pytest_a_browser_origin_not_in_allow_origins_gets_403 (test_server.ApiServer.test_a_browser_origin_not_in_allow_origins_gets_403) ... ok
test_an_unknown_session_returns_404 (test_server.ApiServer.test_an_unknown_session_returns_404) ... ok
test_answer_skips_contentless_end_or_error_events (test_server.ApiServer.test_answer_skips_contentless_end_or_error_events) ... ok
test_auto_create_session_accepts_a_new_id (test_server.ApiServer.test_auto_create_session_accepts_a_new_id) ... ok
test_run_returns_the_whole_turn_as_one_list (test_server.ApiServer.test_run_returns_the_whole_turn_as_one_list) ... ok
test_run_sse_streams_partials_the_session_does_not_store (test_server.ApiServer.test_run_sse_streams_partials_the_session_does_not_store) ... ok
test_the_deprecated_per_id_path_nests_the_state (test_server.ApiServer.test_the_deprecated_per_id_path_nests_the_state) ... ok
----------------------------------------------------------------------
Ran 7 tests in 0.318s
OKThe saved output reports 7 server tests, and test_recording_opt_in.py checks that the recording script refuses to run without --record.
Read the tests and optional recording script
python record.py --record makes new model calls and overwrites the saved run files. It is an optional authoring operation, not the setup or the ordinary one-question activity. Keep the original records if you want to compare runs.
"""Replay the recorded model output through the real ADK API server app, offline.
Run: python -m unittest -v test_server.py
get_fast_api_app builds the same FastAPI app that `adk api_server` serves; only the model is replaced.
"""
import json
import logging
import os
import unittest
import warnings
from pathlib import Path
from unittest import mock
warnings.filterwarnings("ignore")
logging.disable(logging.CRITICAL)
os.environ["ADK_DISABLE_LOAD_DOTENV"] = "1" # keep ancestor .env files out of offline tests
os.environ["DOCS_MODEL"] = "replay-model" # agent.py passes this string to ADK's model registry
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 client as c # noqa: E402
HERE = Path(__file__).resolve().parent
RECORD = json.loads((HERE / "recorded-runs.json").read_text(encoding="utf-8"))
class ReplayLlm(BaseLlm):
"""Yields the recorded responses of one model call per request, partial chunks included."""
queue: list = []
@classmethod
def supported_models(cls):
return [r"replay-.*"]
async def generate_content_async(self, llm_request, stream=False):
while ReplayLlm.queue:
turn = ReplayLlm.queue.pop(0)
yield LlmResponse(content=types.Content.model_validate(turn["content"]), partial=turn["partial"] or None)
if not turn["partial"]:
return
LLMRegistry.register(ReplayLlm)
# unittest discovery may import agent packages before this test module sets DOCS_MODEL.
from docs_assistant import agent as loaded_agent
loaded_agent.root_agent.model = ReplayLlm(model="replay-model")
def server(**kwargs) -> TestClient:
app = get_fast_api_app(agents_dir=str(HERE), web=False, session_service_uri="memory://",
artifact_service_uri="memory://", allow_origins=["http://localhost:3000"], **kwargs)
return TestClient(app)
class ApiServer(unittest.TestCase):
def setUp(self):
warnings.simplefilter("ignore") # unittest re-enables warnings per test
self.http = server()
def test_answer_skips_contentless_end_or_error_events(self):
events = [{"author": "docs_assistant", "content": {"parts": [{"text": "Answer"}]}},
{"author": "docs_assistant", "content": None}]
self.assertEqual(c.answer(events), "Answer")
def test_run_returns_the_whole_turn_as_one_list(self):
ReplayLlm.queue = list(RECORD["run"]["model_turns"])
self.assertEqual(self.http.get("/list-apps").json(), RECORD["apps"])
c.create_session(self.http, "s-run", {"team": "payments"})
events = c.run(self.http, "s-run", RECORD["question"])
self.assertEqual([c.describe(e) for e in events], RECORD["run"]["lines"])
self.assertEqual(c.answer(events), RECORD["run"]["answer"])
stored = self.http.get(f"/apps/{c.APP}/users/{c.USER}/sessions/s-run").json()
self.assertEqual(stored["state"], {"team": "payments"})
def test_run_sse_streams_partials_the_session_does_not_store(self):
ReplayLlm.queue = list(RECORD["sse"]["model_turns"])
c.create_session(self.http, "s-sse")
events = list(c.run_sse(self.http, "s-sse", RECORD["question"]))
self.assertEqual([c.describe(e) for e in events], RECORD["sse"]["lines"])
stored = self.http.get(f"/apps/{c.APP}/users/{c.USER}/sessions/s-sse").json()
self.assertEqual(len(stored["events"]), RECORD["sse"]["stored"])
self.assertFalse(any(e.get("partial") for e in stored["events"]))
def test_an_unknown_session_returns_404(self):
body = {"app_name": c.APP, "user_id": c.USER, "session_id": "no-such-session", "new_message": c.message("hi")}
r = self.http.post("/run", json=body)
self.assertEqual((r.status_code, r.json()), (404, RECORD["failures"]["unknown_session"]["body"]))
def test_auto_create_session_accepts_a_new_id(self):
ReplayLlm.queue = list(RECORD["run"]["model_turns"])
http = server(auto_create_session=True)
events = c.run(http, "brand-new", RECORD["question"])
self.assertEqual(c.answer(events), RECORD["run"]["answer"])
def test_a_browser_origin_not_in_allow_origins_gets_403(self):
body = {"app_name": c.APP, "user_id": c.USER, "session_id": "s-run", "new_message": c.message("hi")}
r = self.http.post("/run", json=body, headers={"Origin": "https://docs-portal.example"})
self.assertEqual((r.status_code, r.text), (403, RECORD["failures"]["origin"]["body"]))
r = self.http.options("/run", headers={"Origin": "http://localhost:3000", "Access-Control-Request-Method": "POST"})
self.assertEqual(r.headers.get("access-control-allow-origin"), "http://localhost:3000")
def test_the_deprecated_per_id_path_nests_the_state(self):
r = self.http.post(f"/apps/{c.APP}/users/{c.USER}/sessions/s-old", json={"state": {"team": "payments"}})
self.assertEqual(r.json()["state"], RECORD["failures"]["deprecated_path"]["state"])
self.assertEqual(c.create_session(self.http, "s-new", {"team": "payments"})["state"], {"team": "payments"})
if __name__ == "__main__":
unittest.main()"""Record the docs assistant served by `adk api_server`, driven over HTTP and from `adk run`.
Run from this folder with GOOGLE_API_KEY set: python record.py --record
Starts the server on port 8765, writes run-output.txt and recorded-runs.json, then stops it.
"""
import json
import os
import subprocess
import sys
import time
from pathlib import Path
import httpx
import client as c
HERE = Path(__file__).resolve().parent
ADK = str(Path(sys.executable).with_name("adk"))
QUESTION = "How do I undo the last production deploy?"
ENV = {**os.environ, "GOOGLE_GENAI_USE_ENTERPRISE": "0", "GOOGLE_GENAI_USE_VERTEXAI": "0", "PYTHONWARNINGS": "ignore"}
SERVER = [ADK, "api_server", "--port", "8765", "--session_service_uri", "sqlite:///sessions.db",
"--allow_origins", "http://localhost:3000", "."]
def model_turns(events):
"""The model's own responses, which the offline tests replay; thought signatures dropped."""
turns = []
for e in events:
content = e.get("content") or {}
if content.get("role") == "model":
parts = [{k: v for k, v in p.items() if k != "thoughtSignature"} for p in content["parts"]]
turns.append({"content": {"role": "model", "parts": parts}, "partial": bool(e.get("partial"))})
return turns
def short(event):
keep = {k: event[k] for k in ("author", "invocationId", "partial", "content", "finishReason", "usageMetadata") if k in event}
return json.loads(json.dumps(keep).replace(json.dumps(event.get("content", {}).get("parts", [{}])[0].get("thoughtSignature", "")), '"<signature>"'))
def main() -> None:
(HERE / "sessions.db").unlink(missing_ok=True)
out, rec = [], {"model": "gemini-3.8-flash", "question": QUESTION, "server": " ".join(["adk", *SERVER[1:]])}
server = subprocess.Popen(SERVER, cwd=HERE, env=ENV, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
try:
with httpx.Client(base_url="http://127.0.0.1:8765", timeout=120) as http:
for _ in range(60):
try:
rec["apps"] = http.get("/list-apps").json()
break
except httpx.ConnectError:
time.sleep(0.5)
session = c.create_session(http, "s-run", {"team": "payments"})
rec["created"] = {k: session[k] for k in ("id", "appName", "userId", "state", "events")}
out += [f"$ {rec['server']}", f"$ python client.py {QUESTION!r}", f"apps: {rec['apps']}"]
events = c.run(http, "s-run", QUESTION)
rec["run"] = {"lines": [c.describe(e) for e in events], "answer": c.answer(events),
"events": [short(e) for e in events], "model_turns": model_turns(events)}
out += rec["run"]["lines"] + [f"answer: {rec['run']['answer']}", ""]
c.create_session(http, "s-sse")
t0, seconds, streamed = time.perf_counter(), [], []
for e in c.run_sse(http, "s-sse", QUESTION):
streamed.append(e)
seconds.append(round(time.perf_counter() - t0, 2))
stored = http.get(f"/apps/{c.APP}/users/{c.USER}/sessions/s-sse").json()
rec["sse"] = {"lines": [c.describe(e) for e in streamed], "seconds": seconds, "answer": c.answer(streamed),
"stored": len(stored["events"]), "model_turns": model_turns(streamed)}
out += [f"$ python client.py --stream {QUESTION!r}"] + rec["sse"]["lines"]
out += [f"yielded {len(streamed)} events; the session stored {rec['sse']['stored']}", ""]
fail = {}
body = {"app_name": c.APP, "user_id": c.USER, "session_id": "no-such-session", "new_message": c.message("hi")}
r = http.post("/run", json=body)
fail["unknown_session"] = {"status": r.status_code, "body": r.json()}
body["session_id"] = "s-run"
r = http.post("/run", json=body, headers={"Origin": "https://docs-portal.example"})
fail["origin"] = {"status": r.status_code, "body": r.text}
r = http.options("/run", headers={"Origin": "http://localhost:3000", "Access-Control-Request-Method": "POST"})
fail["allowed_origin"] = {"status": r.status_code, "allow_origin": r.headers.get("access-control-allow-origin")}
r = http.post(f"/apps/{c.APP}/users/{c.USER}/sessions/s-old", json={"state": {"team": "payments"}})
fail["deprecated_path"] = {"status": r.status_code, "state": r.json()["state"]}
rec["failures"] = fail
out += ["$ curl -X POST /run with session_id no-such-session", f"HTTP {fail['unknown_session']['status']} {json.dumps(fail['unknown_session']['body'])}",
"$ curl -X POST /run -H 'Origin: https://docs-portal.example'", f"HTTP {fail['origin']['status']} {fail['origin']['body']}",
"$ curl -X POST /apps/docs_assistant/users/dev-1/sessions/s-old -d '{\"state\": {\"team\": \"payments\"}}'",
f"HTTP {fail['deprecated_path']['status']} state: {json.dumps(fail['deprecated_path']['state'])}", ""]
finally:
server.terminate()
server.wait(timeout=20)
cli = [ADK, "run", "--in_memory", "--replay", "replay.json", "docs_assistant"]
proc = subprocess.run(cli, cwd=HERE, env=ENV, capture_output=True, text=True, timeout=300)
lines = [l for l in proc.stdout.splitlines() if l.strip() and not l.startswith(("Log setup", "To access latest log"))]
rec["cli_replay"] = {"command": "adk run --in_memory --replay replay.json docs_assistant", "stdout": lines, "exit": proc.returncode}
out += [f"$ {rec['cli_replay']['command']}"] + lines + [""]
cli = [ADK, "run", "--in_memory", "--jsonl", "docs_assistant", "Which page explains tokens?"]
proc = subprocess.run(cli, cwd=HERE, env=ENV, capture_output=True, text=True, timeout=300)
jsonl = [json.loads(l) for l in proc.stdout.splitlines() if l.startswith("{")]
rec["cli_jsonl"] = {"command": "adk run --in_memory --jsonl docs_assistant 'Which page explains tokens?'", "records": jsonl}
out += [f"$ {rec['cli_jsonl']['command']}"] + [json.dumps(j)[:150] + " …" for j in jsonl]
(HERE / "sessions.db").unlink(missing_ok=True)
(HERE / "run-output.txt").write_text("\n".join(out).rstrip() + "\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__":
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")
main()- Build an agent as YAML config and in Visual Builder: define the same assistant without Python and see what the builder endpoints accept.
Check your understanding
Read the explanation
The lesson distinguishes origin checks from caller authentication; a client without Origin passes this check.
References
Checked October 22, 2026, against adk.dev and google-adk 2.12.0. The session and run request shapes come from the google.adk.cli.api_server module.
Previously on the blog: Google ADK Setup: adk web, CLI & Python Runner Guide covers the ADK 1.x version of these commands.