#!/usr/bin/env python3 # ============================================================ # NETSHOW ALIVE · SAMPLE TEXT-TURN RELAY · Python port · v1 # ------------------------------------------------------------ # The same relay as relay-node.mjs, in Python 3.8+ standard # library only — nothing to pip install. One contract, three # stacks: this file speaks the canonical text-turn contract # (full doc: /alive-site/preview/text-turn/), so your page code # is identical whichever relay you run. # # Your provider key never touches a browser page — and never # touches NetShow. The avatar page POSTs a turn here; this # relay forwards it to the model endpoint YOU configure below, # and streams the reply back as plain text — exactly the stream # the five-line feeder drinks: # # for await (const chunk of yourModelStream) feeder.delta(chunk); # # THE 1-2-3 # --------- # 1. Download this file onto your server (Python 3.8+, no pip). # # 2. Start it with YOUR model endpoint in the environment. # The key comes from YOUR shell, on YOUR machine — it is # never written into this file and never sent to the page: # # # OpenAI-compatible (OpenAI, Groq, Mistral, vLLM, LM Studio…) # MODEL_URL=https://api.openai.com/v1/chat/completions \ # MODEL_NAME=gpt-4o-mini \ # MODEL_KEY=$OPENAI_API_KEY \ # python3 relay.py # # # Anthropic # MODEL_KIND=anthropic \ # MODEL_URL=https://api.anthropic.com/v1/messages \ # MODEL_NAME=claude-sonnet-5 \ # MODEL_KEY=$ANTHROPIC_API_KEY \ # python3 relay.py # # # Local model, no key at all (Ollama shown) # MODEL_URL=http://127.0.0.1:11434/v1/chat/completions \ # MODEL_NAME=llama3.2 \ # python3 relay.py # # 3. Put /turn on the SAME ORIGIN as your page. The avatar # element and this relay both live by the same-origin law, # so proxy the route through your web server: # # # nginx — proxy_buffering off keeps the stream streaming # location = /turn { # proxy_pass http://127.0.0.1:8787/turn; # proxy_buffering off; # proxy_read_timeout 180s; # } # # Then the page side is the documented feeder, verbatim, # plus one fetch (the whole wiring is on the preview page): # # const res = await fetch('/turn', { method: 'POST', # headers: { 'content-type': 'application/json' }, # body: JSON.stringify({ user: question, history }) }); # # CONFIG (environment only — this file never holds a secret) # ---------------------------------------------------------- # MODEL_URL required. Full URL of your chat endpoint. # MODEL_NAME model id to request (default: provider default). # MODEL_KEY optional. Read from YOUR environment at boot. # Unset = no auth header (local models). # MODEL_KIND 'openai' (default) or 'anthropic' — request, # auth-header, and stream shape. # SYSTEM_PROMPT optional. When set, it is PINNED server-side # and the page's `system` field is ignored — # set this in production so visitors cannot # rewrite your agent's persona from devtools. # MAX_TOKENS reply budget (default 512). # RELAY_PORT listen port (default 8787). # RELAY_BIND listen address (default 127.0.0.1 — keep it # loopback and let your web server front it). # # SAFETY, BY CONSTRUCTION # ----------------------- # · The upstream URL comes from server config ONLY. Nothing in # the request can redirect this relay at another host. # · Turn text is untrusted DATA: validated, length-capped, and # forwarded only as JSON string values — never into headers, # URLs, shell commands, or file paths. # · No CORS headers are emitted, ever. The relay answers only # its own origin (front it with your web server, step 3). # · Upstream calls carry a 180 s socket timeout (connect + any # silent gap between chunks; Python's stdlib has no total- # duration abort — the one contract nuance of this port) and # the upstream read stops the moment the page disconnects. # In-flight turns are capped (429 beyond). # · The key is used in exactly one place (the auth header of # the upstream request) and is never logged or echoed. # · Provider error bodies are logged server-side and NEVER # forwarded to the browser. # · Oversized request bodies (> 64 KB) get the connection # closed, mirroring the canonical relay. The body must carry # a Content-Length (every browser fetch and curl does). # ============================================================ import json import os import sys import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib import error as urlerr from urllib import request as urlreq from urllib.parse import urlsplit # ---- config: environment only ------------------------------------------- KIND = (os.environ.get('MODEL_KIND') or 'openai').lower() MODEL_URL = os.environ.get('MODEL_URL') or '' MODEL_NAME = os.environ.get('MODEL_NAME') or '' MODEL_KEY = os.environ.get('MODEL_KEY') or '' # never logged, never echoed SYSTEM_PIN = os.environ.get('SYSTEM_PROMPT') or '' # server-pinned persona (production) try: _mt = int(os.environ.get('MAX_TOKENS') or 512) except ValueError: _mt = 512 MAX_TOKENS = min(4096, max(16, _mt)) PORT = int(os.environ.get('RELAY_PORT') or 8787) BIND = os.environ.get('RELAY_BIND') or '127.0.0.1' MAX_BODY = 64 * 1024 # request body cap MAX_USER = 4000 # chars per user turn MAX_SYS = 8000 # chars of system prompt MAX_HISTORY = 40 # prior turns kept MAX_INFLIGHT = 4 # concurrent upstream turns UPSTREAM_TIMEOUT_S = 180 if not MODEL_URL: print('relay.py: set MODEL_URL (see the header of this file)', file=sys.stderr) sys.exit(1) _u = urlsplit(MODEL_URL) if _u.scheme not in ('http', 'https') or not _u.netloc: print('relay.py: MODEL_URL is not a valid http(s) URL', file=sys.stderr) sys.exit(1) if KIND not in ('openai', 'anthropic'): print("relay.py: MODEL_KIND must be 'openai' or 'anthropic'", file=sys.stderr) sys.exit(1) UPSTREAM_SHOWN = '%s://%s%s' % (_u.scheme, _u.netloc, _u.path) # ---- input boundary: the page's JSON is untrusted data ------------------ def clean_turn(raw): if not isinstance(raw, dict): return {'err': 'body must be a JSON object'} user = raw.get('user') user = user.strip()[:MAX_USER] if isinstance(user, str) else '' if not user: return {'err': "field 'user' (non-empty string) is required"} if SYSTEM_PIN: system = SYSTEM_PIN[:MAX_SYS] # pinned: page cannot override else: s = raw.get('system') system = s[:MAX_SYS] if isinstance(s, str) else '' history = [] if 'history' in raw: if not isinstance(raw['history'], list): return {'err': "field 'history' must be an array"} for h in raw['history'][-MAX_HISTORY:]: if not isinstance(h, dict): return {'err': 'history entries must be objects'} if h.get('role') not in ('user', 'assistant'): return {'err': "history roles are 'user' | 'assistant'"} if not isinstance(h.get('content'), str): return {'err': 'history content must be a string'} history.append({'role': h['role'], 'content': h['content'][:MAX_SYS]}) return {'user': user, 'system': system, 'history': history} # ---- provider shapes ---------------------------------------------------- def build_request(turn): headers = {'Content-Type': 'application/json'} if KIND == 'anthropic': if MODEL_KEY: headers['x-api-key'] = MODEL_KEY headers['anthropic-version'] = '2023-06-01' body = { 'model': MODEL_NAME or 'claude-sonnet-5', 'max_tokens': MAX_TOKENS, 'stream': True, 'messages': turn['history'] + [{'role': 'user', 'content': turn['user']}], } if turn['system']: body['system'] = turn['system'] else: if MODEL_KEY: headers['Authorization'] = 'Bearer ' + MODEL_KEY msgs = [] if turn['system']: msgs.append({'role': 'system', 'content': turn['system']}) msgs += turn['history'] + [{'role': 'user', 'content': turn['user']}] body = {'max_tokens': MAX_TOKENS, 'stream': True, 'messages': msgs} if MODEL_NAME: body['model'] = MODEL_NAME return headers, json.dumps(body).encode('utf-8') def delta_text(obj): """Pull the text delta out of one SSE data payload (provider-specific).""" if KIND == 'anthropic': if obj.get('type') == 'content_block_delta': t = (obj.get('delta') or {}).get('text') return t if isinstance(t, str) else '' return '' try: t = obj['choices'][0]['delta']['content'] except (KeyError, IndexError, TypeError): return '' return t if isinstance(t, str) else '' def whole_text(obj): """Pull the full text out of a non-streamed reply (fallback path).""" if KIND == 'anthropic': c = obj.get('content') return ''.join(x.get('text') or '' for x in c) if isinstance(c, list) else '' try: t = obj['choices'][0]['message']['content'] except (KeyError, IndexError, TypeError): return '' return t if isinstance(t, str) else '' # ---- the relay ---------------------------------------------------------- _inflight = 0 _inflight_lock = threading.Lock() class Relay(BaseHTTPRequestHandler): server_version = 'alive-text-relay' sys_version = '' # advertise no runtime version def _json(self, code, obj): body = json.dumps(obj).encode('utf-8') self.send_response(code) self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): if self.path == '/healthz': return self._json(200, {'ok': True, 'kind': KIND, 'model': MODEL_NAME or 'default', 'upstream': UPSTREAM_SHOWN, 'pinnedSystem': bool(SYSTEM_PIN)}) self._json(404, {'error': 'this relay serves POST /turn and GET /healthz'}) def do_POST(self): global _inflight if self.path != '/turn': return self._json(404, {'error': 'this relay serves POST /turn and GET /healthz'}) try: size = int(self.headers.get('Content-Length') or 0) except ValueError: size = 0 if size > MAX_BODY: # mirror req.destroy(): close, no reply self.close_connection = True try: self.connection.close() except OSError: pass return try: raw = json.loads((self.rfile.read(size) if size else b'') or b'{}') except (ValueError, UnicodeDecodeError): return self._json(400, {'error': 'body is not JSON'}) turn = clean_turn(raw) if 'err' in turn: return self._json(400, {'error': turn['err']}) with _inflight_lock: if _inflight >= MAX_INFLIGHT: return self._json(429, {'error': 'relay busy — try again in a moment'}) _inflight += 1 t0 = time.time() resp = None try: headers, body = build_request(turn) try: resp = urlreq.urlopen(urlreq.Request(MODEL_URL, data=body, headers=headers), timeout=UPSTREAM_TIMEOUT_S) except urlerr.HTTPError as e: detail = '' try: detail = e.read(500).decode('utf-8', 'replace') except OSError: pass print('[turn] upstream %d in %dms: %s' % (e.code, (time.time() - t0) * 1000, detail), file=sys.stderr) # server-side only return self._json(502, {'error': 'model endpoint answered %d — see relay log' % e.code}) except (urlerr.URLError, OSError) as e: print('[turn] failed after %dms: %s' % ((time.time() - t0) * 1000, getattr(e, 'reason', e)), file=sys.stderr) return self._json(502, {'error': 'could not reach the model endpoint — see relay log'}) self.send_response(200) self.send_header('Content-Type', 'text/plain; charset=utf-8') self.send_header('Cache-Control', 'no-store') self.send_header('X-Robots-Tag', 'noindex, nofollow') self.send_header('X-Brain-Model', MODEL_NAME or 'default') self.end_headers() ctype = (resp.headers.get('Content-Type') or '').lower() sent = 0 try: if 'text/event-stream' in ctype: # SSE → plain text chunks, incrementally for raw_line in resp: line = raw_line.decode('utf-8', 'replace').strip() if not line.startswith('data:'): continue data = line[5:].strip() if not data or data == '[DONE]': continue try: obj = json.loads(data) except ValueError: continue # keep-alives etc. text = delta_text(obj) if text: self.wfile.write(text.encode('utf-8')) self.wfile.flush() sent += len(text) else: # endpoint answered with one JSON body — send the text in one piece try: obj = json.load(resp) except ValueError: obj = None text = whole_text(obj) if obj else '' if text: self.wfile.write(text.encode('utf-8')) sent = len(text) except (BrokenPipeError, ConnectionResetError): # page went away → stop paying for tokens print('[turn] client disconnected after %dms' % ((time.time() - t0) * 1000), file=sys.stderr) self.close_connection = True return print('[turn] ok %d chars in %dms' % (sent, (time.time() - t0) * 1000), flush=True) finally: if resp is not None: try: resp.close() except OSError: pass with _inflight_lock: _inflight -= 1 def log_message(self, fmt, *args): # quiet default access log; turns log above pass if __name__ == '__main__': print('relay.py listening on http://%s:%d → %s @ %s (key %s, system %s)' % (BIND, PORT, KIND, UPSTREAM_SHOWN, 'from env' if MODEL_KEY else 'NOT SET — keyless local mode', 'PINNED server-side' if SYSTEM_PIN else 'page-supplied'), flush=True) ThreadingHTTPServer((BIND, PORT), Relay).serve_forever()