Python 88.3%
TypeScript 7.6%
Shell 4.1%
1#!/usr/bin/env python32"""Live probes for Gemini tools / File Search / Interactions / Environments / auth_tokens (API Atlas, 2026-09-18).3Every call goes through scripts.live.gemini_request (logged to reports/live-requests.jsonl, key never printed).4Raw sanitized responses -> tmp-live/gemini-tools/<step>.json ; summary -> _summary.json5Run: .venv/bin/python tmp-live/gemini-tools/run_live.py [step ...]6"""7from __future__ import annotations89import base6410import json11import sys12import time13import traceback14from datetime import datetime, timedelta, timezone15from pathlib import Path1617ROOT = Path(__file__).resolve().parents[2]18sys.path.insert(0, str(ROOT))19from scripts.live import GEMINI_BASE, gemini_request, save_sanitized, log_request # noqa: E4022021OUT = ROOT / "tmp-live" / "gemini-tools"22LITE = "gemini-3.5-flash-lite"23FLASH = "gemini-3.5-flash"24PRICE = {LITE: (0.30, 2.50), FLASH: (0.50, 3.00), "gemini-3.8-flash": (0.50, 3.00)} # USD per 1M (in, out); flash = conservative guess25SUMMARY: dict[str, dict] = {}26PNG_1x1 = base64.b64encode(bytes.fromhex(27 "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4890000000d4944415478da63f8ffff3f0300050001"28 "0d0a2db40000000049454e44ae426082")).decode()293031def cost(model: str, usage: dict | None) -> float:32 if not usage:33 return 0.034 pi, po = PRICE.get(model, (0.50, 3.00))35 return (usage.get("promptTokenCount", 0) * pi + (usage.get("candidatesTokenCount", 0) + usage.get("thoughtsTokenCount", 0)) * po) / 1e6363738def gen(model: str, body: dict, step: str, note: str = "") -> tuple[int, dict]:39 st, r, h = gemini_request("POST", f"/v1beta/models/{model}:generateContent", body, note=f"gemini-tools:{step} {note}")40 if isinstance(r, dict):41 c = cost(model, r.get("usageMetadata"))42 if c:43 log_request("gemini", "COST", f"models/{model}:generateContent", st, c, f"gemini-tools:{step} usage est")44 save_sanitized({"request": body, "status": st, "response": r, "headers": {k: v for k, v in h.items() if k.lower().startswith(("x-goog", "server-timing"))}}, OUT / f"{step}.json")45 return st, r464748def steps(r) -> list[dict]:49 return (r.get("steps") or r.get("outputs") or []) if isinstance(r, dict) else []505152def parts(r: dict) -> list[dict]:53 try:54 return r["candidates"][0]["content"]["parts"]55 except Exception: # noqa: BLE00156 return []575859def part_kinds(ps: list[dict]) -> list[str]:60 return [",".join(k for k in p.keys()) for p in ps]616263WEATHER = {"name": "get_weather", "description": "Get the current weather for a city.",64 "parameters": {"type": "OBJECT", "properties": {"city": {"type": "STRING", "description": "City name"}}, "required": ["city"]}}65TIME = {"name": "get_time", "description": "Get the current local time in a city.",66 "parameters": {"type": "OBJECT", "properties": {"city": {"type": "STRING"}}, "required": ["city"]}}676869def step_a_function_calling() -> None:70 body = {"contents": [{"role": "user", "parts": [{"text": "Weather in Paris?"}]}],71 "tools": [{"functionDeclarations": [WEATHER]}],72 "toolConfig": {"functionCallingConfig": {"mode": "ANY", "allowedFunctionNames": ["get_weather"]}},73 "generationConfig": {"maxOutputTokens": 64}}74 st, r = gen(LITE, body, "a1_fc_forced", "mode ANY forced")75 ps = parts(r)76 fc = next((p for p in ps if "functionCall" in p), None)77 SUMMARY["a1_fc_forced"] = {"status": st, "part_kinds": part_kinds(ps), "functionCall": fc and fc["functionCall"],78 "has_thoughtSignature": bool(fc and fc.get("thoughtSignature")), "finishReason": r.get("candidates", [{}])[0].get("finishReason"),79 "usage": r.get("usageMetadata")}80 if st == 200 and fc:81 model_content = r["candidates"][0]["content"]82 body2 = {"contents": [body["contents"][0], model_content,83 {"role": "user", "parts": [{"functionResponse": {"name": "get_weather", "response": {"temperature_c": 18, "conditions": "light rain"}}}]}],84 "tools": [{"functionDeclarations": [WEATHER]}], "generationConfig": {"maxOutputTokens": 32}}85 st2, r2 = gen(LITE, body2, "a2_fc_response", "functionResponse round trip")86 SUMMARY["a2_fc_response"] = {"status": st2, "part_kinds": part_kinds(parts(r2)), "text": "".join(p.get("text", "") for p in parts(r2))[:200],87 "usage": r2.get("usageMetadata")}88 # a2b: same round trip WITHOUT echoing thoughtSignature -> does the API reject it?89 stripped = {"role": "model", "parts": [{k: v for k, v in p.items() if k != "thoughtSignature"} for p in model_content["parts"]]}90 body3 = dict(body2, contents=[body["contents"][0], stripped, body2["contents"][2]])91 st3, r3 = gen(LITE, body3, "a2b_fc_no_thought_signature", "functionResponse without thoughtSignature")92 SUMMARY["a2b_fc_no_thought_signature"] = {"status": st3, "error": (r3.get("error") or {}).get("message", "")[:300] if isinstance(r3, dict) else str(r3)[:200],93 "part_kinds": part_kinds(parts(r3))}94 # a3 parallel95 body = {"contents": [{"role": "user", "parts": [{"text": "What is the weather in Paris and the time in Tokyo? Use both tools."}]}],96 "tools": [{"functionDeclarations": [WEATHER, TIME]}],97 "toolConfig": {"functionCallingConfig": {"mode": "ANY"}}, "generationConfig": {"maxOutputTokens": 128}}98 st, r = gen(LITE, body, "a3_fc_parallel", "two functions, mode ANY")99 ps = parts(r)100 SUMMARY["a3_fc_parallel"] = {"status": st, "part_kinds": part_kinds(ps), "calls": [p["functionCall"] for p in ps if "functionCall" in p],101 "thoughtSignature_on_parts": [bool(p.get("thoughtSignature")) for p in ps], "usage": r.get("usageMetadata")}102 # a4 parametersJsonSchema103 decl = {"name": "get_weather", "description": "Get the current weather for a city.",104 "parametersJsonSchema": {"type": "object", "properties": {"city": {"type": "string"}, "unit": {"type": "string", "enum": ["C", "F"]}}, "required": ["city"]}}105 body = {"contents": [{"role": "user", "parts": [{"text": "Weather in Rome in Celsius?"}]}], "tools": [{"functionDeclarations": [decl]}],106 "toolConfig": {"functionCallingConfig": {"mode": "ANY"}}, "generationConfig": {"maxOutputTokens": 64}}107 st, r = gen(LITE, body, "a4_fc_parametersJsonSchema", "parametersJsonSchema variant")108 ps = parts(r)109 SUMMARY["a4_fc_parametersJsonSchema"] = {"status": st, "part_kinds": part_kinds(ps), "calls": [p["functionCall"] for p in ps if "functionCall" in p], "usage": r.get("usageMetadata")}110 # a5 mode NONE and a6 VALIDATED probe (cheap)111 body = {"contents": [{"role": "user", "parts": [{"text": "Weather in Paris? Reply in 3 words."}]}], "tools": [{"functionDeclarations": [WEATHER]}],112 "toolConfig": {"functionCallingConfig": {"mode": "NONE"}}, "generationConfig": {"maxOutputTokens": 32}}113 st, r = gen(LITE, body, "a5_fc_mode_none", "mode NONE")114 SUMMARY["a5_fc_mode_none"] = {"status": st, "part_kinds": part_kinds(parts(r)), "usage": r.get("usageMetadata")}115 body["toolConfig"] = {"functionCallingConfig": {"mode": "VALIDATED"}}116 st, r = gen(LITE, body, "a6_fc_mode_validated", "mode VALIDATED probe")117 SUMMARY["a6_fc_mode_validated"] = {"status": st, "part_kinds": part_kinds(parts(r)), "error": (r.get("error") or {}).get("message", "")[:300] if isinstance(r, dict) else None}118 # a7 behavior NON_BLOCKING in generateContent (documented Live-only) -> record error or acceptance119 decl_nb = dict(WEATHER, behavior="NON_BLOCKING")120 body = {"contents": [{"role": "user", "parts": [{"text": "Weather in Paris?"}]}], "tools": [{"functionDeclarations": [decl_nb]}],121 "toolConfig": {"functionCallingConfig": {"mode": "ANY"}}, "generationConfig": {"maxOutputTokens": 32}}122 st, r = gen(LITE, body, "a7_fc_behavior_non_blocking", "behavior NON_BLOCKING in generateContent")123 SUMMARY["a7_fc_behavior_non_blocking"] = {"status": st, "part_kinds": part_kinds(parts(r)), "error": (r.get("error") or {}).get("message", "")[:300] if isinstance(r, dict) else None}124125126def step_b_google_search() -> None:127 body = {"contents": [{"parts": [{"text": "What is today's date in Toronto? 5 words."}]}], "tools": [{"googleSearch": {}}], "generationConfig": {"maxOutputTokens": 48}}128 st, r = gen(LITE, body, "b_google_search", "googleSearch grounding")129 gm = (r.get("candidates") or [{}])[0].get("groundingMetadata") if isinstance(r, dict) else None130 SUMMARY["b_google_search"] = {"status": st, "part_kinds": part_kinds(parts(r)), "groundingMetadata_keys": sorted(gm.keys()) if gm else None,131 "chunk_keys": sorted({k for c in (gm or {}).get("groundingChunks", []) for k in c}) if gm else None,132 "web_chunk_keys": sorted({k for c in (gm or {}).get("groundingChunks", []) for k in c.get("web", {})}) if gm else None,133 "support_keys": sorted({k for s in (gm or {}).get("groundingSupports", []) for k in s}) if gm else None,134 "usage": r.get("usageMetadata"), "error": (r.get("error") or {}).get("message") if isinstance(r, dict) else None}135136137def step_c_url_context() -> None:138 body = {"contents": [{"parts": [{"text": "Summarize https://example.com in 5 words."}]}], "tools": [{"urlContext": {}}], "generationConfig": {"maxOutputTokens": 48}}139 st, r = gen(LITE, body, "c_url_context", "urlContext")140 cand = (r.get("candidates") or [{}])[0] if isinstance(r, dict) else {}141 SUMMARY["c_url_context"] = {"status": st, "part_kinds": part_kinds(parts(r)), "candidate_keys": sorted(cand.keys()), "urlContextMetadata": cand.get("urlContextMetadata"),142 "usage_keys": sorted((r.get("usageMetadata") or {}).keys()), "usage": r.get("usageMetadata"),143 "error": (r.get("error") or {}).get("message") if isinstance(r, dict) else None}144145146def step_d_code_execution() -> None:147 body = {"contents": [{"parts": [{"text": "Run print(2+2) in Python and report the output."}]}], "tools": [{"codeExecution": {}}], "generationConfig": {"maxOutputTokens": 128}}148 st, r = gen(LITE, body, "d_code_execution", "codeExecution")149 ps = parts(r)150 SUMMARY["d_code_execution"] = {"status": st, "part_kinds": part_kinds(ps),151 "executableCode": next((p["executableCode"] for p in ps if "executableCode" in p), None),152 "codeExecutionResult": next((p["codeExecutionResult"] for p in ps if "codeExecutionResult" in p), None),153 "usage": r.get("usageMetadata"), "error": (r.get("error") or {}).get("message") if isinstance(r, dict) else None}154155156def step_e_google_maps() -> None:157 body = {"contents": [{"parts": [{"text": "Name one cafe near me. 5 words."}]}], "tools": [{"googleMaps": {}}],158 "toolConfig": {"retrievalConfig": {"latLng": {"latitude": 43.6532, "longitude": -79.3832}}}, "generationConfig": {"maxOutputTokens": 48}}159 st, r = gen(LITE, body, "e_google_maps", "googleMaps grounding")160 gm = (r.get("candidates") or [{}])[0].get("groundingMetadata") if isinstance(r, dict) else None161 SUMMARY["e_google_maps"] = {"status": st, "part_kinds": part_kinds(parts(r)), "groundingMetadata_keys": sorted(gm.keys()) if gm else None,162 "chunk_keys": sorted({k for c in (gm or {}).get("groundingChunks", []) for k in c}) if gm else None,163 "maps_chunk_keys": sorted({k for c in (gm or {}).get("groundingChunks", []) for k in c.get("maps", {})}) if gm else None,164 "usage": r.get("usageMetadata"), "error": (r.get("error") or {}).get("message") if isinstance(r, dict) else None}165 # e2: enableWidget variant166 body["tools"] = [{"googleMaps": {"enableWidget": True}}]167 st, r = gen(LITE, body, "e2_google_maps_widget", "googleMaps enableWidget")168 gm = (r.get("candidates") or [{}])[0].get("groundingMetadata") if isinstance(r, dict) else None169 SUMMARY["e2_google_maps_widget"] = {"status": st, "groundingMetadata_keys": sorted(gm.keys()) if gm else None, "has_widget_token": bool(gm and gm.get("googleMapsWidgetContextToken")),170 "error": (r.get("error") or {}).get("message") if isinstance(r, dict) else None}171172173def step_f_computer_use() -> None:174 model = "gemini-2.5-computer-use-preview-10-2025"175 body = {"contents": [{"role": "user", "parts": [{"text": "Do nothing. If you must act, call wait_5_seconds."}, {"inlineData": {"mimeType": "image/png", "data": PNG_1x1}}]}],176 "tools": [{"computerUse": {"environment": "ENVIRONMENT_BROWSER"}}], "generationConfig": {"maxOutputTokens": 64}}177 st, r = gen(model, body, "f_computer_use_generateContent", "legacy computerUse tool")178 ps = parts(r)179 SUMMARY["f_computer_use_generateContent"] = {"status": st, "model": model, "part_kinds": part_kinds(ps), "calls": [p["functionCall"] for p in ps if "functionCall" in p],180 "usage": r.get("usageMetadata"), "error": (r.get("error") or {}).get("message", "")[:400] if isinstance(r, dict) else None}181 # f2: Interactions API computer_use tool (current surface) on a cheap model182 body = {"model": LITE, "input": "Do nothing. Reply DONE.", "tools": [{"type": "computer_use", "environment": "browser"}], "store": False}183 st, r, _ = gemini_request("POST", "/v1beta/interactions", body, note="gemini-tools:f2 interactions computer_use probe")184 save_sanitized({"request": body, "status": st, "response": r}, OUT / "f2_computer_use_interactions.json")185 SUMMARY["f2_computer_use_interactions"] = {"status": st, "keys": sorted(r.keys()) if isinstance(r, dict) else None, "outputs_types": [o.get("type") for o in steps(r)] if isinstance(r, dict) else None,186 "error": (r.get("error") or {}) if isinstance(r, dict) and r.get("error") else None}187188189def poll_operation(name: str, step: str, tries: int = 20) -> dict:190 last: dict = {}191 for _ in range(tries):192 st, last, _ = gemini_request("GET", f"/v1beta/{name}", note=f"gemini-tools:{step} poll")193 if isinstance(last, dict) and last.get("done"):194 break195 time.sleep(2)196 return last197198199def step_g_file_search() -> None:200 st, store, _ = gemini_request("POST", "/v1beta/fileSearchStores", {"displayName": "atlas-probe-2026-09-18"}, note="gemini-tools:g1 create store")201 save_sanitized({"status": st, "response": store}, OUT / "g1_store_create.json")202 SUMMARY["g1_store_create"] = {"status": st, "keys": sorted(store.keys()) if isinstance(store, dict) else None, "error": store.get("error") if isinstance(store, dict) else None}203 if st != 200:204 return205 name = store["name"]206 text = ("The Atlas probe document. The secret fruit is kumquat. The secret number is 4711. " * 3)[:200].encode()207 # resumable upload: start208 st, r, h = gemini_request("POST", f"/upload/v1beta/{name}:uploadToFileSearchStore", {"displayName": "probe.txt", "customMetadata": [{"key": "kind", "stringValue": "probe"}]},209 extra_headers={"X-Goog-Upload-Protocol": "resumable", "X-Goog-Upload-Command": "start",210 "X-Goog-Upload-Header-Content-Length": str(len(text)), "X-Goog-Upload-Header-Content-Type": "text/plain"},211 note="gemini-tools:g2 upload start")212 upload_url = next((v for k, v in h.items() if k.lower() == "x-goog-upload-url"), None)213 SUMMARY["g2_upload_start"] = {"status": st, "has_upload_url": bool(upload_url), "upload_headers": sorted(k for k in h if k.lower().startswith("x-goog-upload"))}214 op: dict = {}215 if upload_url:216 path = upload_url[len(GEMINI_BASE):] if upload_url.startswith(GEMINI_BASE) else upload_url217 st, op, _ = gemini_request("POST", path, data=text, content_type="text/plain",218 extra_headers={"X-Goog-Upload-Offset": "0", "X-Goog-Upload-Command": "upload, finalize"},219 note="gemini-tools:g3 upload finalize")220 save_sanitized({"status": st, "response": op}, OUT / "g3_upload_finalize.json")221 SUMMARY["g3_upload_finalize"] = {"status": st, "operation_keys": sorted(op.keys()) if isinstance(op, dict) else None, "done": op.get("done") if isinstance(op, dict) else None}222 if isinstance(op, dict) and op.get("name") and not op.get("done"):223 op = poll_operation(op["name"], "g4")224 save_sanitized({"operation": op}, OUT / "g4_operation_done.json")225 SUMMARY["g4_operation_done"] = {"done": op.get("done"), "response_keys": sorted((op.get("response") or {}).keys()), "metadata": op.get("metadata"), "error": op.get("error")}226 st, docs, _ = gemini_request("GET", f"/v1beta/{name}/documents", note="gemini-tools:g5 list documents")227 save_sanitized({"status": st, "response": docs}, OUT / "g5_documents_list.json")228 doc_list = docs.get("documents", []) if isinstance(docs, dict) else []229 SUMMARY["g5_documents_list"] = {"status": st, "count": len(doc_list), "doc_keys": sorted(doc_list[0].keys()) if doc_list else None, "state": doc_list[0].get("state") if doc_list else None}230 if doc_list:231 st, d, _ = gemini_request("GET", f"/v1beta/{doc_list[0]['name']}", note="gemini-tools:g5b get document")232 SUMMARY["g5b_document_get"] = {"status": st, "keys": sorted(d.keys()) if isinstance(d, dict) else None}233 st, r = gen(LITE, {"contents": [{"parts": [{"text": "What is the secret fruit and the secret number? 6 words."}]}],234 "tools": [{"fileSearch": {"fileSearchStoreNames": [name]}}], "generationConfig": {"maxOutputTokens": 48}}, "g6_file_search_query", "fileSearch tool")235 gm = (r.get("candidates") or [{}])[0].get("groundingMetadata") if isinstance(r, dict) else None236 SUMMARY["g6_file_search_query"] = {"status": st, "text": "".join(p.get("text", "") for p in parts(r))[:200], "groundingMetadata_keys": sorted(gm.keys()) if gm else None,237 "chunk_keys": sorted({k for c in (gm or {}).get("groundingChunks", []) for k in c}) if gm else None,238 "retrievedContext_keys": sorted({k for c in (gm or {}).get("groundingChunks", []) for k in c.get("retrievedContext", {})}) if gm else None,239 "usage": r.get("usageMetadata"), "error": (r.get("error") or {}).get("message") if isinstance(r, dict) else None}240 # g6b: metadataFilter variant241 st, r = gen(LITE, {"contents": [{"parts": [{"text": "What is the secret number? 3 words."}]}],242 "tools": [{"fileSearch": {"fileSearchStoreNames": [name], "metadataFilter": "kind=probe"}}], "generationConfig": {"maxOutputTokens": 32}}, "g6b_file_search_filter", "fileSearch metadataFilter")243 SUMMARY["g6b_file_search_filter"] = {"status": st, "text": "".join(p.get("text", "") for p in parts(r))[:120], "error": (r.get("error") or {}).get("message") if isinstance(r, dict) else None}244 # g7 store get, then cleanup245 st, s, _ = gemini_request("GET", f"/v1beta/{name}", note="gemini-tools:g7 get store")246 save_sanitized({"status": st, "response": s}, OUT / "g7_store_get.json")247 SUMMARY["g7_store_get"] = {"status": st, "store": s}248 st_list, lst, _ = gemini_request("GET", "/v1beta/fileSearchStores?pageSize=5", note="gemini-tools:g7b list stores")249 SUMMARY["g7b_stores_list"] = {"status": st_list, "keys": sorted(lst.keys()) if isinstance(lst, dict) else None}250 for d in doc_list:251 st, r, _ = gemini_request("DELETE", f"/v1beta/{d['name']}", note="gemini-tools:g8 delete document")252 SUMMARY.setdefault("g8_document_delete", []).append({"status": st, "body": r})253 st, r, _ = gemini_request("DELETE", f"/v1beta/{name}?force=true", note="gemini-tools:g9 delete store force")254 SUMMARY["g9_store_delete"] = {"status": st, "body": r}255256257def step_h_interactions() -> None:258 st, r, _ = gemini_request("GET", "/v1beta/interactions", note="gemini-tools:h0 list interactions probe")259 save_sanitized({"status": st, "response": r}, OUT / "h0_interactions_list.json")260 SUMMARY["h0_interactions_list"] = {"status": st, "keys": sorted(r.keys()) if isinstance(r, dict) else None, "error": (r.get("error") or {}).get("message", "")[:200] if isinstance(r, dict) else None}261 body = {"model": LITE, "input": "Reply with OK.", "generation_config": {"max_output_tokens": 16}}262 st, r, h = gemini_request("POST", "/v1beta/interactions", body, note="gemini-tools:h1 create interaction")263 save_sanitized({"request": body, "status": st, "response": r}, OUT / "h1_interaction_create.json")264 SUMMARY["h1_interaction_create"] = {"status": st, "keys": sorted(r.keys()) if isinstance(r, dict) else None, "id_prefix": (r.get("id") or "")[:6] if isinstance(r, dict) else None,265 "status_field": r.get("status") if isinstance(r, dict) else None, "outputs_types": [o.get("type") for o in steps(r)] if isinstance(r, dict) else None,266 "usage": r.get("usage") if isinstance(r, dict) else None, "error": r.get("error") if isinstance(r, dict) else None}267 if st != 200:268 return269 iid = r["id"]270 # h2 stream271 body = {"model": LITE, "input": "Reply with OK.", "stream": True, "previous_interaction_id": iid, "generation_config": {"max_output_tokens": 16}}272 st, lines, _ = gemini_request("POST", "/v1beta/interactions", body, stream=True, note="gemini-tools:h2 stream + previous_interaction_id")273 events: list[dict] = []274 raw: list[str] = []275 if st == 200:276 for ln in lines:277 raw.append(ln)278 if ln.startswith("data:"):279 try:280 events.append(json.loads(ln[5:].strip()))281 except Exception: # noqa: BLE001282 pass283 else:284 raw = [json.dumps(lines)] if not hasattr(lines, "__next__") else list(lines)285 save_sanitized({"request": body, "status": st, "raw_lines": raw, "events": events}, OUT / "h2_interaction_stream.json")286 SUMMARY["h2_interaction_stream"] = {"status": st, "event_types": [e.get("event_type") for e in events], "sse_field_names": sorted({ln.split(":")[0] for ln in raw if ":" in ln})[:6],287 "last_event_keys": sorted(events[-1].keys()) if events else None}288 iid2 = next((e.get("interaction", {}).get("id") for e in events if e.get("event_type") == "interaction.complete"), None) or \289 next((e.get("interaction", {}).get("id") for e in events if isinstance(e.get("interaction"), dict)), None)290 # h3 chaining non-stream291 body = {"model": LITE, "input": "What did I ask you first? 5 words.", "previous_interaction_id": iid, "generation_config": {"max_output_tokens": 24}}292 st, r3, _ = gemini_request("POST", "/v1beta/interactions", body, note="gemini-tools:h3 previous_interaction_id chain")293 save_sanitized({"request": body, "status": st, "response": r3}, OUT / "h3_interaction_chain.json")294 SUMMARY["h3_interaction_chain"] = {"status": st, "previous_interaction_id_echo": r3.get("previous_interaction_id") if isinstance(r3, dict) else None,295 "text": " ".join(c.get("text", "") for o in steps(r3) for c in ([o] if o.get("type") == "text" else o.get("content", []))) [:120] if isinstance(r3, dict) else None,296 "error": r3.get("error") if isinstance(r3, dict) else None}297 # h4 GET298 st, r4, _ = gemini_request("GET", f"/v1beta/interactions/{iid}", note="gemini-tools:h4 get interaction")299 save_sanitized({"status": st, "response": r4}, OUT / "h4_interaction_get.json")300 SUMMARY["h4_interaction_get"] = {"status": st, "keys": sorted(r4.keys()) if isinstance(r4, dict) else None}301 # h4b GET with query params (steps/outputs) probe302 st, r4b, _ = gemini_request("GET", f"/v1beta/interactions/{iid}?include_steps=true", note="gemini-tools:h4b get interaction include_steps probe")303 SUMMARY["h4b_interaction_get_steps_probe"] = {"status": st, "keys": sorted(r4b.keys()) if isinstance(r4b, dict) else None}304 # h5 store:false305 body = {"model": LITE, "input": "Reply with OK.", "store": False, "generation_config": {"max_output_tokens": 8}}306 st, r5, _ = gemini_request("POST", "/v1beta/interactions", body, note="gemini-tools:h5 store=false")307 SUMMARY["h5_store_false"] = {"status": st, "id_present": bool(isinstance(r5, dict) and r5.get("id")), "keys": sorted(r5.keys()) if isinstance(r5, dict) else None}308 if isinstance(r5, dict) and r5.get("id"):309 st, r5b, _ = gemini_request("GET", f"/v1beta/interactions/{r5['id']}", note="gemini-tools:h5b get unstored interaction")310 SUMMARY["h5b_get_unstored"] = {"status": st, "error": (r5b.get("error") or {}).get("message", "")[:200] if isinstance(r5b, dict) else None}311 # h6 function tool in interactions312 body = {"model": LITE, "input": "Weather in Paris?", "tools": [{"type": "function", "name": "get_weather", "description": "Get the weather for a city.",313 "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}],314 "tool_choice": {"allowed_tools": {"mode": "any"}}, "generation_config": {"max_output_tokens": 64}}315 st, r6, _ = gemini_request("POST", "/v1beta/interactions", body, note="gemini-tools:h6 interactions function tool")316 save_sanitized({"request": body, "status": st, "response": r6}, OUT / "h6_interaction_function_tool.json")317 SUMMARY["h6_interaction_function_tool"] = {"status": st, "outputs_types": [o.get("type") for o in steps(r6)] if isinstance(r6, dict) else None,318 "fc": next((o for o in steps(r6) if o.get("type") == "function_call"), None) if isinstance(r6, dict) else None,319 "error": r6.get("error") if isinstance(r6, dict) else None}320 if st == 200 and isinstance(r6, dict):321 fc = next((o for o in steps(r6) if o.get("type") == "function_call"), None)322 if fc:323 body = {"model": LITE, "previous_interaction_id": r6["id"], "input": [{"type": "function_result", "name": fc.get("name"), "call_id": fc.get("id"), "result": [{"type": "text", "text": "{\"temperature_c\": 18}"}]}],324 "tools": body["tools"], "generation_config": {"max_output_tokens": 32}}325 st, r6b, _ = gemini_request("POST", "/v1beta/interactions", body, note="gemini-tools:h6b function_result")326 save_sanitized({"request": body, "status": st, "response": r6b}, OUT / "h6b_interaction_function_result.json")327 SUMMARY["h6b_interaction_function_result"] = {"status": st, "outputs_types": [o.get("type") for o in steps(r6b)] if isinstance(r6b, dict) else None,328 "error": r6b.get("error") if isinstance(r6b, dict) else None}329 # h7 google_search tool in interactions330 body = {"model": LITE, "input": "What is today's date in Toronto? 5 words.", "tools": [{"type": "google_search"}], "generation_config": {"max_output_tokens": 48}}331 st, r7, _ = gemini_request("POST", "/v1beta/interactions", body, note="gemini-tools:h7 interactions google_search")332 save_sanitized({"request": body, "status": st, "response": r7}, OUT / "h7_interaction_google_search.json")333 SUMMARY["h7_interaction_google_search"] = {"status": st, "outputs_types": [o.get("type") for o in steps(r7)] if isinstance(r7, dict) else None,334 "error": r7.get("error") if isinstance(r7, dict) else None}335 # h8 deep research background + cancel immediately336 body = {"agent": "deep-research-preview-04-2026", "input": "Reply with OK. Do not research anything.", "background": True}337 st, r8, _ = gemini_request("POST", "/v1beta/interactions", body, note="gemini-tools:h8 deep research background create")338 save_sanitized({"request": body, "status": st, "response": r8}, OUT / "h8_deep_research_create.json")339 SUMMARY["h8_deep_research_create"] = {"status": st, "keys": sorted(r8.keys()) if isinstance(r8, dict) else None, "status_field": r8.get("status") if isinstance(r8, dict) else None,340 "agent": r8.get("agent") if isinstance(r8, dict) else None, "error": r8.get("error") if isinstance(r8, dict) else None}341 if st == 200 and isinstance(r8, dict) and r8.get("id"):342 st, r8b, _ = gemini_request("POST", f"/v1beta/interactions/{r8['id']}/cancel", note="gemini-tools:h8b cancel")343 save_sanitized({"status": st, "response": r8b}, OUT / "h8b_deep_research_cancel.json")344 SUMMARY["h8b_deep_research_cancel"] = {"status": st, "status_field": r8b.get("status") if isinstance(r8b, dict) else None, "error": r8b.get("error") if isinstance(r8b, dict) else None}345 time.sleep(2)346 st, r8c, _ = gemini_request("GET", f"/v1beta/interactions/{r8['id']}", note="gemini-tools:h8c get after cancel")347 SUMMARY["h8c_deep_research_after_cancel"] = {"status": st, "status_field": r8c.get("status") if isinstance(r8c, dict) else None, "usage": r8c.get("usage") if isinstance(r8c, dict) else None}348 st, r8d, _ = gemini_request("DELETE", f"/v1beta/interactions/{r8['id']}", note="gemini-tools:h8d delete deep research interaction")349 SUMMARY["h8d_deep_research_delete"] = {"status": st}350 # h9 DELETE351 st, r9, _ = gemini_request("DELETE", f"/v1beta/interactions/{iid}", note="gemini-tools:h9 delete interaction")352 save_sanitized({"status": st, "response": r9}, OUT / "h9_interaction_delete.json")353 SUMMARY["h9_interaction_delete"] = {"status": st, "body": r9 if not isinstance(r9, bytes) else r9[:100].decode("utf-8", "replace")}354 st, r9b, _ = gemini_request("GET", f"/v1beta/interactions/{iid}", note="gemini-tools:h9b get after delete")355 SUMMARY["h9b_get_after_delete"] = {"status": st, "error": (r9b.get("error") or {}).get("message", "")[:200] if isinstance(r9b, dict) else None}356 for extra in [iid2, (r3 or {}).get("id") if isinstance(r3, dict) else None, (r6 or {}).get("id") if isinstance(r6, dict) else None, (r7 or {}).get("id") if isinstance(r7, dict) else None]:357 if extra:358 gemini_request("DELETE", f"/v1beta/interactions/{extra}", note="gemini-tools:h9c cleanup")359360361def step_h_environments() -> None:362 st, r, _ = gemini_request("GET", "/v1beta/environments", note="gemini-tools:env1 list environments")363 save_sanitized({"status": st, "response": r}, OUT / "env1_environments_list.json")364 SUMMARY["env1_environments_list"] = {"status": st, "keys": sorted(r.keys()) if isinstance(r, dict) else None, "error": (r.get("error") or {}).get("message", "")[:200] if isinstance(r, dict) else None}365 st, r, _ = gemini_request("POST", "/v1beta/environments", {}, note="gemini-tools:env2 create empty environment")366 save_sanitized({"status": st, "response": r}, OUT / "env2_environment_create.json")367 SUMMARY["env2_environment_create"] = {"status": st, "keys": sorted(r.keys()) if isinstance(r, dict) else None, "response": r if isinstance(r, dict) and st == 200 else (r.get("error") if isinstance(r, dict) else None)}368 eid = r.get("id") or r.get("name") if isinstance(r, dict) and st == 200 else None369 if eid:370 eid_short = eid.split("/")[-1]371 st, g, _ = gemini_request("GET", f"/v1beta/environments/{eid_short}", note="gemini-tools:env3 get environment")372 SUMMARY["env3_environment_get"] = {"status": st, "keys": sorted(g.keys()) if isinstance(g, dict) else None}373 st, f, _ = gemini_request("GET", f"/v1beta/environments/{eid_short}/files/", note="gemini-tools:env4 list root files")374 save_sanitized({"status": st, "response": f if not isinstance(f, bytes) else f[:500].decode("utf-8", "replace")}, OUT / "env4_environment_files.json")375 SUMMARY["env4_environment_files"] = {"status": st, "keys": sorted(f.keys()) if isinstance(f, dict) else None}376 st, d, _ = gemini_request("DELETE", f"/v1beta/environments/{eid_short}", note="gemini-tools:env5 delete environment")377 SUMMARY["env5_environment_delete"] = {"status": st, "body": d if not isinstance(d, bytes) else d[:100].decode("utf-8", "replace")}378 # discovery-doc style paths probe (environments:list)379 st, r, _ = gemini_request("GET", "/v1beta/environments:list", note="gemini-tools:env6 discovery-style path probe")380 SUMMARY["env6_environments_colon_list"] = {"status": st, "keys": sorted(r.keys()) if isinstance(r, dict) else None}381382383def step_i_auth_tokens() -> None:384 now = datetime.now(timezone.utc)385 body = {"uses": 1, "expireTime": (now + timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%SZ"), "newSessionExpireTime": (now + timedelta(minutes=2)).strftime("%Y-%m-%dT%H:%M:%SZ")}386 for ver in ("v1alpha", "v1beta"):387 st, r, _ = gemini_request("POST", f"/{ver}/auth_tokens", body, note=f"gemini-tools:i auth_tokens create {ver}")388 red = dict(r) if isinstance(r, dict) else r389 if isinstance(red, dict) and red.get("name"):390 red["name"] = red["name"].split("/")[0] + "/***REDACTED-TOKEN***"391 save_sanitized({"request": body, "status": st, "response": red}, OUT / f"i_auth_token_{ver}.json")392 SUMMARY[f"i_auth_token_{ver}"] = {"status": st, "keys": sorted(r.keys()) if isinstance(r, dict) else None, "name_prefix": (r.get("name") or "")[:12] if isinstance(r, dict) else None,393 "error": (r.get("error") or {}).get("message", "")[:200] if isinstance(r, dict) else None}394 if st == 200 and isinstance(r, dict):395 (OUT / f".token_{ver}").write_text(r["name"]) # consumed by run_live_ws.py, deleted afterwards396397398STEPS = {"a": step_a_function_calling, "b": step_b_google_search, "c": step_c_url_context, "d": step_d_code_execution, "e": step_e_google_maps,399 "f": step_f_computer_use, "g": step_g_file_search, "h": step_h_interactions, "env": step_h_environments, "i": step_i_auth_tokens}400401if __name__ == "__main__":402 OUT.mkdir(parents=True, exist_ok=True)403 sel = sys.argv[1:] or list(STEPS)404 for s in sel:405 try:406 STEPS[s]()407 except Exception as e: # noqa: BLE001408 SUMMARY[f"{s}_EXCEPTION"] = {"error": repr(e)[:300], "trace": traceback.format_exc()[-800:]}409 print(f"step {s} done")410 prev = json.loads((OUT / "_summary.json").read_text()) if (OUT / "_summary.json").exists() else {}411 prev.update(SUMMARY)412 save_sanitized(prev, OUT / "_summary.json")413 print(json.dumps(SUMMARY, indent=1, default=str)[:12000])414