> ## 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. # Native Claude SSE and tool loops > Consume native events, validate calls, preserve reasoning state, and return tool results. This page extends [native Claude integration](/en/guides/native-claude.md) 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-claude-stream.py API_NATIVE_MODE=tools python3 examples/native-claude-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 /v1/messages`, `stream: true`, `x-api-key`, and `anthropic-version` | | Message start | Validate the assistant message and empty content in `message_start` | | Blocks | Associate start / delta / stop by `index`; reject duplicates, invalid indices, and deltas after closure | | Arguments | Accumulate `input_json_delta.partial_json` and parse JSON at block closure | | Completion | Require closed blocks, `message_delta.stop_reason`, `message_stop`, and clean EOF | | Results | Replay all assistant content; return all `tool_result` blocks together in the next user message, linked by `tool_use_id` | Only `tool_use` enters tool execution; only `end_turn` accepts a final answer without tools. Token-limit stops, refusals, missing completion, and stream errors fail. The fixed `max_tokens: 1024` is a local budget requiring model support; complex requests can still truncate. The example accumulates thinking and signature deltas into the original thinking block, preserving redacted data and block order for tool-round replay. Only text is displayed. Thinking is not explicitly enabled, but supported thinking blocks returned by a channel are retained. Missing signatures fail rather than being synthesized or modified. ## Complete Python example ```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.") ``` ## 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. Claude additionally limits each turn to 16 content blocks and each argument string to 16,384 characters. Server tools, citation deltas, and unlisted blocks 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. [Claude streaming events](https://platform.claude.com/docs/en/build-with-claude/streaming) · [Tool results](https://platform.claude.com/docs/en/agents-and-tools/tool-use/handle-tool-calls) [阅读中文](/zh/guides/native-claude-streaming.md) This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.