- SignalDesk2 hr ago
Original Summary
最近用kimi code 习惯以后想用嫖opencode console的免费模型,就研究了这个,搓的不好欢迎指导!下面是代码! #!/usr/bin/env python3 """Local OpenAI-compatible proxy for OpenCode Console free-tier models. OpenCode Console rejects free models (403 FreeTierError) unless the tool names look like OpenCode's native tools (lowercase read/write/edit/bash/ glob/grep). kimi-code sends capitalized names (Read/Write/Edit/Bash/Glob/ Grep). This proxy renames those six tools on the way out and renames them back on the way in, so kimi sees its own names and OpenCode sees its own. Everything else is forwarded unchanged. No third-party dependencies. """ import http.server import json import socketserver import sys import urllib.error import urllib.request LISTEN_HOST = "127.0.0.1" LISTEN_PORT = 18080 UPSTREAM = "https://opencode.ai/inference/openai/v1" TO_OC = {"Read": "read", "Write": "write", "Edit": "edit", "Bash": "bash", "Glob": "glob", "Grep": "grep"} FROM_OC = {v: k for k, v in TO_OC.items()} HOP_BY_HOP = {"host", "connection", "content-length", "transfer-encoding", "accept-encoding", "keep-alive", "te", "trailer", "upgrade"} def rewrite_request(body: dict) -> dict: for t in body.get("tools") or []: fn = t.get("function") or {} name = fn.get("name") if name in TO_OC: fn["name"] = TO_OC[name] for m in body.get("messages") or []: for tc in m.get("tool_calls") or []: fn = tc.get("function") or {} if fn.get("name") in TO_OC: fn["name"] = TO_OC[fn["name"]] return body def rewrite_choice_names(obj: dict) -> dict: for ch in obj.get("choices") or []: for key in ("message", "delta"): part = ch.get(key) or {} for tc in part.get("tool_calls") or []: fn = tc.get("function") or {} if fn.get("name") in FROM_OC: fn["name"] = FROM_OC[fn["name"]] return obj class Handler(http.server.BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt, args): sys.stderr.write("[oc-proxy] " + (fmt % args) + "\n") def do_GET(self): self._forward(None) def do_POST(self): n = int(self.headers.get("Content-Length", 0) or 0) raw = self.rfile.read(n) if n else b"" body = None if raw: try: body = rewrite_request(json.loads(raw)) except (ValueError, TypeError): body = None payload = json.dumps(body).encode() if body is not None else raw self._forward(payload) def _forward(self, payload): url = UPSTREAM + self.path[len("/v1"):] if self.path.startswith("/v1") else UPSTREAM + self.path headers = {k: v for k, v in self.headers.items() if k.lower() not in HOP_BY_HOP} req = urllib.request.Request(url, data=payload, headers=headers, method=self.command) try: resp = urllib.request.urlopen(req, timeout=600) status, rheaders, stream = resp.status, resp.headers, resp except urllib.error.HTTPError as e: status, rheaders, stream = e.code, e.headers, e ctype = (rheaders.get("Content-Type") or "").lower() self.send_response(status) for k, v in rheaders.items(): if k.lower() in HOP_BY_HOP or k.lower() == "content-length": continue self.send_header(k, v) self.send_header("Connection", "close") self.end_headers() self.close_connection = True if "text/event-stream" in ctype: for raw_line in stream: line = raw_line.decode("utf-8", "replace").rstrip("\r\n") if line.startswith("data:"): data = line[5:].strip() if data and data != "[DONE]": try: obj = rewrite_choice_names(json.loads(data)) line = "data: " + json.dumps(obj, ensure_ascii=False) except ValueError: pass self.wfile.write((line + "\n").encode("utf-8")) self.wfile.flush() else: body = stream.read() if "application/json" in ctype: try: body = json.dumps(rewrite_choice_names(json.loads(body)), ensure_ascii=False).encode() except ValueError: pass self.wfile.write(body) self.wfile.flush() class Server(socketserver.ThreadingMixIn, http.server.HTTPServer): daemon_threads = True allow_reuse_address = True if __name__ == "__main__": port = int(sys.argv[1]) if len(sys.argv) > 1 else LISTEN_PORT print(f"oc-proxy listening on http://{LISTEN_HOST}:{port}/v1 -> {UPSTREAM}", flush=True) Server((LISTEN_HOST, port), Handler).serve_forever() #!/usr/bin/env bash # Start the local OpenCode free-tier proxy on demand (idempotent). # Usage: ./start.sh start if not running # ./start.sh stop stop it # ./start.sh status show state set -euo pipefail DIR="$(cd "$(dirname "$0")" && pwd)" PORT="${OC_PROXY_PORT:-18765}" PY="${OC_PROXY_PYTHON:-python3}" LOG="$DIR/proxy.log" PIDFILE="$DIR/proxy.pid" # Only count the port as "ours" if the listener is this proxy.py. listener_pids() { lsof -nP -iTCP:"$PORT" -sTCP:LISTEN -t 2>/dev/null || true } is_listening() { local p for p in $(listener_pids); do if ps -o command= -p "$p" | grep -q "proxy.py"; then return 0; fi done return 1 } port_taken_by_other() { local p for p in $(listener_pids); do if ! ps -o command= -p "$p" | grep -q "proxy.py"; then return 0; fi done return 1 } case "${1:-start}" in stop) stopped=0 for p in $(listener_pids); do if ps -o command= -p "$p" | grep -q "proxy.py"; then kill "$p" && stopped=1 fi done rm -f "$PIDFILE" if [[ $stopped -eq 1 ]]; then echo "stopped"; else echo "not running"; fi ;; status) if is_listening; then echo "running on 127.0.0.1:$PORT"; else echo "not running"; fi ;; start|"") if is_listening; then echo "already running on 127.0.0.1:$PORT" exit 0 fi if port_taken_by_other; then echo "port $PORT is used by another process; set OC_PROXY_PORT to choose another" >&2 exit 1 fi cd "$DIR" nohup "$PY" "$DIR/proxy.py" "$PORT" >>"$LOG" 2>&1 & echo $! >"$PIDFILE" for _ in $(seq 1 20); do if is_listening; then echo "started on 127.0.0.1:$PORT (pid $(cat "$PIDFILE")), log: $LOG" exit 0 fi sleep 0.2 done echo "failed to start, see $LOG" >&2 exit 1 ;; ) echo "usage: $0 [start|stop|status]" >&2 exit 2 ;; esac 1 个帖子 - 1 位参与者 阅读完整话题
- 情报分类:商业与市场研究
- 分类依据:内容涉及商业、投资或市场动态
- 信息来源:服务器 / LINUX DO - 最新话题
- 发布时间:2026/10/16 20:43:39
- No replies yet