#!/usr/bin/env python3 """ grounded_search.py — Gemini grounded web search for agents. A zero-dependency (stdlib only) wrapper around Google's Gemini `generateContent` API with the `google_search` grounding tool. Returns a synthesized answer PLUS the real web sources Gemini grounded it on — so any agent can fetch current, cited facts instead of relying on stale training data. Why this exists: the model's knowledge has a cutoff. Grounded search closes the gap. Use it for anything time-sensitive (product launches, traction, numbers). CLI: python grounded_search.py "your question" python grounded_search.py "your question" --json # machine-readable python grounded_search.py "your question" --no-grounding # plain model answer python grounded_search.py "your question" --model gemini-flash-latest --system "Be terse." Import: from grounded_search import search res = search("who shipped Claude Cowork and when?") print(res["answer"]) for s in res["sources"]: print(s["title"], s["url"]) API key resolution (first match wins; NEVER hardcode the key in the repo — this vault publishes to a website on git push): 1. $GEMINI_API_KEY 2. file at $GEMINI_API_KEY_FILE 3. ~/.config/gemini/api_key (recommended: lives outside the repo) """ from __future__ import annotations import argparse import json import os import sys import time import urllib.error import urllib.request from pathlib import Path API_URL = "https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent" DEFAULT_MODEL = "gemini-flash-latest" class SearchError(RuntimeError): """Raised for any auth / quota / network / parse failure.""" def _load_api_key() -> str: """Resolve the API key without ever requiring it to live in the repo.""" env_key = os.environ.get("GEMINI_API_KEY") if env_key and env_key.strip(): return env_key.strip() candidates = [] file_env = os.environ.get("GEMINI_API_KEY_FILE") if file_env: candidates.append(Path(file_env)) candidates.append(Path.home() / ".config" / "gemini" / "api_key") for path in candidates: try: if path.is_file(): key = path.read_text(encoding="utf-8").strip() if key: return key except OSError: continue raise SearchError( "No Gemini API key found. Provide it via one of:\n" " - $GEMINI_API_KEY environment variable\n" " - $GEMINI_API_KEY_FILE pointing at a key file\n" " - ~/.config/gemini/api_key (recommended; chmod 600)\n" "Do NOT commit the key into the vault — it publishes to the website." ) def search( query: str, *, model: str = DEFAULT_MODEL, grounded: bool = True, system: str | None = None, timeout: int = 90, max_retries: int = 4, ) -> dict: """Run a grounded search. Returns a dict with answer + sources + metadata.""" api_key = _load_api_key() body: dict = {"contents": [{"parts": [{"text": query}]}]} if grounded: body["tools"] = [{"google_search": {}}] if system: body["system_instruction"] = {"parts": [{"text": system}]} data = json.dumps(body).encode("utf-8") url = API_URL.format(model=model) headers = {"Content-Type": "application/json", "x-goog-api-key": api_key} last_err: SearchError | None = None for attempt in range(max_retries): req = urllib.request.Request(url, data=data, headers=headers, method="POST") try: with urllib.request.urlopen(req, timeout=timeout) as resp: payload = json.loads(resp.read().decode("utf-8")) return _parse(query, model, payload) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace") # 429 = quota/rate limit; 5xx = transient. Back off and retry. if exc.code in (429, 500, 502, 503) and attempt < max_retries - 1: wait = _retry_after(exc) or float(2 ** attempt) time.sleep(min(wait, 30.0)) last_err = SearchError(_explain_http(exc.code, detail)) continue raise SearchError(_explain_http(exc.code, detail)) from exc except urllib.error.URLError as exc: last_err = SearchError(f"Network error: {exc.reason}") if attempt < max_retries - 1: time.sleep(float(2 ** attempt)) continue raise last_err from exc raise last_err or SearchError("Unknown error contacting Gemini API.") def _retry_after(exc: urllib.error.HTTPError) -> float | None: if not exc.headers: return None raw = exc.headers.get("Retry-After") if not raw: return None try: return float(raw) except (TypeError, ValueError): return None def _explain_http(code: int, detail: str) -> str: hints = { 401: "Invalid or missing API key.", 403: "Key valid but not authorized for this model/feature (check API enablement).", 429: "Quota/rate limit exceeded. Grounding (google_search) usually requires " "billing enabled on the Google Cloud project; or you've hit the free tier. " "Enable billing or wait, then retry.", 400: "Bad request (check model name and payload shape).", } hint = hints.get(code, "") return f"HTTP {code} from Gemini API. {hint}\nResponse: {detail}".rstrip() def _parse(query: str, model: str, payload: dict) -> dict: candidates = payload.get("candidates") or [] if not candidates: feedback = payload.get("promptFeedback") raise SearchError(f"No candidates returned. promptFeedback={feedback}") cand = candidates[0] parts = ((cand.get("content") or {}).get("parts")) or [] answer = "".join(p.get("text", "") for p in parts).strip() meta = cand.get("groundingMetadata") or {} sources: list[dict] = [] seen: set[str] = set() for chunk in meta.get("groundingChunks") or []: web = chunk.get("web") or {} uri = web.get("uri") if uri and uri not in seen: seen.add(uri) sources.append({"title": web.get("title") or uri, "url": uri}) return { "query": query, "model": model, "answer": answer, "sources": sources, "queries_run": meta.get("webSearchQueries") or [], "grounded": bool(meta), "usage": payload.get("usageMetadata") or {}, } def _main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="grounded_search.py", description="Gemini grounded web search — current, cited facts for agents.", ) parser.add_argument("query", help="the search query / question") parser.add_argument("--model", default=DEFAULT_MODEL, help=f"default: {DEFAULT_MODEL}") parser.add_argument("--system", default=None, help="optional system instruction") parser.add_argument("--no-grounding", action="store_true", help="disable google_search grounding (plain model answer)") parser.add_argument("--json", action="store_true", help="emit raw JSON result") args = parser.parse_args(argv) # Force UTF-8 output so grounded text (em-dashes, quotes, accents) survives # a Windows cp1252 console — otherwise it mojibakes into captured content. for stream in (sys.stdout, sys.stderr): try: stream.reconfigure(encoding="utf-8") except (AttributeError, ValueError): pass try: result = search( args.query, model=args.model, grounded=not args.no_grounding, system=args.system, ) except SearchError as exc: print(f"ERROR: {exc}", file=sys.stderr) return 1 if args.json: print(json.dumps(result, indent=2, ensure_ascii=False)) else: print(result["answer"] or "(no answer text returned)") if result["sources"]: print("\nSources:") for i, src in enumerate(result["sources"], 1): print(f" [{i}] {src['title']} — {src['url']}") elif not args.no_grounding: print("\n(no grounding sources returned)") return 0 if __name__ == "__main__": raise SystemExit(_main())