> ## Documentation Index > Fetch the complete documentation index at: https://docs.inputapi.com/llms.txt > Use this file to discover all available pages before exploring further. # Claude 原生 SSE 与工具闭环 > 接收原生事件、验证完整调用、保留推理状态并回传工具结果。 本页扩展 [原生 Claude 入门](/zh/guides/native-claude.md)，同一个 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-claude-stream.py API_NATIVE_MODE=tools python3 examples/native-claude-stream.py ``` `API_NATIVE_MODE` 默认 `text`；`tools` 模式也允许模型直接回答。测试固定了模拟回答，真实模型措辞和是否调用工具均不保证。 ## 事件与闭环 | 步骤 | 本例处理 | | - | - | | 请求 | `POST /v1/messages`，`stream: true`，`x-api-key`，`anthropic-version` | | 消息开始 | 验证 `message_start` 的 assistant 消息与空 content | | 内容块 | 用 `index` 关联 start / delta / stop；拒绝重复、越界和关闭后的增量 | | 工具参数 | 累加 `input_json_delta.partial_json`，块关闭时解析 JSON | | 消息结束 | 所有块关闭，读取 `message_delta.stop_reason`，再等待 `message_stop` 和正常 EOF | | 回传 | 重放完整 assistant 内容；下一条 user 消息一次包含全部 `tool_result`，用 `tool_use_id` 关联 | `tool_use` 结束原因才进入工具执行；`end_turn` 才接受无工具的最终文本。`max_tokens`、拒答、缺少完成事件及流内错误都会失败。示例固定 `max_tokens: 1024`，这是需要模型支持的本地预算，不保证复杂请求不会截断。 `thinking_delta` 与 `signature_delta` 分别累加到原 thinking 块，`redacted_thinking.data` 原样保存。工具回合会重放这些块及原始顺序，但终端只输出 text。示例没有主动启用 thinking；若渠道返回已支持的思考块，也不会丢弃。缺少签名会失败，不自行生成或修改签名。 ## 完整 Python 示例 ```python theme={null} import http.client import io import json import math import os 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() history = [{"role": "user", "content": prompt}] try: for round_index in range(3): body = {"model": model, "max_tokens": 1024, "stream": True, "system": "Answer briefly; use multiply for arithmetic when available.", "messages": history} if mode == "tools": body["tools"] = [{"name": "multiply", "description": "Multiply two finite numbers.", "input_schema": schema}] request = urllib.request.Request(root_url + "/v1/messages", method="POST", headers={"Content-Type": "application/json", "Accept": "text/event-stream", "x-api-key": key, "anthropic-version": "2023-06-01"}, data=json.dumps(body, allow_nan=False).encode()) blocks, closed, arguments = [], set(), {} started, stopped, finalizing, reason = False, False, False, None with opener.open(request, timeout=60) as response: for event_name, event in read_sse(response): kind = event["type"] if event_name not in ("message", kind): raise ValueError("SSE event type mismatch.") if kind == "ping": continue if stopped or kind == "error": raise ValueError("Error or event after message_stop.") if kind == "message_start": message = event["message"] if (started or message.get("type") != "message" or message.get("role") != "assistant" or message.get("content") != [] or message.get("stop_reason") is not None): raise ValueError("Invalid message start.") started = True elif not started: raise ValueError("Missing message_start.") elif kind == "content_block_start": index, block = event["index"], event["content_block"] if finalizing or type(index) is not int or index != len(blocks) or len(blocks) >= 16: raise ValueError("Invalid block order or limit.") if block["type"] == "tool_use": if mode != "tools" or block.get("input") != {} or len(arguments) >= 4: raise ValueError("Unexpected tool block or limit.") arguments[index] = "" elif block["type"] == "text": if not isinstance(block.get("text"), str): raise ValueError("Invalid text.") elif block["type"] == "thinking": block.setdefault("signature", "") if not isinstance(block.get("thinking"), str) or not isinstance(block.get("signature"), str): raise ValueError("Invalid thinking block.") elif block["type"] == "redacted_thinking": if not isinstance(block.get("data"), str) or not block["data"]: raise ValueError("Invalid redacted block.") else: raise ValueError("Unsupported content block.") blocks.append(block) elif kind in ("content_block_delta", "content_block_stop"): index = event["index"] if (finalizing or type(index) is not int or index < 0 or index >= len(blocks) or index in closed): raise ValueError("Unknown or closed block.") block = blocks[index] if kind == "content_block_stop": if block["type"] == "tool_use": block["input"] = json.loads(arguments[index]) if block["type"] == "thinking" and not block["signature"]: raise ValueError("Missing thinking signature.") closed.add(index) else: delta = event["delta"] pair = (block["type"], delta["type"]) if pair == ("tool_use", "input_json_delta"): arguments[index] += delta["partial_json"] if len(arguments[index]) > 16_384: raise ValueError("Arguments limit.") elif pair == ("text", "text_delta"): block["text"] += delta["text"] elif pair == ("thinking", "thinking_delta"): block["thinking"] += delta["thinking"] elif pair == ("thinking", "signature_delta"): block["signature"] += delta["signature"] else: raise ValueError("Unsupported delta.") elif kind == "message_delta": if len(closed) != len(blocks): raise ValueError("Unclosed blocks.") finalizing = True value = event["delta"].get("stop_reason") if value is not None: if reason is not None: raise ValueError("Duplicate stop reason.") reason = value elif kind == "message_stop": if not finalizing or reason is None or len(closed) != len(blocks): raise ValueError("Incomplete message.") stopped = True else: raise ValueError("Unsupported event.") if not stopped: raise ValueError("Missing message_stop.") calls = [block for block in blocks if block["type"] == "tool_use"] if not calls: text = "".join(block["text"] for block in blocks if block["type"] == "text") if reason != "end_turn" or not text.strip(): raise ValueError("No complete visible answer.") print(text) break if reason != "tool_use" or round_index == 2: raise ValueError("Invalid tool stop or round limit.") results = [] for call in calls: call_id, args = call.get("id"), call.get("input") if (call.get("name") != "multiply" or not isinstance(call_id, str) or not call_id.strip() or call_id in seen_ids): raise ValueError("Unknown tool or repeated ID.") if not isinstance(args, dict) or set(args) != {"a", "b"}: raise ValueError("Invalid arguments.") 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.") seen_ids.add(call_id) results.append({"type": "tool_result", "tool_use_id": call_id, "content": json.dumps({"value": value}, allow_nan=False)}) # Preserve every assistant block, including opaque reasoning signatures. history.extend([{"role": "assistant", "content": blocks}, {"role": "user", "content": 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 Claude 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、异常断连、业务未完成和不支持的协议扩展均返回非零退出状态。不会打印密钥、上游错误体、思考文本或不完整答案；遇到未知事件/内容块采取失败退出策略，需要显式扩展解析器才能支持它们。 Claude 另限制每轮 16 个内容块、每个工具参数字符串 16,384 个字符。服务器内置工具、引用增量及其他未列出的块不在示例支持范围。 ## 验证 运行 `npm run check:examples`，测试会启动本机 HTTP fixture 并执行这份脚本，检查纯文本与工具模式、完整历史和签名、结果关联、UTF-8 分块、错误与断流、无重试及第二轮失败短路。没有真实模型请求；真实渠道是否返回正确事件和保留签名，仍需部署端验证。 [Claude 流式事件](https://platform.claude.com/docs/en/build-with-claude/streaming) · [工具结果格式](https://platform.claude.com/docs/en/agents-and-tools/tool-use/handle-tool-calls) [Read in English](/en/guides/native-claude-streaming.md) This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.