depau / Claude proxy

```bash python3 proxy.py ``` ``` ANTHROPIC_BASE_URL=http://127.0.0.1:8787 claude ```

#!/usr/bin/env python3 """Logging HTTP origin proxy for Claude Code. LOCAL test aid only. Claude is pointed at this proxy via ANTHROPIC_BASE_URL; every request is logged and forwarded to the real Anthropic backend untouched. Optionally the remote-settings endpoint is answered with a local mock payload (--mock) — note a custom base URL makes real remote settings ineligible, so the mock is mainly for manual curl testing, not the live settings fetch. Usage: python3 proxy.py # listen 127.0.0.1:8787 python3 proxy.py --mock --port 8788 # answer settings GET with mock python3 proxy.py --upstream http://localhost:9000 Then: ANTHROPIC_BASE_URL=http://127.0.0.1:8787 claude -p ... """ import argparse, hashlib, json, re, time from urllib.parse import urlparse import http.client from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer LOG = lambda *a: print(f"[{time.strftime('%H:%M:%S')}]", *a, flush=True) MOCK_SETTINGS = { "$schema": "https://json.schemastore.org/claude-code-settings.json", "companyAnnouncements": [ "PROXY-INJECTED announcement - this came from your local passthrough proxy, not Anthropic." ], "permissions": { "allow": ["Bash(echo *)", "Read(./**)"], "deny": [], "ask": [], "disableBypassPermissionsMode": "off" } } HOP_HEADERS = {"connect", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailers", "transfer-encoding", "upgrade", "host", "proxy-connection"} def route(server, method: str, path: str, headers, body: bytes, wfile): upstream = server.upstream LOG(f"{method} {path} Host={headers.get('Host', '?')} -> {upstream}") # Mock the settings endpoint. if server.mock and method == "GET" and re.search(r"/api/claude_code/settings\b", path): payload = server.mock_body wfile.write(b"HTTP/1.1 200 OK\r\n") wfile.write(b"Content-Type: application/json\r\n") wfile.write(f'ETag: "{hashlib.sha256(payload).hexdigest()}"\r\n'.encode()) wfile.write(f"Content-Length: {len(payload)}\r\n".encode()) wfile.write(b"\r\n") wfile.write(payload) wfile.flush() LOG(f" -> MOCK 200 (settings) len={len(payload)}") return # Forward to upstream. u = urlparse(upstream) conn = (http.client.HTTPSConnection(u.netloc, timeout=30) if u.scheme == "https" else http.client.HTTPConnection(u.netloc, timeout=30)) hs = {k: v for k, v in headers.items() if k.lower() not in HOP_HEADERS} try: conn.request(method, path, body=body or None, headers=hs) r = conn.getresponse() rbody = r.read() LOG(f" -> {r.status} {r.reason} len={len(rbody)}") wfile.write(f"HTTP/1.1 {r.status} {r.reason}\r\n".encode()) for k, v in r.getheaders(): if k.lower() not in HOP_HEADERS: wfile.write(f"{k}: {v}\r\n".encode()) # Delimit the body ourselves so the client never hangs on keep-alive. wfile.write(f"Content-Length: {len(rbody)}\r\n".encode()) wfile.write(b"Connection: close\r\n\r\n") wfile.write(rbody) wfile.flush() except Exception as e: LOG(f" !! forward error: {e}") wfile.write(b"HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\n\r\n") wfile.flush() finally: conn.close() class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt, *args): pass def do_GET(self): route(self.server, "GET", self.path, self.headers, None, self.wfile) def do_HEAD(self): route(self.server, "HEAD", self.path, self.headers, None, self.wfile) def do_POST(self): route(self.server, "POST", self.path, self.headers, self._read_body(), self.wfile) def _read_body(self): cl = int(self.headers.get("Content-Length") or 0) return self.rfile.read(cl) if cl else b"" def main(): ap = argparse.ArgumentParser() ap.add_argument("--host", default="127.0.0.1") ap.add_argument("--port", type=int, default=8787) ap.add_argument("--upstream", default="https://api.anthropic.com") ap.add_argument("--mock", action="store_true", help="answer GET /api/claude_code/settings with a mock payload") args = ap.parse_args() srv = ThreadingHTTPServer((args.host, args.port), Handler) srv.upstream = args.upstream srv.mock = args.mock srv.mock_body = json.dumps(MOCK_SETTINGS).encode() LOG(f"listening on http://{args.host}:{args.port} upstream={args.upstream} mock={args.mock}") try: srv.serve_forever() except KeyboardInterrupt: print("\nbye") srv.server_close() if __name__ == "__main__": main()