> ## 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. # Gemini 原生 SSE 与工具闭环 > 接收原生事件、验证完整调用、保留推理状态并回传工具结果。 本页扩展 [原生 Gemini 入门](/zh/guides/native-gemini)，同一个 Python 文件提供纯文本 SSE 和函数工具闭环。两种模式都读取真实 SSE 数据流，完整校验后才打印最终可见答案；不会逐片向终端展示未完成文本。先向管理员确认模型和渠道支持原生流式及函数工具。 ## 准备与运行 使用 Python 3.10+，仅依赖标准库，无需安装供应商 SDK。预先设置网关令牌 `API_KEY`，然后设置不含 `/v1` 或 `/v1beta` 的根地址；允许部署路径前缀。 ```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. ``` 从文档目录运行；单独保存时，去掉命令中的 `examples/`。第一条只请求一句问候；第二条提供 `multiply`，让模型计算 7 × 8，再将结果回传给模型生成最终回答。 ```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` 默认 `text`；`tools` 模式也允许模型直接回答。测试固定了模拟回答，真实模型措辞和是否调用工具均不保证。 ## 事件与闭环 | 步骤 | 本例处理 | | - | - | | 请求 | `POST /v1beta/models/{model}:streamGenerateContent?alt=sse`，`x-goog-api-key` | | SSE data | 每个事件是一个 Gemini JSON 响应，不是 Chat delta 或 Responses 事件 | | 候选 | 仅接受索引 0 的单个候选；按接收顺序保存 `content.parts` | | 完成 | 要求 `finishReason: STOP` 和正常 EOF；允许末尾只有 usageMetadata 的帧，拒绝完成后再次出现候选 | | 工具 | `tools[].functionDeclarations` 定义；读取完整 `functionCall.args` | | 回传 | 先重放完整 model parts，再在一条 user 内容中回传全部 `functionResponse` | `functionCall.id` 存在时必须原样回传，重复或空 ID 会失败；没有 ID 时按调用顺序与名称关联，不杜撰 ID。即使同一函数调用多次，也逐一返回对应结果。 保留每个 part 上的 `thoughtSignature`，包括空文本和仅有签名的 part；不把多个签名合并，不把 model 内容压成一段文本。`thought: true` 仅影响终端展示，不会从回放历史删除。签名是否由目标渠道正确转发仍需联调。 本例使用默认的完整 `functionCall.args` 模式，没有启用细粒度参数流式扩展。遇到 `partialArgs` 或 `willContinue: true` 会退出，不能像 Claude 那样拼接任意 JSON 字符串。多个完整调用可以分布在多个 SSE 帧中；所有候选结束后才执行。模型 ID 沿用入门示例的裸 ID 字符限制。 ## 完整 Python 示例 ```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.") ``` ## 失败与资源边界 SSE 读取器处理增量 UTF-8、LF / CRLF / CR、注释及多行 `data`，忽略 `id` / `retry`，不会因服务端建议重连就自动重放请求。每次 socket 操作超时 60 秒；这不是整段工作流的总耗时上限。每轮最多 512 个数据事件、4096 行、1,048,576 个解码字符，每行最多 65,536 个字符；这些都是示例边界，不是网关限制。 工具只允许本地纯计算 `multiply`，参数必须恰好是有限数值 `a`、`b`，拒绝布尔值、额外字段和非有限结果。最多 3 次模型请求，每轮最多 4 次工具调用；达到第三轮还要求工具时直接退出。工具结果全部准备好后才发送下一轮。对带副作用的工具，仍需另行实现授权、幂等执行与恢复，不能用这个乘法示例替代事务设计。 HTTP 失败、重定向、错误内容类型、非法 JSON、截断 SSE、异常断连、业务未完成和不支持的协议扩展均返回非零退出状态。不会打印密钥、上游错误体、思考文本或不完整答案；遇到未知事件/内容块采取失败退出策略，需要显式扩展解析器才能支持它们。 Gemini 另限制每轮 256 个 parts、工具参数序列化后 16,384 个字符。细粒度参数分片、多个候选、媒体块、Live API 和异步工具调度不在示例支持范围。 ## 验证 运行 `npm run check:examples`，测试会启动本机 HTTP fixture 并执行这份脚本，检查纯文本与工具模式、完整历史和签名、结果关联、UTF-8 分块、错误与断流、无重试及第二轮失败短路。没有真实模型请求；真实渠道是否返回正确事件和保留签名，仍需部署端验证。 [Gemini 原生 API](https://ai.google.dev/api/generate-content) · [签名保留](https://ai.google.dev/gemini-api/docs/generate-content/thought-signatures) [Read in English](/en/guides/native-gemini-streaming) This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.