#!/usr/bin/env python3 """Check a private OpenAI-compatible model API using only Python's standard library. Local protocol tests: python api-smoke-test.py --self-test Live check: python api-smoke-test.py --output api-check.json No local GPU or third-party Python package is needed on the client. """ import argparse from datetime import datetime, timezone import json import os from pathlib import Path import sys from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen class CheckError(RuntimeError): pass class Client: def __init__(self, base_url, model, api_key="dummy", timeout=600): self.base_url = base_url.rstrip("/") self.model = model self.api_key = api_key self.timeout = timeout def open(self, path, payload=None): headers = {"Authorization": "Bearer " + self.api_key} data = None if payload is not None: data = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" request = Request(self.base_url + path, data=data, headers=headers) try: return urlopen(request, timeout=self.timeout) except HTTPError as exc: # Do not dump headers or keys into logs or the saved report. raise CheckError(f"HTTP {exc.code} from {path}. Check server logs, model ID, and MODEL_API_KEY.") from exc except (URLError, TimeoutError, OSError) as exc: raise CheckError(f"Cannot reach {self.base_url}{path}. Keep the Full SSH tunnel open and wait for model startup. ({type(exc).__name__})") from exc def models(self): with self.open("/models") as response: result = json.load(response) names = [item.get("id") for item in result.get("data", [])] if self.model not in names: raise CheckError(f"Expected model {self.model!r}; /models returned {names!r}.") return names def chat(self, messages, tools=None): payload = { "model": self.model, "messages": messages, "stream": True, "temperature": 0.2, "max_tokens": 8192, "chat_template_kwargs": {"reasoning_effort": "low", "clear_thinking": True}, } if tools is not None: payload.update(tools=tools, tool_choice="auto") assistant = {"role": "assistant", "content": ""} calls = {} reasoning = "" finished = None saw_done = False with self.open("/chat/completions", payload) as response: event_lines = [] for raw_line in response: line = raw_line.decode("utf-8").rstrip("\r\n") if line.startswith(":"): continue # SSE keepalive; this is not a model token. if line.startswith("data:"): event_lines.append(line[5:].lstrip(" ")) continue if line or not event_lines: continue data = "\n".join(event_lines) event_lines = [] if data == "[DONE]": saw_done = True break chunk = json.loads(data) if "error" in chunk: raise CheckError("The server returned an error inside the stream. Check its logs.") for choice in chunk.get("choices", []): if choice.get("index", 0) != 0: continue delta = choice.get("delta", {}) if delta.get("content"): assistant["content"] += delta["content"] reasoning += delta.get("reasoning") or delta.get("reasoning_content") or "" if choice.get("finish_reason"): finished = choice["finish_reason"] for fragment in delta.get("tool_calls", []): index = fragment.get("index", 0) call = calls.setdefault(index, {"id": "", "type": "function", "function": {"name": "", "arguments": ""}}) if fragment.get("id"): call["id"] += fragment["id"] function = fragment.get("function", {}) call["function"]["name"] += function.get("name") or "" call["function"]["arguments"] += function.get("arguments") or "" if not saw_done or not finished: raise CheckError("The stream ended before a complete response. Retry after checking the tunnel and server logs.") if finished == "length": raise CheckError("The model exhausted 8192 output tokens. Confirm reasoning_effort=low reached the model; inspect server logs before increasing the budget.") if calls: assistant["tool_calls"] = [calls[index] for index in sorted(calls)] if reasoning: # Retain the model's own reasoning for its tool roundtrip, but never print # it or include it in the report. The checkpoint uses the 'reasoning' field. assistant["reasoning"] = reasoning return assistant def run_checks(client, quiet=False): results = [] def passed(label): results.append(label) if not quiet: print("PASS: " + label, flush=True) client.models() passed("Expected model is listed") messages = [{"role": "user", "content": "Remember the project code ORCHID-27 for this conversation. Reply with just READY."}] first = client.chat(messages) if not first["content"].strip(): raise CheckError("The first streamed chat contained no answer text.") passed("Streamed chat returned answer text") followup = client.chat(messages + [first, {"role": "user", "content": "What is the project code? Reply with only the code."}]) if "ORCHID-27" not in followup["content"]: raise CheckError("The follow-up did not retain the project code from the supplied conversation.") passed("Conversation follow-up retained context") tools = [{"type": "function", "function": { "name": "add_numbers", "description": "Add two integers and return their sum.", "parameters": {"type": "object", "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}}, "required": ["a", "b"], "additionalProperties": False}, }}] task = [{"role": "user", "content": "Use the add_numbers tool to calculate 19 + 23. You must call the tool. After its result arrives, reply with only the total."}] assistant = client.chat(task, tools) calls = assistant.get("tool_calls", []) if len(calls) != 1 or calls[0]["function"]["name"] != "add_numbers" or not calls[0]["id"]: raise CheckError("Expected one structured add_numbers tool call. Check --enable-auto-tool-choice and --tool-call-parser glm47.") call = calls[0] values = json.loads(call["function"]["arguments"]) if set(values) != {"a", "b"} or any(type(value) is not int for value in values.values()) or sorted(values.values()) != [19, 23]: raise CheckError("The tool call did not contain the expected integer arguments 19 and 23.") passed("Streaming tool-call fragments formed valid arguments") total = values["a"] + values["b"] # This tool is actually executed on the client. tool_message = {"role": "tool", "tool_call_id": call["id"], "content": json.dumps({"sum": total})} final = client.chat(task + [assistant, tool_message], tools) if final.get("tool_calls") or final["content"].strip().rstrip(".") != "42": raise CheckError("The model did not complete the tool roundtrip with the expected total 42.") passed("Executed tool result was accepted; final answer is 42") return results def self_test(): """Use a local fake server to exercise fragmented SSE and the real tool loop.""" from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from threading import Thread requests = [] class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_GET(self): assert self.path == "/v1/models" self.send_response(200) self.end_headers() self.wfile.write(json.dumps({"data": [{"id": "mock-glm"}]}).encode()) def do_POST(self): assert self.path == "/v1/chat/completions" assert self.headers["Authorization"] == "Bearer dummy" body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) assert body["stream"] and body["chat_template_kwargs"]["reasoning_effort"] == "low" requests.append(body) turn = len(requests) if turn == 1: deltas = [{"content": "REA"}, {"content": "DY"}] finish = "stop" elif turn == 2: assert body["messages"][-2]["content"] == "READY" deltas = [{"content": "ORCHID-"}, {"content": "27"}] finish = "stop" elif turn == 3: deltas = [ {"reasoning": "private test reasoning"}, {"tool_calls": [{"index": 0, "id": "call_mock", "function": {"name": "add_numbers", "arguments": '{"a":19,'}}]}, {"tool_calls": [{"index": 0, "function": {"arguments": '"b":23}'}}]}, ] finish = "tool_calls" else: assert turn == 4 assert json.loads(body["messages"][-1]["content"]) == {"sum": 42} assert body["messages"][-1]["tool_call_id"] == "call_mock" assert body["messages"][-2]["reasoning"] == "private test reasoning" deltas, finish = [{"content": "42"}], "stop" self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.end_headers() self.wfile.write(b": keepalive\n\n") for delta in deltas: self.wfile.write(("data: " + json.dumps({"choices": [{"index": 0, "delta": delta}]}) + "\n\n").encode()) self.wfile.write(("data: " + json.dumps({"choices": [{"delta": {}, "finish_reason": finish}]}) + "\n\n").encode()) self.wfile.write(b"data: [DONE]\n\n") server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = Thread(target=server.serve_forever, daemon=True) thread.start() try: checks = run_checks(Client(f"http://127.0.0.1:{server.server_port}/v1", "mock-glm"), quiet=True) assert len(requests) == 4 and len(checks) == 5 finally: server.shutdown() server.server_close() thread.join(timeout=2) print("PASS: local mock protocol checks (SSE keepalive, fragmented content, context, fragmented tool arguments, executed tool roundtrip).") print("This checks the client only; no real model or GPU was tested.") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--base-url", default="http://127.0.0.1:8000/v1") parser.add_argument("--model", default="glm-5.3-uncensored") parser.add_argument("--api-key", default=os.environ.get("MODEL_API_KEY", "dummy"), help="Prefer MODEL_API_KEY in the environment; defaults to dummy for tunnel-only mode.") parser.add_argument("--timeout", type=float, default=600, help="Socket timeout in seconds (default: 600).") parser.add_argument("--output", default="api-check.json", help="Local JSON report path (default: api-check.json).") parser.add_argument("--self-test", action="store_true", help="Run local mock tests without contacting a model.") args = parser.parse_args() if args.self_test: self_test() return 0 report = {"checked_at": datetime.now(timezone.utc).isoformat(), "base_url": args.base_url, "model": args.model, "passed": False} try: report["checks"] = run_checks(Client(args.base_url, args.model, args.api_key, args.timeout)) report["passed"] = True report["note"] = "API behavior checked on this endpoint only. This does not establish uncensored behavior, task quality, speed, or sustained uptime." except (CheckError, ValueError, KeyError, TimeoutError, OSError) as exc: report["error"] = str(exc) print("FAIL: " + str(exc), file=sys.stderr) output = Path(args.output).expanduser() output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") print("Saved API check: " + str(output.resolve())) return 0 if report["passed"] else 1 if __name__ == "__main__": raise SystemExit(main())