import http.client
import io
import json
import math
import os
import re
import urllib.error
import urllib.request
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
def read_sse(response):
# Universal newlines and incremental UTF-8 decoding handle split network bytes.
if response.headers.get_content_type() != "text/event-stream":
raise ValueError("Expected SSE.")
event, data, size, events, lines = "message", [], 0, 0, 0
with io.TextIOWrapper(response, encoding="utf-8", newline=None) as reader:
while True:
line = reader.readline(65_537)
if not line:
if data:
raise ValueError("Unterminated SSE event.")
return
lines += 1
size += len(line)
if len(line) > 65_536 or size > 1_048_576 or lines > 4096:
raise ValueError("Stream size limit.")
if not line.endswith("\n"):
raise ValueError("Unterminated SSE line.")
line = line[:-1]
if not line:
if data:
events += 1
if events > 512:
raise ValueError("Event limit.")
yield event, json.loads("\n".join(data))
event, data = "message", []
elif not line.startswith(":"):
field, _, value = line.partition(":")
if value.startswith(" "):
value = value[1:]
if field == "event":
event = value
elif field == "data":
data.append(value)
root_url = os.environ.get("API_ROOT_URL", "").rstrip("/")
key = os.environ.get("API_KEY", "")
model = os.environ.get("API_MODEL", "")
mode = os.environ.get("API_NATIVE_MODE", "text")
if not root_url or not key or not model or mode not in ("text", "tools"):
raise SystemExit("Set API_ROOT_URL, API_KEY, API_MODEL; API_NATIVE_MODE must be text or tools.")
opener = urllib.request.build_opener(NoRedirect())
schema = {"type": "object", "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
"required": ["a", "b"]}
prompt = "What is 7 times 8? Use multiply." if mode == "tools" else "Say hello in one short sentence."
seen_ids = set()
if not re.fullmatch(r"[A-Za-z0-9._-]+", model) or model in (".", ".."):
raise SystemExit("Use a bare model ID without a path or action suffix.")
history = [{"role": "user", "parts": [{"text": prompt}]}]
try:
for round_index in range(3):
body = {"contents": history, "systemInstruction": {"parts": [
{"text": "Answer briefly; use multiply for arithmetic when available."}]}}
if mode == "tools":
body["tools"] = [{"functionDeclarations": [{"name": "multiply",
"description": "Multiply two finite numbers.", "parameters": schema}]}]
# Default complete functionCall.args mode; do not enable partialArgs streaming.
request = urllib.request.Request(root_url + "/v1beta/models/" + model + ":streamGenerateContent?alt=sse",
method="POST", headers={"Content-Type": "application/json", "Accept": "text/event-stream", "x-goog-api-key": key},
data=json.dumps(body, allow_nan=False).encode())
parts, reason = [], None
with opener.open(request, timeout=60) as response:
for event_name, event in read_sse(response):
if event_name not in ("message", "") or event.get("error") or event.get("promptFeedback", {}).get("blockReason"):
raise ValueError("Native error or blocked prompt.")
candidates = event.get("candidates", [])
if not candidates:
if "usageMetadata" not in event:
raise ValueError("Missing candidate.")
continue
if not isinstance(candidates, list) or len(candidates) != 1 or reason is not None:
raise ValueError("Multiple candidates or output after completion.")
candidate = candidates[0]
if type(candidate.get("index", 0)) is not int or candidate.get("index", 0) != 0:
raise ValueError("Unexpected candidate index.")
content = candidate.get("content", {})
if content.get("role", "model") != "model" or not isinstance(content.get("parts", []), list):
raise ValueError("Invalid model content.")
for part in content.get("parts", []):
if not isinstance(part, dict) or set(part) - {"text", "thought", "thoughtSignature", "functionCall"}:
raise ValueError("Unsupported part.")
if "thought" in part and type(part["thought"]) is not bool:
raise ValueError("Invalid thought flag.")
if "thoughtSignature" in part and not isinstance(part["thoughtSignature"], str):
raise ValueError("Invalid signature.")
if "functionCall" in part:
call = part["functionCall"]
if (mode != "tools" or "text" in part or part.get("thought") or not isinstance(call, dict)
or "partialArgs" in call or call.get("willContinue") not in (None, False)
or set(call) - {"name", "args", "id", "willContinue"}):
raise ValueError("Unsupported or incomplete function call.")
elif "text" in part:
if not isinstance(part["text"], str):
raise ValueError("Invalid text.")
elif set(part) != {"thoughtSignature"}:
raise ValueError("Missing part content.")
parts.append(part)
if len(parts) > 256:
raise ValueError("Parts limit.")
if candidate.get("finishReason"):
reason = candidate["finishReason"]
if reason != "STOP":
raise ValueError("No successful finishReason.")
calls = [part["functionCall"] for part in parts if "functionCall" in part]
if not calls:
text = "".join(part.get("text", "") for part in parts if not part.get("thought", False))
if not text.strip():
raise ValueError("No visible answer.")
print(text)
break
if len(calls) > 4 or round_index == 2:
raise ValueError("Call or round limit.")
results = []
for call in calls:
args = call.get("args")
if call.get("name") != "multiply" or not isinstance(args, dict) or set(args) != {"a", "b"}:
raise ValueError("Unknown tool or invalid arguments.")
if len(json.dumps(args)) > 16_384:
raise ValueError("Arguments limit.")
if any(type(v) not in (int, float) or not math.isfinite(v) for v in args.values()):
raise ValueError("Expected finite numbers.")
value = args["a"] * args["b"]
if not math.isfinite(value):
raise ValueError("Nonfinite result.")
result = {"name": call["name"], "response": {"value": value}}
if "id" in call:
call_id = call["id"]
if not isinstance(call_id, str) or not call_id.strip() or call_id in seen_ids:
raise ValueError("Invalid or repeated call ID.")
seen_ids.add(call_id)
result["id"] = call_id
results.append({"functionResponse": result})
# Keep original part order and signature carriers; do not flatten model output.
history.extend([{"role": "model", "parts": parts}, {"role": "user", "parts": results}])
except urllib.error.HTTPError as error:
error.close()
raise SystemExit(f"Request failed (HTTP {error.code}). No automatic retry.")
except (http.client.HTTPException, OSError, ValueError, TypeError, KeyError, IndexError, AttributeError, OverflowError):
raise SystemExit("Native Gemini stream failed; no incomplete answer published. No automatic retry.")