"
+ err.encode("utf8", "replace") + b"")
done.set()
return
if not code:
# Landing without code (e.g. GET /) — keep waiting
self._reply(b"Waiting for OAuth redirect...
") return result["code"] = code result["state"] = st self._reply( b"" b"You can close this tab and return to the terminal.
" b"" ) done.set() def do_HEAD(self): self.send_response(200) self.end_headers() # Prefer dual-stack when available so localhost → ::1 still hits us httpd = None last_err = None for bind_host in ("127.0.0.1", "localhost", "0.0.0.0"): try: httpd = http.server.ThreadingHTTPServer((bind_host, 1455), Handler) httpd.allow_reuse_address = True print(f"[oauth] callback listening on http://{bind_host}:1455/auth/callback", flush=True) break except OSError as e: last_err = e httpd = None if httpd is None: raise RuntimeError( f"Port 1455 busy or unbindable ({last_err}). " "Stop other Codex/OpenCode login servers, or paste the redirect URL manually when prompted." ) def serve(): while not done.is_set(): httpd.timeout = 0.5 try: httpd.handle_request() except Exception as e: if not done.is_set(): print(f"[oauth] serve error: {e}", flush=True) threading.Thread(target=serve, daemon=True).start() print("\n=== Codex OAuth ===", flush=True) print("1. Browser should open. Sign in and select your workspace.", flush=True) print("2. You should land on http://localhost:1455/auth/callback?code=…", flush=True) print("3. If the browser spins or shows connection refused, copy the FULL", flush=True) print(" URL from the address bar and paste it below, then press Enter.", flush=True) print(f"\nAuth URL:\n{url}\n", flush=True) try: webbrowser.open(url) except Exception: pass def wait_for_paste(): # stdin paste races the HTTP callback; whichever finishes first wins try: line = input("Paste redirect URL (or press Enter to wait for browser):\n> ").strip() except EOFError: return if not line or done.is_set(): return code, st, err = _parse_callback_query(line) if err: result["error"] = err done.set() return if code: result["code"] = code result["state"] = st or state print("[oauth] accepted pasted redirect URL", flush=True) done.set() else: print("[oauth] no code= in that input; still waiting for browser…", flush=True) paste_thread = threading.Thread(target=wait_for_paste, daemon=True) paste_thread.start() if not done.wait(timeout): try: httpd.server_close() except Exception: pass raise TimeoutError( "OAuth timed out — callback never received.\n" "Common causes: WSL/SSH (browser is on another host), port 1455 blocked,\n" "or another app holding 1455. Re-run and paste the redirect URL, or run login\n" "on the same machine as the browser." ) try: httpd.server_close() except Exception: pass if result["error"]: raise RuntimeError(result["error"]) if not result["code"]: raise RuntimeError("OAuth finished without an authorization code") if result["state"] and result["state"] != state: raise RuntimeError( f"OAuth state mismatch (got {result['state']!r}, expected {state!r}). " "Another login may be using port 1455." ) print("[oauth] exchanging code for tokens…", flush=True) return _exchange_code(result["code"], verifier) # --------------------------------------------------------------------------- # Chat ↔ Responses translation # --------------------------------------------------------------------------- def _text(content) -> str: if content is None: return "" if isinstance(content, str): return content if isinstance(content, list): parts = [] for p in content: if isinstance(p, dict): parts.append(p.get("text") or "") else: parts.append(str(p)) return "".join(parts) return str(content) def chat_to_responses(body: dict) -> dict: instructions = None items = [] for m in body.get("messages") or []: role = m.get("role") if role == "system": t = _text(m.get("content")) if t: instructions = f"{instructions}\n\n{t}" if instructions else t elif role == "user": items.append({"role": "user", "content": _text(m.get("content"))}) elif role == "assistant": t = _text(m.get("content")) if t: items.append({"role": "assistant", "content": t}) for tc in m.get("tool_calls") or []: fn = tc.get("function") or {} items.append({ "type": "function_call", "call_id": tc.get("id") or "", "name": fn.get("name") or "", "arguments": fn.get("arguments") or "{}", }) elif role == "tool": items.append({ "type": "function_call_output", "call_id": m.get("tool_call_id") or "", "output": _text(m.get("content")), }) out: dict[str, Any] = { "model": body.get("model") or "gpt-5-codex", "input": items, "stream": True, "store": False, } if instructions: out["instructions"] = instructions tools = body.get("tools") or [] if tools: rtools = [] for t in tools: if t.get("type") == "function" and "function" in t: fn = t["function"] rtools.append({ "type": "function", "name": fn.get("name", ""), "description": fn.get("description") or "", "parameters": fn.get("parameters") or {"type": "object", "properties": {}}, }) elif t.get("type") == "function" and "name" in t: rtools.append(t) if rtools: out["tools"] = rtools out["tool_choice"] = body.get("tool_choice") or "auto" out["parallel_tool_calls"] = True return out def sse_line(obj: dict) -> bytes: return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n".encode("utf-8") def _emit_text(wfile, model: str, text: str): if not text: return wfile.write(sse_line({ "id": "chatcmpl-proxy", "object": "chat.completion.chunk", "model": model, "choices": [{"index": 0, "delta": {"content": text}, "finish_reason": None}], })) wfile.flush() def _emit_done(wfile, model: str, finish: str, usage: dict | None = None): chunk = { "id": "chatcmpl-proxy", "object": "chat.completion.chunk", "model": model, "choices": [{"index": 0, "delta": {}, "finish_reason": finish}], } if usage: chunk["usage"] = usage wfile.write(sse_line(chunk)) wfile.write(b"data: [DONE]\n\n") wfile.flush() def stream_codex_as_chat(codex_body: dict, auth: dict, wfile, model: str): headers = { "Content-Type": "application/json", "Authorization": f"Bearer {auth['access_token']}", "OpenAI-Beta": "responses=experimental", "originator": ORIGINATOR, "User-Agent": "codex_proxy/1.0", } if auth.get("account_id"): headers["ChatGPT-Account-ID"] = auth["account_id"] # Role message for agent.py (expects assistant role on first chunk sometimes) wfile.write(sse_line({ "id": "chatcmpl-proxy", "object": "chat.completion.chunk", "model": model, "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}], })) wfile.flush() print(f"[codex] → model={codex_body.get('model')} account={auth.get('account_id')!r} " f"tools={len(codex_body.get('tools') or [])} input_items={len(codex_body.get('input') or [])}", flush=True) req = urllib.request.Request(CODEX_URL, json.dumps(codex_body).encode(), headers) content = "" calls: dict[int, dict] = {} call_index: dict[str, int] = {} usage: dict = {} seen_types: list[str] = [] raw_preview: list[str] = [] try: with urllib.request.urlopen(req, timeout=600) as r: print(f"[codex] ← HTTP {getattr(r, 'status', 200)}", flush=True) for raw in r: line = raw.decode("utf8", "replace").rstrip() if len(raw_preview) < 8 and line: raw_preview.append(line[:300]) if not line: continue # Some gateways send "event: foo" then "data: {...}" if line.startswith("event:"): continue if line.startswith("data:"): payload = line[5:].lstrip() else: # bare JSON line payload = line if payload == "[DONE]": break try: ev = json.loads(payload) except json.JSONDecodeError: continue if not isinstance(ev, dict): continue et = ev.get("type") or "" if et and et not in seen_types: seen_types.append(et) print(f"[codex] event {et}", flush=True) if et in ("response.output_text.delta", "response.content_part.delta"): delta = ev.get("delta") or "" if isinstance(delta, dict): delta = delta.get("text") or delta.get("content") or "" content += delta _emit_text(wfile, model, delta) elif et == "response.output_text.done": # full text if deltas were skipped full = ev.get("text") or "" if full and not content: content = full _emit_text(wfile, model, full) elif et == "response.output_item.added": item = ev.get("item") or {} if item.get("type") == "function_call": idx = len(calls) cid = item.get("call_id") or item.get("id") or f"call_{idx}" call_index[item.get("id") or cid] = idx call_index[cid] = idx name = item.get("name") or "" calls[idx] = { "id": cid, "type": "function", "function": {"name": name, "arguments": item.get("arguments") or ""}, } wfile.write(sse_line({ "id": "chatcmpl-proxy", "object": "chat.completion.chunk", "model": model, "choices": [{"index": 0, "delta": { "tool_calls": [{"index": idx, "id": cid, "type": "function", "function": {"name": name, "arguments": ""}}], }, "finish_reason": None}], })) wfile.flush() elif item.get("type") == "message": for part in item.get("content") or []: if part.get("type") in ("output_text", "text") and part.get("text"): if not content: content += part["text"] _emit_text(wfile, model, part["text"]) elif et == "response.function_call_arguments.delta": item_id = ev.get("item_id") or "" idx = call_index.get(item_id) if idx is None and calls: idx = max(calls) if idx is not None and idx in calls: d = ev.get("delta") or "" calls[idx]["function"]["arguments"] += d wfile.write(sse_line({ "id": "chatcmpl-proxy", "object": "chat.completion.chunk", "model": model, "choices": [{"index": 0, "delta": { "tool_calls": [{"index": idx, "function": {"arguments": d}}], }, "finish_reason": None}], })) wfile.flush() elif et == "response.function_call_arguments.done": item_id = ev.get("item_id") or "" idx = call_index.get(item_id) if idx is not None and idx in calls: if ev.get("arguments") is not None: calls[idx]["function"]["arguments"] = ev["arguments"] if ev.get("name"): calls[idx]["function"]["name"] = ev["name"] elif et == "response.output_item.done": item = ev.get("item") or {} if item.get("type") == "function_call": cid = item.get("call_id") or item.get("id") or "" idx = call_index.get(item.get("id") or cid) if idx is None: idx = len(calls) call_index[item.get("id") or cid] = idx call_index[cid] = idx calls[idx] = {"id": cid, "type": "function", "function": {"name": "", "arguments": ""}} if item.get("name"): calls[idx]["function"]["name"] = item["name"] if item.get("arguments") is not None: calls[idx]["function"]["arguments"] = item["arguments"] if item.get("call_id"): calls[idx]["id"] = item["call_id"] elif item.get("type") == "message": for part in item.get("content") or []: if part.get("type") in ("output_text", "text") and part.get("text"): if part["text"] not in content: content += part["text"] _emit_text(wfile, model, part["text"]) elif et == "response.completed": resp = ev.get("response") or {} if "usage" in resp: u = resp["usage"] usage = { "prompt_tokens": u.get("input_tokens", 0), "completion_tokens": u.get("output_tokens", 0), "total_tokens": u.get("total_tokens", 0), } if not content and not calls: for item in resp.get("output") or []: if item.get("type") == "message": for part in item.get("content") or []: if part.get("type") in ("output_text", "text"): t = part.get("text") or "" content += t _emit_text(wfile, model, t) elif item.get("type") == "function_call": idx = len(calls) calls[idx] = { "id": item.get("call_id") or f"call_{idx}", "type": "function", "function": { "name": item.get("name") or "", "arguments": item.get("arguments") or "{}", }, } wfile.write(sse_line({ "id": "chatcmpl-proxy", "object": "chat.completion.chunk", "model": model, "choices": [{"index": 0, "delta": { "tool_calls": [{ "index": idx, "id": calls[idx]["id"], "type": "function", "function": dict(calls[idx]["function"]), }], }, "finish_reason": None}], })) wfile.flush() elif et in ("response.failed", "error"): err = ev.get("error") or ev msg = f"[codex_proxy upstream error] {err}" print(msg, flush=True) content += msg _emit_text(wfile, model, msg) except urllib.error.HTTPError as e: err = e.read().decode("utf8", "replace") msg = f"[codex_proxy HTTP {e.code}] {err[:2000]}" print(msg, flush=True) _emit_text(wfile, model, msg) _emit_done(wfile, model, "stop") return except Exception as e: msg = f"[codex_proxy error] {e}" print(msg, flush=True) _emit_text(wfile, model, msg) _emit_done(wfile, model, "stop") return if not content and not calls: diag = ( f"[codex_proxy] empty upstream response. events={seen_types or '(none)'} " f"preview={raw_preview[:3]}" ) print(diag, flush=True) _emit_text(wfile, model, diag) print(f"[codex] done content_len={len(content)} tool_calls={len(calls)} usage={usage}", flush=True) _emit_done(wfile, model, "tool_calls" if calls else "stop", usage or None) # --------------------------------------------------------------------------- # HTTP server # --------------------------------------------------------------------------- class Handler(http.server.BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt, *args): sys.stderr.write("[codex_proxy] " + (fmt % args) + "\n") def _read_json(self) -> dict: n = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(n) if n else b"{}" return json.loads(raw.decode("utf8") or "{}") def _send(self, code: int, body: bytes, content_type: str = "application/json"): self.send_response(code) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(body) def do_OPTIONS(self): self.send_response(204) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "*") self.end_headers() def do_GET(self): path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/" if path in ("/health", "/v1/health"): self._send(200, b'{"ok":true}') elif path in ("/props", "/v1/props"): # llama-server–style props; agent.py uses n_ctx for context_size props = { "n_ctx": CONTEXT_LENGTH, "default_generation_settings": {"n_ctx": CONTEXT_LENGTH}, } self._send(200, json.dumps(props).encode()) elif path in ("/v1/models", "/models"): try: ids = fetch_available_models() except Exception as e: err = json.dumps({"error": {"message": str(e), "type": "models_error"}}).encode() self._send(502, err) return models = { "object": "list", "data": [ { "id": m, "object": "model", "owned_by": "openai", "context_length": CONTEXT_LENGTH, "top_provider": {"context_length": CONTEXT_LENGTH}, } for m in ids ], } self._send(200, json.dumps(models).encode()) else: self._send(404, b'{"error":"not found"}') def do_POST(self): path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/" if path not in ("/v1/chat/completions", "/chat/completions"): self._send(404, b'{"error":"not found"}') return try: body = self._read_json() auth = ensure_auth() codex_body = chat_to_responses(body) model = body.get("model") or "gpt-5-codex" self.close_connection = True self.send_response(200) self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Cache-Control", "no-cache") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Connection", "close") self.end_headers() stream_codex_as_chat(codex_body, auth, self.wfile, model) except Exception as e: try: msg = json.dumps({"error": {"message": str(e), "type": "proxy_error"}}).encode() self._send(500, msg) except Exception: try: self.wfile.write(sse_line({"error": {"message": str(e)}})) self.wfile.write(b"data: [DONE]\n\n") except Exception: pass def main(): global AUTH_PATH ap = argparse.ArgumentParser(description="Chat Completions → Codex Responses proxy") ap.add_argument("--host", default=HOST) ap.add_argument("--port", type=int, default=PORT) ap.add_argument("--login", action="store_true", help="OAuth login and exit") ap.add_argument("--auth-file", default=AUTH_PATH) args = ap.parse_args() AUTH_PATH = args.auth_file if args.login: auth = oauth_login() print(f"Saved tokens to {AUTH_PATH}") print(f"account={auth.get('account_id')}") return if not load_auth(): print("No tokens found. Run: python3 codex_proxy.py --login", file=sys.stderr) sys.exit(1) class QuietServer(http.server.ThreadingHTTPServer): allow_reuse_address = True server = QuietServer((args.host, args.port), Handler) print(f"codex_proxy listening on http://{args.host}:{args.port}/v1") print(f"auth file: {AUTH_PATH}") print("agent.py:") print(" /set api http://127.0.0.1:%d/v1" % args.port) print(" /set api_key dummy") try: ids = fetch_available_models(force=True) print(f" # {len(ids)} models from Codex catalog:") for m in ids: print(f" /set model {m}") except Exception as e: print(f" # could not list models: {e}") print(" # retry: curl http://127.0.0.1:%d/v1/models" % args.port) try: server.serve_forever() except KeyboardInterrupt: print("\nbye") if __name__ == "__main__": main()