#!/usr/bin/env python3 """Local proxy between Handy's text clean-up and OpenCode Go, on 127.0.0.1:8082 (user unit handy-go-proxy.service). OpenCode Go refuses requests without an `x-opencode-session` header (HTTP 400, since at least 2026-10-14), and Handy can't send custom headers. Handy's Custom provider points here instead (http://127.0.0.1:8082/v1); this adds: - Authorization: opencode's own current sign-in, read with `opencode auth export opencode` (re-read when it has expired or Go answers 401, so a renewed sign-in needs no Handy restart) - x-opencode-session: one id per day, as Go's docs ask ("a stable session ID for each conversation") - User-Agent: says what it is, as Go's docs ask (not a generic library name) Handy authenticates to the proxy with a local secret (~/.config/handy-go-proxy/secret, its "API key"), and the Host header must be 127.0.0.1 or localhost, so web pages can't use it. Only POST /v1/chat/completions and GET /v1/models are passed on. Logs one line per request (model, status, time), never text or tokens. Python because the standard library has the HTTP server and client; no packages needed. Guide: https://blog.jacq.cc/guides/local-voice-dictation-kde-wayland/ (MIT licence) """ import datetime import http.server import json import os import secrets import subprocess import sys import time import urllib.error import urllib.request UPSTREAM = "https://opencode.ai/inference/go/v1" PORT = int(os.environ.get("HANDY_GO_PROXY_PORT", "8082")) SECRET_FILE = os.path.expanduser("~/.config/handy-go-proxy/secret") USER_AGENT = "Handy/0.9.8 (dictation clean-up; via handy-go-proxy)" ALLOWED_HOSTS = {f"127.0.0.1:{PORT}", f"localhost:{PORT}"} def log(msg): print(msg, file=sys.stderr, flush=True) class Token: """opencode's active sign-in for the opencode integration (OAuth access token or service-account key).""" value = None expires = 0.0 @classmethod def get(cls, force=False): if force or not cls.value or time.time() > cls.expires - 60: out = subprocess.run(["opencode", "auth", "export", "opencode"], capture_output=True, text=True, timeout=60, stdin=subprocess.DEVNULL).stdout active = [c for c in json.loads(out) if c.get("integrationID") == "opencode" and c.get("active")] if not active: raise RuntimeError("opencode isn't signed in (opencode auth login opencode)") v = active[0]["value"] cls.value = v.get("access") or v.get("key") or v.get("apiKey") or v.get("token") cls.expires = v["expires"] / 1000 if v.get("type") == "oauth" and v.get("expires") else float("inf") log(f"token read from opencode ({v.get('type')}), valid until " f"{datetime.datetime.fromtimestamp(cls.expires).isoformat(' ', 'minutes') if cls.expires != float('inf') else 'no expiry'}") return cls.value class Session: day = None id = None @classmethod def get(cls): today = datetime.date.today().isoformat() if cls.day != today: cls.day, cls.id = today, f"handy-{today}-{secrets.token_hex(8)}" return cls.id class Handler(http.server.BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "handy-go-proxy" def log_message(self, *args): # no default access log (it would show client paths only; we log our own line) pass def reply(self, status, body, ctype="application/json"): self.send_response(status) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def error(self, status, message): self.reply(status, json.dumps({"error": {"message": message}}).encode()) def allowed(self): if self.headers.get("Host") not in ALLOWED_HOSTS: self.error(403, "wrong Host") return False if not secrets.compare_digest(self.headers.get("Authorization", ""), f"Bearer {SECRET}"): self.error(401, "wrong or missing local key (see ~/.config/handy-go-proxy/secret)") return False return True def do_GET(self): if self.path == "/health": self.reply(200, b"ok\n", "text/plain") elif self.path == "/v1/models": if self.allowed(): self.forward("GET", None) else: self.error(404, "not found") def do_POST(self): body = self.rfile.read(int(self.headers.get("Content-Length") or 0)) if self.path != "/v1/chat/completions": self.error(404, "not found") elif self.allowed(): self.forward("POST", body) def forward(self, method, body): start = time.monotonic() try: model = json.loads(body).get("model", "?") if body else "-" except ValueError: model = "?" status, data, ctype = 502, b"", "application/json" for attempt in (1, 2): try: token = Token.get(force=attempt == 2) except Exception as e: # opencode missing, not signed in, bad output status, data = 502, json.dumps({"error": {"message": f"handy-go-proxy: {e}"}}).encode() break req = urllib.request.Request(UPSTREAM + self.path[len("/v1"):], data=body, method=method, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {token}", "User-Agent": USER_AGENT, "x-opencode-session": Session.get(), }) try: with urllib.request.urlopen(req, timeout=90) as resp: status, data = resp.status, resp.read() ctype = resp.headers.get("Content-Type", ctype) except urllib.error.HTTPError as e: status, data = e.code, e.read() ctype = e.headers.get("Content-Type", ctype) except Exception as e: # network, timeout status, data = 502, json.dumps({"error": {"message": f"handy-go-proxy: {e}"}}).encode() if status != 401 or attempt == 2: break log("Go answered 401: reading the token from opencode again") log(f"{method} {self.path} model={model} -> {status} in {int((time.monotonic() - start) * 1000)} ms") self.reply(status, data, ctype) if __name__ == "__main__": try: with open(SECRET_FILE) as f: SECRET = f.read().strip() except OSError as e: sys.exit(f"no local key: {e} (make it first: see the guide)") if len(SECRET) < 20: sys.exit(f"local key in {SECRET_FILE} is too short") server = http.server.ThreadingHTTPServer(("127.0.0.1", PORT), Handler) log(f"listening on http://127.0.0.1:{PORT}/v1 -> {UPSTREAM}") server.serve_forever()