Native Gemini SSE and tool loops

export API_ROOT_URL="https://gateway.example.com"
export API_MODEL="YOUR_NATIVE_MODEL_ID"
# API_KEY must already be set in the process environment.

API_NATIVE_MODE=text python3 examples/native-gemini-stream.py
API_NATIVE_MODE=tools python3 examples/native-gemini-stream.py

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.")