Python 88.3%
TypeScript 7.6%
Shell 4.1%
1#!/usr/bin/env python32"""Live API (BidiGenerateContent WebSocket) probes — API Atlas 2026-09-18.31. raw websockets, x-goog-api-key header, text-only turn on each candidate model -> ordered server message keys42. google-genai SDK client.aio.live.connect variant53. ephemeral token (v1alpha auth_tokens) -> WebSocket with ?access_token= (token never printed)6Logged via scripts.live.log_request; raws -> tmp-live/gemini-tools/j_*.json7Run: .venv/bin/python tmp-live/gemini-tools/run_live_ws.py8"""9from __future__ import annotations1011import asyncio12import json13import os14import sys15from datetime import datetime, timedelta, timezone16from pathlib import Path1718import websockets1920ROOT = Path(__file__).resolve().parents[2]21sys.path.insert(0, str(ROOT))22from scripts.live import gemini_request, log_request, save_sanitized, mask # noqa: E4022324OUT = ROOT / "tmp-live" / "gemini-tools"25HOST = "wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.{ver}.GenerativeService.BidiGenerateContent"26MODELS = ["gemini-2.5-flash-native-audio-latest", "gemini-3.1-flash-live-preview", "gemini-3.8-live"]27S: dict = json.loads((OUT / "_summary.json").read_text()) if (OUT / "_summary.json").exists() else {}282930def redact(obj):31 """Drop audio payloads / signatures for the saved raw; keep structure."""32 if isinstance(obj, dict):33 out = {}34 for k, v in obj.items():35 if k in ("data", "thoughtSignature", "newHandle", "handle") and isinstance(v, str) and len(v) > 40:36 out[k] = f"<{len(v)} chars>"37 else:38 out[k] = redact(v)39 return out40 if isinstance(obj, list):41 return [redact(x) for x in obj]42 return obj434445async def text_turn(model: str, ver: str = "v1beta", headers: dict | None = None, query: str = "", step: str = "", modality: str = "TEXT",46 extra_setup: dict | None = None, prompt: str = "Reply with OK.") -> dict:47 url = HOST.format(ver=ver) + query48 seq: list[str] = []49 msgs: list[dict] = []50 result: dict = {"model": model, "ver": ver, "modality": modality}51 t0 = datetime.now(timezone.utc)52 try:53 async with websockets.connect(url, additional_headers=headers or {}, max_size=2**24) as ws:54 setup = {"setup": {"model": f"models/{model}", "generationConfig": {"responseModalities": [modality]}, **(extra_setup or {})}}55 await ws.send(json.dumps(setup))56 first = json.loads(await asyncio.wait_for(ws.recv(), 30))57 msgs.append(first)58 seq.append(",".join(first.keys()))59 if "setupComplete" not in first:60 result["error"] = "no setupComplete"61 await ws.send(json.dumps({"clientContent": {"turns": [{"role": "user", "parts": [{"text": prompt}]}], "turnComplete": True}}))62 for _ in range(60):63 raw = await asyncio.wait_for(ws.recv(), 30)64 m = json.loads(raw)65 msgs.append(m)66 keys = list(m.keys())67 sc = m.get("serverContent") or {}68 sub = [k for k in sc.keys() if k != "modelTurn"] + (["modelTurn(" + ",".join(",".join(p.keys()) for p in sc.get("modelTurn", {}).get("parts", [])) + ")"] if "modelTurn" in sc else [])69 seq.append(",".join(keys) + (":" + "+".join(sub) if sub else ""))70 if sc.get("turnComplete"):71 break72 # wait briefly for trailing usageMetadata / goAway73 try:74 while True:75 m = json.loads(await asyncio.wait_for(ws.recv(), 2))76 msgs.append(m)77 seq.append(",".join(m.keys()))78 except (asyncio.TimeoutError, websockets.ConnectionClosed):79 pass80 result["status"] = 10181 except websockets.InvalidStatus as e: # HTTP error during handshake82 result["status"] = e.response.status_code83 result["error"] = mask(e.response.body.decode("utf-8", "replace")[:400]) if e.response.body else str(e)84 except websockets.ConnectionClosed as e:85 result["status"] = 100086 result["close_code"] = e.rcvd.code if e.rcvd else None87 result["close_reason"] = mask((e.rcvd.reason if e.rcvd else "") or "")[:400]88 except Exception as e: # noqa: BLE00189 result["status"] = 090 result["error"] = mask(repr(e))[:400]91 result["sequence"] = seq92 text = "".join(p.get("text", "") for m in msgs for p in (m.get("serverContent") or {}).get("modelTurn", {}).get("parts", []))93 result["text"] = text[:200]94 result["usageMetadata"] = next((m["usageMetadata"] for m in reversed(msgs) if "usageMetadata" in m), None)95 result["elapsed_s"] = round((datetime.now(timezone.utc) - t0).total_seconds(), 2)96 usage = result["usageMetadata"] or {}97 est = (usage.get("promptTokenCount", 0) * 3.0 + usage.get("responseTokenCount", 0) * 12.0) / 1e6 # conservative Live audio-model text rates98 log_request("gemini", "WSS", f"/ws/google.ai.generativelanguage.{ver}.GenerativeService.BidiGenerateContent", result["status"], est,99 f"gemini-tools:{step} model={model} modality={modality} seq={len(seq)} {result.get('error') or ''}")100 save_sanitized({"setup": setup if "setup" in dir() else None, "result": result, "messages": redact(msgs)}, OUT / f"{step}.json")101 return result102103104async def sdk_variant(model: str) -> dict:105 from google import genai106 client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])107 seq: list[str] = []108 res: dict = {"model": model}109 try:110 async with client.aio.live.connect(model=model, config={"response_modalities": ["AUDIO"], "output_audio_transcription": {}}) as session:111 await session.send_client_content(turns={"role": "user", "parts": [{"text": "Reply with OK."}]}, turn_complete=True)112 text = ""113 async for msg in session.receive():114 kinds = [k for k in ("setup_complete", "server_content", "tool_call", "tool_call_cancellation", "usage_metadata", "go_away", "session_resumption_update") if getattr(msg, k, None) is not None]115 seq.append("+".join(kinds))116 if msg.text:117 text += msg.text118 if msg.server_content and msg.server_content.output_transcription and msg.server_content.output_transcription.text:119 text += msg.server_content.output_transcription.text120 if msg.usage_metadata:121 res["usage"] = msg.usage_metadata.model_dump(exclude_none=True)122 if msg.server_content and msg.server_content.turn_complete:123 break124 res.update(status=101, text=text[:100], sequence=seq)125 except Exception as e: # noqa: BLE001126 res.update(status=0, error=mask(repr(e))[:400], sequence=seq)127 log_request("gemini", "WSS", "/ws/…BidiGenerateContent (google-genai SDK client.aio.live.connect)", res["status"], 0.0005, f"gemini-tools:j_sdk model={model} {res.get('error') or ''}")128 return res129130131async def main() -> None:132 key = os.environ["GEMINI_API_KEY"]133 hdr = {"x-goog-api-key": key}134 # 1. raw websocket text-only turn per model135 for i, m in enumerate(MODELS):136 # TEXT was rejected (close 1007) on all three models in the first run -> AUDIO + output transcription137 r = await text_turn(m, headers=hdr, step=f"j{i+1}_ws_{m}", modality="AUDIO", extra_setup={"outputAudioTranscription": {}})138 S[f"j{i+1}_ws_{m}"] = r139 print(m, r.get("status"), r.get("error"), r.get("sequence"))140 if r.get("status") == 101:141 break142 ok_model = next((m for i, m in enumerate(MODELS) if S.get(f"j{i+1}_ws_{m}", {}).get("status") == 101), None)143 # 1b. audio modality + transcription + tools quick probe (one turn) on the working model144 if ok_model:145 r = await text_turn(ok_model, headers=hdr, step="j4_ws_audio_transcription", modality="AUDIO",146 extra_setup={"outputAudioTranscription": {}, "inputAudioTranscription": {}, "sessionResumption": {}, "contextWindowCompression": {"slidingWindow": {}}})147 S["j4_ws_audio_transcription"] = {k: v for k, v in r.items()}148 print("audio", r.get("status"), r.get("error"), r.get("sequence"))149 # 1c. tools in Live: NON_BLOCKING function declaration, mode ANY -> toolCall message150 tools = [{"functionDeclarations": [{"name": "get_weather", "description": "Get the weather for a city.", "behavior": "NON_BLOCKING",151 "parameters": {"type": "OBJECT", "properties": {"city": {"type": "STRING"}}, "required": ["city"]}}]}]152 r = await text_turn(ok_model, headers=hdr, step="j5_ws_toolcall", prompt="Weather in Paris? Use the tool.",153 extra_setup={"tools": tools, "toolConfig": {"functionCallingConfig": {"mode": "ANY"}}})154 S["j5_ws_toolcall"] = r155 print("toolcall", r.get("status"), r.get("error"), r.get("sequence"))156 # 2. SDK variant157 if ok_model:158 r = await sdk_variant(ok_model)159 S["j6_sdk_live_connect"] = r160 print("sdk", r)161 # 3. ephemeral token -> WebSocket (token passed as access_token query param; docs say v1alpha only)162 target = ok_model or MODELS[2]163 variants = [("j8a_ephemeral_query_v1alpha", "v1alpha", "query", None),164 ("j8b_ephemeral_header_v1alpha", "v1alpha", "header", None),165 ("j8c_ephemeral_query_v1alpha_constrained", "v1alpha", "query", {"bidiGenerateContentSetup": {"model": f"models/{target}"}})]166 for step, ver, how, constraint in variants:167 now = datetime.now(timezone.utc)168 body = {"uses": 1, "expireTime": (now + timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%SZ"),169 "newSessionExpireTime": (now + timedelta(minutes=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), **(constraint or {})}170 st, tok, _ = gemini_request("POST", f"/{ver}/auth_tokens", body, note=f"gemini-tools:{step} create token ({'constrained' if constraint else 'plain'})")171 S[step + "_create"] = {"status": st, "keys": sorted(tok.keys()) if isinstance(tok, dict) else None, "error": (tok.get("error") or {}).get("message", "")[:300] if isinstance(tok, dict) else None}172 if st != 200:173 print("token create failed", step, S[step + "_create"])174 continue175 token = tok["name"]176 if how == "query":177 r = await text_turn(target, ver=ver, query=f"?access_token={token}", step=step, modality="AUDIO", extra_setup={"outputAudioTranscription": {}})178 else:179 r = await text_turn(target, ver=ver, headers={"Authorization": f"Token {token}"}, step=step, modality="AUDIO", extra_setup={"outputAudioTranscription": {}})180 S[step] = r181 print("ephemeral", step, r.get("status"), r.get("error"), r.get("close_code"), r.get("close_reason"), r.get("sequence"))182 if r.get("status") == 101:183 break184 for f in OUT.glob(".token_*"):185 f.unlink()186 save_sanitized(S, OUT / "_summary.json")187188189if __name__ == "__main__":190 asyncio.run(main())191