handy-go-proxy.py

Files / handy-go-proxy / handy-go-proxy.py

Files in handy-go-proxy/

• Directoryhandy-go-proxy/

  • handy-go-proxy.py 6.8 KB
  • handy-go-proxy.service 672 bytes
  • LICENSE 1.0 KB

DownloadRawPython · 159 lines · 6.8 KB

From Dictation on Linux with Handy, and tidy text with OpenCode Go. MIT licence: LICENSE.

handy-go-proxy/handy-go-proxy.py

#!/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()