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