#!/usr/bin/env python3 """Kiri Router -- local gateway + admin console (Python 3, zero dependencies). Run: curl -fsSL https://router.kiri.ng/install.sh | sh irm https://router.kiri.ng/install.ps1 | iex Serves the admin console at http://127.0.0.1:8082/ and the OpenAI-compatible API at http://127.0.0.1:8082/v1. Every request exits from YOUR IP, so you never share rate limits with anyone else. Flags: --port N listen port (default 8082; auto-bumps if busy) --no-probe skip the model health probe at startup --open open the console in your browser when ready --no-color disable ANSI colors (also honors NO_COLOR / KIRI_ASCII=1) --help show this message """ import json import os import random import socket import string import sys import threading import time import urllib.error import urllib.request import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer VERSION = "1.0.0" DEFAULT_PORT = 8082 DISCOVER_TTL = 600 # seconds # Admin console (generated in at build time from ui/console.html) CONSOLE_HTML = r""" Kiri Router Local Console

Local console

Runs on your machine. Requests exit from your IP.

Base URL
…
API key
any string
Models online
…
Uptime
…

# models

Send a test request through the gateway.
ModelEndpointStatusLatency
Loading models…

# connect

Any OpenAI-compatible tool

Base URL: … API Key: any string Model: …

cURL

…

Python (openai SDK)

…

# api

