> ## Documentation Index > Fetch the complete documentation index at: https://docs.playapi.ai/llms.txt > Use this file to discover all available pages before exploring further. # Native Gemini SSE and tool loops > Consume native events, validate calls, preserve reasoning state, and return tool results. This page extends [native Gemini integration](/en/guides/native-gemini) with one Python file supporting text SSE and function tool loops. Both modes consume SSE and print the final visible answer only after validation; partial text is not displayed as it arrives. Confirm native streaming and function support with your deployment administrator. ## Setup and execution Use Python 3.10+ with the standard library; no provider SDK is required. Set your gateway token in `API_KEY`, then use a root URL without `/v1` or `/v1beta`. Deployment path prefixes are supported. ```sh theme={null} 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. ``` Run from the documentation directory, or omit `examples/` if you saved the file separately. The first command requests a greeting; the second offers `multiply`, asks for 7 × 8, and returns the calculation to the model for its final answer. ```sh theme={null} API_NATIVE_MODE=text python3 examples/native-gemini-stream.py API_NATIVE_MODE=tools python3 examples/native-gemini-stream.py ``` `API_NATIVE_MODE` defaults to `text`. In `tools` mode, a model may also answer directly. Test responses are fixed fixtures; real wording and tool selection are not guaranteed. ## Events and the tool loop | Stage | Example behavior | | - | - | | Request | `POST /v1beta/models/{model}:streamGenerateContent?alt=sse` with `x-goog-api-key` | | SSE data | Each event is a Gemini JSON response, not a Chat delta or Responses event | | Candidate | Accept one candidate at index 0; preserve content parts in arrival order | | Completion | Require `finishReason: STOP` and clean EOF; allow trailing usage-only frames but reject further candidates | | Tools | Define `tools[].functionDeclarations`; read complete `functionCall.args` | | Results | Replay complete model parts, then return all `functionResponse` parts in one user content | When a call supplies an ID, the example echoes it unchanged and rejects empty or repeated IDs. Without IDs, it associates calls and results by order and name, without inventing IDs. Repeated calls to the same function each receive their own result. Every thought signature remains on its original part, including empty text and signature-only carriers. Parts are neither merged nor flattened. Thoughts are hidden only from terminal display, not removed from replay history. Verify signature forwarding on your actual channel. This example uses complete function argument objects and does not enable fine-grained argument streaming. It fails on `partialArgs` or `willContinue: true` instead of treating them as complete calls. It does not concatenate arbitrary JSON strings as Claude does. Complete calls can arrive in separate SSE frames; execution waits for candidate completion. Bare model ID restrictions match the introductory example. ## Complete Python example ```python theme={null} 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.") ``` ## Failure and resource boundaries The SSE reader handles incremental UTF-8, LF / CRLF / CR, comments, and multiline data. It ignores SSE id/retry fields rather than replaying a request automatically. Socket operations time out after 60 seconds; this is not a deadline for the entire workflow. Each turn allows at most 512 data events, 4096 lines, and 1,048,576 decoded characters, with 65,536 characters per line. These are example safeguards, not gateway limits. Only the local pure calculation `multiply` is allowed. Arguments must contain exactly finite numeric a and b; booleans, extra fields, and nonfinite results fail. At most three model requests and four calls per turn are allowed. A third turn requesting more tools fails. All results are prepared before another request. Tools with side effects need separate authorization, idempotent execution, and recovery; multiplication is not a transaction design. HTTP failures, redirects, wrong content types, malformed JSON, truncated SSE, broken connections, unsuccessful completion, and unsupported extensions exit nonzero. Credentials, upstream error bodies, thinking text, and incomplete answers are not printed. Unknown events or blocks fail explicitly until the parser is extended to support them. Gemini additionally limits each turn to 256 parts and serialized arguments to 16,384 characters. Fine-grained argument fragments, multiple candidates, media, Live API, and asynchronous tool scheduling are outside this example. ## Verification Run `npm run check:examples`. Tests execute this script against a local HTTP fixture, covering text/tool modes, complete history and signatures, result association, split UTF-8, errors, disconnects, no retries, and second-turn failure short-circuiting. No real models are called; actual event delivery and signature preservation require deployment testing. [Gemini native API](https://ai.google.dev/api/generate-content) · [Signature preservation](https://ai.google.dev/gemini-api/docs/generate-content/thought-signatures) [阅读中文](/zh/guides/native-gemini-streaming) This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.