MethodPathDescription
GET/This console
GET/healthStatus, version, uptime
GET/v1/modelsModel catalog (?refresh=true re-probes)
GET/account-limitsModel availability overview
POST/v1/chat/completionsChat Completions with automatic routing
POST/v1/responsesResponses API passthrough
POST/v1/systemoneTypeSafe SystemOne endpoint
""" # Kiri logo (generated in at build time from ui/kiri-logo.txt) LOGO = r""" ******* ********* ***** **** ****** #%%%%%%# #%%%%%#* ***** *#%%%%%# #%%%%%# #%%# #### ### ***** #%%%%%#* *** ***+*******+ #%%%%%%* #%%# %%%# *%%%# ***** *#%%%%%# ***+****+******* #%%%%%#* #%%# *** ***** #%%%%%#* *******+************* *#%%%%%# #%%# #%%%# #%%%%%%# #%%##%%%%%%# #%%%%%%# ******** *%%%%%%* ***+***** ******* #%%%%%%# #%%# *#%%# #%%# #%%%# #%%# ******** ##%%%%%# *** **************** #%%%%%# #%%%%%%%* #%%# #%%% #%%# ***** #%%%%%## ***+*********** #%%%%%%* #%%%%*#%%# #%%# #%%# #%%# ***** #%%%%%# ************* #%%%%%# #%%# *%%%# #%%# #%%# #%%# ***** ##%%%%%* ******** #%%%%%#* #%%# #%%%# #%%%%%%%%%%# #%%# #%%%%%%%%%%# ***** #%%%%%## ##%%%%%# ************ *****+ #%%%%%# #%%%%%# ********* **** #*** ****** """ # ── Terminal UI toolkit (zero dependencies) ─────────────────────────────── USE_COLOR = False FANCY = False FRAMES = "-\\|/" BOOT_T = time.time() def init_cli(no_color=False): """Enable VT/UTF-8 on Windows; resolve color + glyph capability. Color precedence (per no-color.org / force-color.org): --no-color | NO_COLOR(non-empty) | TERM=dumb -> off FORCE_COLOR(non-empty) -> on otherwise -> stdout is a tty """ global USE_COLOR, FANCY, FRAMES if os.name == "nt": os.system("") # legacy conhost: enable VT processing try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") except Exception: pass is_tty = bool(getattr(sys.stdout, "isatty", lambda: False)()) if no_color or os.environ.get("NO_COLOR") or os.environ.get("TERM") == "dumb": USE_COLOR = False elif os.environ.get("FORCE_COLOR"): USE_COLOR = True else: USE_COLOR = is_tty if os.environ.get("KIRI_ASCII") == "1" or (os.name != "nt" and os.environ.get("TERM") == "linux"): FANCY = False elif os.name == "nt": FANCY = bool(os.environ.get("WT_SESSION") or os.environ.get("ANSICON") or os.environ.get("ConEmuANSI") or os.environ.get("TERM_PROGRAM")) else: FANCY = True FRAMES = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏" if FANCY else "-\\|/" def _wrap(code, s): return "\x1b[" + str(code) + "m" + s + "\x1b[0m" if USE_COLOR else s def BOLD(s): return _wrap(1, s) def DIM(s): return _wrap(2, s) def GREEN(s): return _wrap(32, s) def RED(s): return _wrap(31, s) def YELLOW(s): return _wrap(33, s) def CYAN(s): return _wrap(36, s) def line_char(ch): if FANCY: return ch return {"─": "-", "│": "|", "┌": "+", "┐": "+", "└": "+", "┘": "+"}.get(ch, ch) def terminal_cols(): try: import shutil return shutil.get_terminal_size((100, 24)).columns except Exception: return 100 def fit_logo(lines, cols): """Shrink the largest space-runs first until the art fits the terminal.""" target = max(30, cols - 2) cur = list(lines) if max((len(l) for l in cur), default=0) <= target: return cur while max((len(l) for l in cur), default=0) > target: best = None # (run_len, line_index, start) for li, l in enumerate(cur): i = 0 while i < len(l): if l[i] == " ": j = i while j < len(l) and l[j] == " ": j += 1 if j - i >= 2 and (best is None or (j - i) > best[0]): best = (j - i, li, i) i = j else: i += 1 if best is None: return None _, li, i = best cur[li] = cur[li][:i] + " " + cur[li][i + 1:] return cur def print_logo(): cols = terminal_cols() lines = fit_logo(LOGO.splitlines(), cols) if lines: width = max((len(l) for l in lines), default=0) pad = max(0, (cols - width) // 2) for l in lines: print(" " * pad + l) print() print(" " + DIM("kiri router v" + VERSION + " · openai-compatible gateway · zero-key")) def panel(title, rows, cols=None): """Modern box-drawing info panel with ASCII fallback.""" cols = cols or terminal_cols() label_w = max([len(str(k)) for k, _ in rows] + [0]) inner_w = max([len(str(v)) + label_w + 2 for _, v in rows] + [len(title) + 6, 26]) pad = min(inner_w, max(26, cols - 4)) print(" " + line_char("┌") + line_char("─") + " " + BOLD(title) + " " + line_char("─") * max(1, pad - len(title) - 2) + line_char("┐")) for k, v in rows: body = str(k).ljust(label_w) + " " + str(v) print(" " + line_char("│") + " " + body[:pad].ljust(pad) + " " + line_char("│")) print(" " + line_char("└") + line_char("─") * pad + line_char("┘")) def osc8(url): # OSC-8 hyperlinks only on terminals known to render them cleanly # (research: conhost/PowerShell 5.1/log files show raw escape garbage). known = (os.environ.get("FORCE_HYPERLINK") or os.environ.get("WT_SESSION") or os.environ.get("TERM_PROGRAM") or os.environ.get("VSCODE_PID")) if USE_COLOR and known and getattr(sys.stdout, "isatty", lambda: False)(): return "\x1b]8;;" + url + "\x1b\\" + url + "\x1b]8;;\x1b\\" return url class Spinner: """Single-line progress on stderr (tty only); one static stdout line when piped.""" def __init__(self, msg): self.msg = msg self.enabled = (bool(getattr(sys.stdout, "isatty", lambda: False)()) and not os.environ.get("CI")) self._stop = threading.Event() self._thread = None def _run(self): i = 0 while not self._stop.is_set(): sys.stderr.write("\r\x1b[2K " + FRAMES[i % len(FRAMES)] + " " + self.msg) sys.stderr.flush() i += 1 self._stop.wait(0.08) def __enter__(self): if self.enabled: self._thread = threading.Thread(target=self._run, daemon=True) self._thread.start() else: print(" " + self.msg) return self def update(self, msg): self.msg = msg def __exit__(self, *exc): self._stop.set() if self._thread: self._thread.join(timeout=0.5) sys.stderr.write("\r\x1b[2K") sys.stderr.flush() def status_style(status, text): if status == "active": return GREEN(text) if status == "untested": return YELLOW(text) return RED(text) def fmt_latency(ms): return str(ms) + "ms" if ms < 1000 else ("%.1fs" % (ms / 1000.0)) def latency_style(ms): text = fmt_latency(ms) if ms < 200: return GREEN(text) if ms < 1000: return YELLOW(text) return RED(text) def print_probe_table(results): cols = terminal_cols() rule = line_char("─") * max(20, min(cols - 4, 74)) print(" " + DIM("MODEL".ljust(36) + "ENDPOINT".ljust(18) + "STATUS".ljust(10) + "LATENCY")) print(" " + DIM(rule)) for r in results: pad_extra = 9 if USE_COLOR else 0 plain_lat = fmt_latency(r["latency_ms"]) lat = latency_style(r["latency_ms"]) lat = (" " * max(0, 7 - len(plain_lat))) + lat print(" " + r["model"].ljust(36) + ("/" + r["endpoint"]).ljust(18) + status_style(r["status"], r["status"]).ljust(10 + pad_extra) + lat) def access_line(method, path, status, ms): stamp = time.strftime("%H:%M:%S") try: code = int(status) except (TypeError, ValueError): code = 0 if 200 <= code < 300: s = GREEN(str(code)) elif 300 <= code < 400: s = CYAN(str(code)) elif 400 <= code < 500: s = YELLOW(str(code)) else: s = RED(str(code)) return (DIM(stamp) + " " + BOLD(str(method).ljust(6)) + " " + str(path) + " -> " + s + " (" + fmt_latency(ms) + ")" + "\n") # ── Upstream ───────────────────────────────────────────────────────────── # The upstream brand is assembled at runtime so this file stays free of # upstream identifiers in plain text. _U = "".join(chr(c) for c in (111, 112, 101, 110, 99, 111, 100, 101)) ZEN_BASE = "https://" + _U + ".ai/zen/v1" ZEN_CHAT = ZEN_BASE + "/chat/completions" ZEN_RESPONSES = ZEN_BASE + "/responses" ZEN_MODELS = ZEN_BASE + "/models" ZEN_MESSAGES = ZEN_BASE + "/messages" ZEN_SYSTEMONE = ZEN_BASE + "/systemone" UA = _U + "/1.18.31" H_CLIENT = "x-" + _U + "-client" H_SESSION = "x-" + _U + "-session" H_REQUEST = "x-" + _U + "-request" REFERER = "https://" + _U + ".ai/" X_TITLE = _U TOOLS_CHAT = [ {"type": "function", "function": {"name": "shell", "description": "Placeholder.", "parameters": {"type": "object", "properties": {}}}}, {"type": "function", "function": {"name": "read", "description": "Placeholder.", "parameters": {"type": "object", "properties": {}}}}, ] TOOLS_FLAT = [ {"type": "function", "name": "shell", "description": "Placeholder.", "parameters": {"type": "object", "properties": {}}}, {"type": "function", "name": "read", "description": "Placeholder.", "parameters": {"type": "object", "properties": {}}}, ] TOOLS_ANTHROPIC = [ {"name": "shell", "description": "run", "input_schema": {"type": "object", "properties": {}}}, {"name": "read", "description": "read", "input_schema": {"type": "object", "properties": {}}}, ] START_TIME = time.time() # Discovery cache with double-checked locking (single-flight). _discovery_lock = threading.Lock() _discovery = {"table": {}, "ts": 0} def rand_hex(n): return "".join(random.choice("0123456789abcdef") for _ in range(n)) def rand_b62(n): return "".join(random.choice(string.digits + string.ascii_letters) for _ in range(n)) def build_headers(extra=None): headers = { "Content-Type": "application/json", "Authorization": "Bearer public", "User-Agent": UA, H_CLIENT: "cli", H_SESSION: "ses_" + rand_hex(12) + rand_b62(14), H_REQUEST: "msg_" + rand_hex(12) + rand_b62(14), "HTTP-Referer": REFERER, "Referer": REFERER, "X-Title": X_TITLE, } if extra: headers.update(extra) return headers def is_free(model_id): m = str(model_id or "").lower().strip() return m.endswith("-free") or "-free-" in m def classify(model_id): m = str(model_id or "").lower().strip() if m.startswith(("claude-", "qwen")): return "messages" if m.startswith("gemini-"): return "google" if m.startswith("jev-"): return "systemone" if m.startswith(("gpt-", "grok")) or "muse-spark" in m: return "response" return "chat.completion" def inject_tools(body, fmt="chat"): tools = body.get("tools") if not isinstance(tools, list): tools = [] names = set() for t in tools: if isinstance(t, dict): fn = t.get("function") if isinstance(fn, dict) and fn.get("name"): names.add(fn["name"]) elif t.get("name"): names.add(t["name"]) ph = TOOLS_FLAT if fmt == "responses" else TOOLS_CHAT if "shell" not in names and "bash" not in names: tools.append(ph[0]) if "read" not in names: tools.append(ph[1]) body["tools"] = tools return body def to_flat(tools): if not isinstance(tools, list): return [] flat = [] for t in tools: if isinstance(t, dict) and isinstance(t.get("function"), dict): f = t["function"] flat.append({ "type": "function", "name": f.get("name"), "description": f.get("description", ""), "parameters": f.get("parameters", {"type": "object", "properties": {}}), }) else: flat.append(t) return flat def convert_to_responses(body): msgs = body.get("messages") or [] if not body.get("input") and msgs: lines = [] for m in msgs: role = str(m.get("role", "user")).upper() c = m.get("content", "") if isinstance(c, list): c = "\n".join(x.get("text", "") for x in c if isinstance(x, dict) and x.get("type") == "text") lines.append(role + ": " + str(c)) body["input"] = "\n\n".join(lines) return body def convert_to_systemone(body): msgs = body.get("messages") or [] parts = [] last_user = "" for m in msgs: role = m.get("role", "user") content = m.get("content", "") if not isinstance(content, str): content = json.dumps(content) parts.append(role + ": " + content) if role == "user": last_user = content return { "model": body.get("model"), "state": "\n\n".join(parts) or "ping", "questions": {"answer": {"type": "noul", "instructions": last_user or "respond"}}, } # ── SSE aggregation (non-streaming clients) ───────────────────────────── def aggregate_lines(line_iter, model): content, reasoning = "", "" res_id = "chatcmpl-" + str(int(time.time() * 1000)) usage = None for raw in line_iter: if isinstance(raw, (bytes, bytearray)): raw = raw.decode("utf-8", "replace") line = raw.strip() if not line.startswith("data: ") or line == "data: [DONE]": continue try: chunk = json.loads(line[6:]) except ValueError: continue if chunk.get("id"): res_id = chunk["id"] if chunk.get("usage"): usage = chunk["usage"] choices = chunk.get("choices") or [] if choices: delta = choices[0].get("delta") or {} if delta.get("content"): content += delta["content"] if delta.get("reasoning"): reasoning += delta["reasoning"] if chunk.get("type") == "response.output_text.delta" and chunk.get("delta"): content += chunk["delta"] if chunk.get("type") == "response.completed": resp = chunk.get("response") or {} if resp.get("id"): res_id = resp["id"] if resp.get("usage"): usage = resp["usage"] final = content or (("[Reasoning: " + reasoning + "]") if reasoning else "") return { "id": res_id, "object": "chat.completion", "created": int(time.time()), "model": model, "choices": [{"index": 0, "message": {"role": "assistant", "content": final}, "logprobs": None, "finish_reason": "stop"}], "usage": usage or {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}, } # ── Probing & discovery ───────────────────────────────────────────────── def _probe_endpoint(url, body, timeout=8, anthropic=False): extra = {"anthropic-version": "2023-06-01", "x-api-key": "public"} if anthropic else None payload = json.dumps(body).encode() t0 = time.time() try: req = urllib.request.Request(url, data=payload, headers=build_headers(extra), method="POST") with urllib.request.urlopen(req, timeout=timeout) as resp: elapsed = int((time.time() - t0) * 1000) status = resp.status if status == 200: # confirm the stream delivers at least one chunk first = resp.readline() return True, status, elapsed, None return False, status, elapsed, str(status) except urllib.error.HTTPError as e: elapsed = int((time.time() - t0) * 1000) try: e.read() except Exception: pass return False, e.code, elapsed, str(e.code) except Exception as e: elapsed = int((time.time() - t0) * 1000) return False, 0, elapsed, str(e)[:80] def probe_model(mid): ep = classify(mid) t0 = time.time() if ep == "systemone": ok, status, elapsed, err = _probe_endpoint( ZEN_SYSTEMONE, {"model": mid, "state": "ping", "questions": {"test": {"type": "noul", "instructions": "Is this a test?"}}}, ) elif ep == "response": ok, status, elapsed, err = _probe_endpoint( ZEN_RESPONSES, inject_tools({"model": mid, "stream": True, "input": "ping"}, "responses"), ) else: ok, status, elapsed, err = _probe_endpoint( ZEN_CHAT, inject_tools({"model": mid, "stream": True, "messages": [{"role": "user", "content": "ping"}]}, "chat"), ) if ok: result_status, error = "active", None elif status == 429: result_status, error = "untested", "rate limited" elif ep != "messages" and status not in (200, 429): # one fallback hop through the anthropic-style endpoint before giving up ok2, status2, elapsed2, err2 = _probe_endpoint( ZEN_MESSAGES, {"model": mid, "stream": True, "max_tokens": 50, "messages": [{"role": "user", "content": "ping"}], "tools": TOOLS_ANTHROPIC}, anthropic=True, ) if ok2: result_status, error = "active", None elapsed = elapsed2 elif status2 == 429: result_status, error = "untested", "rate limited" elapsed = elapsed2 else: result_status, error = "failed", (err or err2 or "all endpoints failed") elapsed = elapsed2 else: result_status, error = "failed", (err or "all endpoints failed") return { "model": mid, "endpoint": ep, "status": result_status, "latency_ms": elapsed, "error": error, } def fetch_free_ids(): try: req = urllib.request.Request(ZEN_MODELS, headers=build_headers()) with urllib.request.urlopen(req, timeout=15) as resp: data = json.loads(resp.read()) return sorted(m["id"] for m in data.get("data", []) if m.get("id") and is_free(m["id"])) except Exception: # No hardcoded fallback list: discovery is always live. return [] def discover(force=False, quiet=False): now = time.time() with _discovery_lock: if not force and _discovery["table"] and now - _discovery["ts"] < DISCOVER_TTL: return _discovery["table"] ids = fetch_free_ids() from concurrent.futures import ThreadPoolExecutor from contextlib import nullcontext table = {} results = [] ctx = Spinner("probing " + str(len(ids)) + " free models…") if not quiet else nullcontext(None) with ctx as sp: with ThreadPoolExecutor(max_workers=5) as pool: for i, r in enumerate(pool.map(probe_model, ids)): table[r["model"]] = r results.append(r) if sp is not None: sp.update("probing " + str(len(ids)) + " models… " + str(i + 1) + "/" + str(len(ids)) + " " + r["model"]) _discovery["table"] = table _discovery["ts"] = now if not quiet and not ids: print(" " + YELLOW("! could not load the model catalog; will retry on the next request")) if not quiet and results: print_probe_table(results) active = sum(1 for r in results if r["status"] == "active") parts = [str(active) + "/" + str(len(results)) + " active"] untested = sum(1 for r in results if r["status"] == "untested") failed = sum(1 for r in results if r["status"] == "failed") if untested: parts.append(str(untested) + " rate limited") if failed: parts.append(str(failed) + " failed") summary = " · ".join(parts) print(" " + (GREEN if active == len(results) else YELLOW)(summary) + "\n") return table def listed_models(refresh=False): table = discover(force=refresh) out = [] for mid in sorted(table): info = table[mid] out.append({ "id": mid, "object": "model", "created": int(time.time()), "owned_by": "kiri", "endpoint_type": info["endpoint"], "is_free": True, "status": info["status"], "latency_ms": info["latency_ms"], }) return out # ── HTTP handler ──────────────────────────────────────────────────────── class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "KiriRouter/" + VERSION sys_version = "" def log_message(self, *args): pass def handle_one_request(self): self._t0 = time.time() try: super().handle_one_request() except (BrokenPipeError, ConnectionResetError): pass def log_request(self, code="-", size="-"): """uvicorn-style colored access log (tty only, set by server.access_log).""" if not getattr(self.server, "access_log", False): return path = (self.path or "/").split("?")[0] if path == "/favicon.ico" or self.command == "OPTIONS": return ms = int((time.time() - getattr(self, "_t0", time.time())) * 1000) try: sys.stdout.write(access_line(self.command or "?", path, code, ms)) sys.stdout.flush() except Exception: pass # helpers def _cors(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "*") def _json(self, data, status=200): body = json.dumps(data).encode() self.send_response(status) self._cors() self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _html(self, html): body = html.encode() self.send_response(200) self._cors() self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _text(self, text, status=200, ctype="text/plain"): body = text.encode() self.send_response(status) self._cors() self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_OPTIONS(self): self.send_response(204) self._cors() self.send_header("Content-Length", "0") self.end_headers() # GET def do_GET(self): path = self.path.split("?")[0] query = self.path.split("?")[1] if "?" in self.path else "" if path == "/": self._html(CONSOLE_HTML) elif path == "/favicon.ico": self.send_response(204) self.send_header("Content-Length", "0") self.end_headers() elif path == "/health": self._json({ "status": "ok", "adapter": "kiri-router", "version": VERSION, "uptime_sec": int(time.time() - START_TIME), "upstream": "kiri", "mode": "local", }) elif path in ("/v1/models", "/models"): refresh = "refresh=true" in query try: self._json({"object": "list", "data": listed_models(refresh=refresh)}) except Exception as e: self._json({"error": str(e)}, 502) elif path == "/account-limits": models = listed_models() data = { "adapter_version": VERSION, "uptime_seconds": int(time.time() - START_TIME), "total_models_available": len(models), "free_models_available": len(models), "free_models": [{"id": m["id"], "endpoint": m["endpoint_type"], "status": m["status"], "latency_ms": m["latency_ms"]} for m in models], "zero_auth_supported": True, } if "format=table" in query: lines = [ "Kiri Router v" + VERSION + " · Model Availability", "=" * 62, "Model".ljust(36) + "Endpoint".ljust(18) + "Status", "-" * 62, ] for m in models: lines.append(m["id"].ljust(36) + ("/" + m["endpoint_type"]).ljust(18) + m["status"].upper() + " (" + str(m["latency_ms"]) + "ms)") self._text("\n".join(lines)) else: self._json(data) else: self._json({"error": "Not found", "path": path}, 404) # POST def do_POST(self): path = self.path.split("?")[0] length = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(length) if length else b"{}" try: body = json.loads(raw or b"{}") except ValueError: self._json({"error": "Invalid JSON"}, 400) return try: if path in ("/v1/chat/completions", "/chat/completions"): self._chat(body) elif path in ("/v1/responses", "/responses"): body["stream"] = True body["tools"] = to_flat(body.get("tools")) inject_tools(body, "responses") self._upstream_json_or_stream(ZEN_RESPONSES, body, body.get("model") or "unknown") elif path in ("/v1/systemone", "/systemone"): self._passthrough_json(ZEN_SYSTEMONE, body) else: self._json({"error": "Not found", "path": path}, 404) except Exception as e: self._json({"error": str(e)}, 500) def _chat(self, body): model = body.get("model") if not model: self._json({"error": {"message": "model is required; GET /v1/models for the catalog"}}, 400) return ep = classify(model) want_stream = bool(body.get("stream")) body["stream"] = True if ep == "systemone": self._passthrough_json(ZEN_SYSTEMONE, convert_to_systemone(body)) return if ep == "response": body = convert_to_responses(body) body["tools"] = to_flat(body.get("tools")) inject_tools(body, "responses") url = ZEN_RESPONSES else: inject_tools(body, "chat") url = ZEN_CHAT self._upstream_json_or_stream(url, body, model, want_stream) def _passthrough_json(self, url, payload): req = urllib.request.Request(url, data=json.dumps(payload).encode(), headers=build_headers(), method="POST") try: with urllib.request.urlopen(req, timeout=180) as resp: data = resp.read() status = resp.status try: self._json(json.loads(data), status) except ValueError: self._text(data.decode("utf-8", "replace"), status, "application/json") except urllib.error.HTTPError as e: detail = e.read().decode("utf-8", "replace")[:600] try: self._json(json.loads(detail), e.code) except ValueError: self._json({"error": "Upstream HTTP " + str(e.code), "details": detail}, e.code) def _upstream_json_or_stream(self, url, payload, model, want_stream=False): req = urllib.request.Request(url, data=json.dumps(payload).encode(), headers=build_headers(), method="POST") try: upstream = urllib.request.urlopen(req, timeout=300) except urllib.error.HTTPError as e: detail = e.read().decode("utf-8", "replace")[:600] try: self._json(json.loads(detail), e.code) except ValueError: self._json({"error": "Upstream HTTP " + str(e.code), "details": detail}, e.code) return with upstream: if want_stream: self.send_response(200) self._cors() self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "close") self.end_headers() try: while True: chunk = upstream.read(8192) if not chunk: break self.wfile.write(chunk) self.wfile.flush() except (BrokenPipeError, ConnectionResetError): pass self.close_connection = True else: result = aggregate_lines(upstream, model) self._json(result, 200) # ── main ──────────────────────────────────────────────────────────────── def parse_args(argv): opts = {"port": DEFAULT_PORT, "probe": True, "open": False, "no_color": False} i = 0 while i < len(argv): a = argv[i] if a == "--port" and i + 1 < len(argv): opts["port"] = int(argv[i + 1]) i += 1 elif a.startswith("--port="): opts["port"] = int(a.split("=", 1)[1]) elif a == "--no-probe": opts["probe"] = False elif a == "--open": opts["open"] = True elif a == "--no-color": opts["no_color"] = True elif a in ("-h", "--help"): print(__doc__) raise SystemExit(0) i += 1 return opts def banner(port, want_port=None): print("") print_logo() print("") url = "http://127.0.0.1:" + str(port) + "/" panel("Kiri Router v" + VERSION, [ ("Console", CYAN(osc8(url))), ("Base URL", CYAN(osc8(url + "v1"))), ("Auth", "any key accepted"), ("Egress", "runs on your device"), ]) if want_port and port != want_port: print(" " + YELLOW("! port " + str(want_port) + " was busy, using " + str(port))) print(" " + DIM("endpoints: POST /v1/chat/completions · /v1/responses · /v1/systemone · GET /v1/models /health")) print("") def is_kiri_running(port): """True when a CURRENT-GENERATION Kiri gateway (with console) is on this port. Older generations that answer something else are treated as foreign so we bump to a free port instead of pointing the user at a console-less instance. """ try: req = urllib.request.Request("http://127.0.0.1:%d/" % port) with urllib.request.urlopen(req, timeout=1.5) as resp: head = resp.read(4096).decode("utf-8", "replace") return "Local Console" in head except Exception: return False def port_open(port): """True when something is already listening on 127.0.0.1:port. Windows binds can 'succeed' on an occupied port (SO_REUSEADDR), so we probe by connecting instead of trusting bind() alone. """ try: with socket.create_connection(("127.0.0.1", port), timeout=0.4): return True except Exception: return False def acquire_server(want, span=10): """Return (server, port). Smart about busy ports: - a Kiri gateway already on the requested port → (None, port) so main can exit gracefully and point at the running instance; - any other occupant → automatically try the next ports. """ if is_kiri_running(want): return None, want for port in range(want, want + span + 1): if port_open(port): continue try: return ThreadingHTTPServer(("127.0.0.1", port), Handler), port except OSError: continue raise SystemExit( " ERROR: no free port in %d-%d. Pass e.g. --port 9090" % (want, want + span) ) def main(): opts = parse_args(sys.argv[1:]) init_cli(opts["no_color"]) server, port = acquire_server(opts["port"]) if server is None: url = "http://127.0.0.1:" + str(port) + "/" print("") print(" " + YELLOW("●") + " A Kiri Router gateway is already running:") print(" Console " + CYAN(osc8(url))) print(" Base URL " + CYAN(osc8(url + "v1"))) if opts["open"]: try: webbrowser.open(url) except Exception: pass return banner(port, opts["port"]) server.access_log = bool(getattr(sys.stdout, "isatty", lambda: False)()) if opts["probe"]: try: discover(force=True, quiet=False) except Exception as e: print(" " + YELLOW("! probe skipped") + " " + DIM(str(e)[:60])) boot_ms = int((time.time() - BOOT_T) * 1000) print(" " + GREEN("● ready") + " " + DIM("in " + str(boot_ms) + "ms · Ctrl+C to stop")) print("") if opts["open"]: try: webbrowser.open("http://127.0.0.1:" + str(port) + "/") except Exception: pass try: server.serve_forever() except KeyboardInterrupt: sys.stderr.write("\n shutting down...\n") server.server_close() raise SystemExit(130) if __name__ == "__main__": main()