#!/usr/bin/env python3 """ AgentRouter <-> OpenCode proxy (OpenAI-compatible endpoints). Fixes incompatibilities: 1. Injects the User-Agent required by AgentRouter (avoids 401). 2. Strips null fields from the request body (avoids 400). 3. Filters billing.summary and `data: null` events from the response. 4. Proper HTTP/1.1 chunked streaming with instant termination on `[DONE]`. 5. Translates Astra function tools calls to /v1/responses transparently. """ import http.client import http.server import json import os import ssl import sys import threading import time import urllib.error import urllib.request UPSTREAM = os.environ.get("AGENTROUTER_UPSTREAM", "https://ps.air-outer.com") PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 4182 API_KEY = os.environ.get("AGENTROUTER_API_KEY", "") DEBUG = os.environ.get("AGENTROUTER_DEBUG", "") == "1" USER_AGENT = "opencode/1.18.16" CTX = ssl.create_default_context() def strip_nulls(value): """Recursively remove dict entries whose value is None.""" if isinstance(value, dict): return {k: strip_nulls(v) for k, v in value.items() if v is not None} if isinstance(value, list): return [strip_nulls(item) for item in value] return value MODEL_FALLBACKS = { "claude-3-7-sonnet-20250219": "gpt-6-astra", "claude-3-7-sonnet": "gpt-6-astra", "claude-3-5-sonnet-20241022": "deepseek-v4-flash", "claude-3-5-sonnet": "deepseek-v4-flash", "claude-3-sonnet-20240229": "deepseek-v4-flash", "claude-3-5-haiku-20241022": "deepseek-v4-flash", "claude-3-haiku-20240307": "deepseek-v4-flash", } def clean_request_body(raw: bytes) -> bytes: """Drop null fields and remap models when needed.""" try: payload = json.loads(raw) except (json.JSONDecodeError, ValueError): return raw if not isinstance(payload, dict): return raw req_model = payload.get("model") max_tokens = payload.get("max_tokens", 100) # Remap quick health pings to Flash if max_tokens <= 5 and req_model in ("claude-opus-5", "claude-opus-4-8"): payload["model"] = "deepseek-v4-flash" elif req_model in MODEL_FALLBACKS: payload["model"] = MODEL_FALLBACKS[req_model] # Disable heavy thinking for instant Flash responses if payload.get("model") == "deepseek-v4-flash" and "thinking" not in payload: payload["thinking"] = {"type": "disabled"} return json.dumps(strip_nulls(payload), separators=(",", ":")).encode() def chat_to_responses_request(payload: dict) -> dict: """Translate OpenAI chat/completions body to OpenAI responses API body.""" req = { "model": payload.get("model", "gpt-6-astra"), "stream": payload.get("stream", False), } responses_input = [] for msg in payload.get("messages", []): role = msg.get("role") content = msg.get("content") tool_calls = msg.get("tool_calls") if role == "tool": responses_input.append({ "type": "function_call_output", "call_id": msg.get("tool_call_id", ""), "output": str(content or "") }) elif role == "assistant" and tool_calls: if content: responses_input.append({"role": "assistant", "content": content}) for tc in tool_calls: fn = tc.get("function", {}) responses_input.append({ "type": "function_call", "call_id": tc.get("id", ""), "name": fn.get("name", ""), "arguments": fn.get("arguments", "{}") }) else: responses_input.append({"role": role, "content": content or ""}) req["input"] = responses_input raw_tools = payload.get("tools") or [] if raw_tools: converted_tools = [] for t in raw_tools: if t.get("type") == "function" and "function" in t: fn = t["function"] item = { "type": "function", "name": fn.get("name"), "description": fn.get("description", ""), "parameters": fn.get("parameters", {}) } if "strict" in fn: item["strict"] = fn["strict"] converted_tools.append(item) else: converted_tools.append(t) req["tools"] = converted_tools for param in ("temperature", "top_p", "max_output_tokens"): if param in payload: req[param] = payload[param] return strip_nulls(req) def responses_to_chat_completion(resp_json: dict, req_model: str) -> dict: """Translate responses JSON back to OpenAI chat.completion format.""" tool_calls = [] content_parts = [] for item in resp_json.get("output", []): itype = item.get("type") if itype == "function_call": tool_calls.append({ "id": item.get("call_id") or item.get("id", ""), "type": "function", "function": { "name": item.get("name", ""), "arguments": item.get("arguments", "{}") } }) elif itype == "message": for part in item.get("content", []): if isinstance(part, dict) and part.get("type") == "text": content_parts.append(part.get("text", "")) elif isinstance(part, str): content_parts.append(part) message_obj = {"role": "assistant"} if tool_calls: message_obj["tool_calls"] = tool_calls message_obj["content"] = "".join(content_parts) if content_parts else None finish_reason = "tool_calls" else: message_obj["content"] = "".join(content_parts) finish_reason = "stop" return { "id": resp_json.get("id", "chatcmpl-astra"), "object": "chat.completion", "created": resp_json.get("created_at", int(time.time())), "model": req_model, "choices": [{ "index": 0, "message": message_obj, "finish_reason": finish_reason }], "usage": resp_json.get("usage", {}) } def is_billing_summary(payload) -> bool: """AgentRouter injects billing.summary objects the AI SDK cannot parse.""" return isinstance(payload, dict) and payload.get("object") == "billing.summary" def clean_sse_line(line: bytes): """Filter a single SSE line. Return None to drop it.""" stripped = line.strip() if not stripped: return line if stripped.startswith(b"data:"): payload = stripped[len(b"data:"):].strip() if payload in (b"", b"null"): return None if payload == b"[DONE]": return line try: parsed = json.loads(payload) except (json.JSONDecodeError, ValueError): return line return None if is_billing_summary(parsed) else line if stripped[:1] in (b"{", b"["): try: parsed = json.loads(stripped) except (json.JSONDecodeError, ValueError): return line if is_billing_summary(parsed): return None return line def clean_nonstream_body(raw: bytes) -> bytes: """Replace a billing.summary response with a readable JSON error.""" if not raw: return raw try: payload = json.loads(raw) except (json.JSONDecodeError, ValueError): return raw if is_billing_summary(payload): sys.stderr.write("[ar] filtered billing.summary from non-stream response\n") return json.dumps( { "error": { "message": "agentrouter billing.summary is not a chat response", "type": "proxy_filtered", } } ).encode() return raw class ProxyHandler(http.server.BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def do_POST(self): self._proxy() def do_GET(self): self._proxy() def do_OPTIONS(self): self.send_response(200) 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.send_header("Content-Length", "0") self.end_headers() def _proxy(self): content_len = int(self.headers.get("Content-Length", 0)) raw_body = self.rfile.read(content_len) if content_len > 0 else None body = clean_request_body(raw_body) if raw_body else None parsed_payload = None if body: try: parsed_payload = json.loads(body) except Exception: pass req_model = parsed_payload.get("model", "") if parsed_payload else "" is_astra = "astra" in req_model.lower() has_tools = bool(parsed_payload and (parsed_payload.get("tools") or parsed_payload.get("functions"))) use_responses_bridge = is_astra and has_tools and self.path.endswith("/chat/completions") headers = { k: v for k, v in self.headers.items() if k.lower() not in ("host", "connection", "transfer-encoding", "accept-encoding") } auth_key = API_KEY or self.headers.get("Authorization", "").replace("Bearer ", "").strip() if auth_key: headers["Authorization"] = f"Bearer {auth_key}" headers["User-Agent"] = USER_AGENT headers["Accept-Encoding"] = "identity" if use_responses_bridge: responses_payload = chat_to_responses_request(parsed_payload) body = json.dumps(responses_payload, separators=(",", ":")).encode() target_path = "/v1/responses" else: target_path = self.path if body is not None: headers["Content-Length"] = str(len(body)) url = f"{UPSTREAM}{target_path}" request = urllib.request.Request( url, data=body, headers=headers, method=self.command ) is_stream = bool(body) and b'"stream":true' in body.replace(b" ", b"") try: response = urllib.request.urlopen(request, context=CTX, timeout=300) if use_responses_bridge: if is_stream: 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("Transfer-Encoding", "chunked") self.send_header("Connection", "keep-alive") self.end_headers() self._stream_responses_as_sse(response, req_model) else: try: raw_response = response.read() except http.client.IncompleteRead as exc: raw_response = exc.partial resp_json = json.loads(raw_response) translated = responses_to_chat_completion(resp_json, req_model) out_bytes = json.dumps(translated).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(out_bytes))) self.send_header("Connection", "keep-alive") self.end_headers() self.wfile.write(out_bytes) return if is_stream: self.send_response(response.status) for k, v in response.headers.items(): if k.lower() not in ( "transfer-encoding", "connection", "content-length", ): self.send_header(k, v) self.send_header("Transfer-Encoding", "chunked") self.send_header("Connection", "keep-alive") self.end_headers() self._stream_sse(response) else: try: raw_response = response.read() except http.client.IncompleteRead as exc: raw_response = exc.partial payload = clean_nonstream_body(raw_response) self.send_response(response.status) for k, v in response.headers.items(): if k.lower() not in ( "transfer-encoding", "connection", "content-length", ): self.send_header(k, v) self.send_header("Content-Length", str(len(payload))) self.send_header("Connection", "keep-alive") self.end_headers() self.wfile.write(payload) except urllib.error.HTTPError as exc: err_body = exc.read() sys.stderr.write( f"[ar] ERROR {exc.code}: {err_body.decode(errors='replace')[:500]}\n" ) status_code = exc.code if status_code == 500 and (b"sensitive_words_detected" in err_body or b"sensitive words detected" in err_body): status_code = 400 self.send_response(status_code) self.send_header( "Content-Type", exc.headers.get("Content-Type", "application/json") ) self.send_header("Content-Length", str(len(err_body))) self.end_headers() self.wfile.write(err_body) except Exception as exc: # noqa: BLE001 sys.stderr.write(f"[ar] EXCEPTION: {exc}\n") payload = json.dumps({"error": {"message": str(exc)}}).encode() self.send_response(502) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def _stream_responses_as_sse(self, response, req_model: str): """Translate /v1/responses SSE stream to OpenAI chat.completion.chunk SSE.""" resp_id = "chatcmpl-" + str(int(time.time())) tool_indices = {} has_tools = False has_sent_role = False created_ts = int(time.time()) try: while True: line = response.readline() if not line: break line_str = line.decode(errors="replace").strip() if not line_str.startswith("data:"): continue raw_json = line_str[5:].strip() if not raw_json or raw_json == "[DONE]": continue try: evt = json.loads(raw_json) except Exception: continue evt_type = evt.get("type") if evt_type == "response.created": resp_id = evt.get("response", {}).get("id", resp_id) created_ts = evt.get("response", {}).get("created_at", created_ts) elif evt_type == "response.output_item.added": item = evt.get("item", {}) itype = item.get("type") if itype == "function_call": has_tools = True item_id = item.get("id") or item.get("call_id") or str(len(tool_indices)) if item_id not in tool_indices: tool_indices[item_id] = len(tool_indices) idx = tool_indices[item_id] chunk = { "id": resp_id, "object": "chat.completion.chunk", "created": created_ts, "model": req_model, "choices": [{ "index": 0, "delta": { "role": "assistant", "content": None, "tool_calls": [{ "index": idx, "id": item.get("call_id") or item.get("id", ""), "type": "function", "function": { "name": item.get("name", ""), "arguments": "" } }] }, "finish_reason": None }] } self._write_chunk(f"data: {json.dumps(chunk)}\n\n".encode()) elif itype == "message" and not has_sent_role: has_sent_role = True chunk = { "id": resp_id, "object": "chat.completion.chunk", "created": created_ts, "model": req_model, "choices": [{ "index": 0, "delta": { "role": "assistant", "content": "" }, "finish_reason": None }] } self._write_chunk(f"data: {json.dumps(chunk)}\n\n".encode()) elif evt_type == "response.function_call_arguments.delta": item_id = evt.get("item_id") idx = tool_indices.get(item_id, 0) chunk = { "id": resp_id, "object": "chat.completion.chunk", "created": created_ts, "model": req_model, "choices": [{ "index": 0, "delta": { "tool_calls": [{ "index": idx, "function": { "arguments": evt.get("delta", "") } }] }, "finish_reason": None }] } self._write_chunk(f"data: {json.dumps(chunk)}\n\n".encode()) elif evt_type in ("response.output_text.delta", "response.text.delta"): delta_text = evt.get("delta", "") delta_obj = {"content": delta_text} if not has_sent_role: has_sent_role = True delta_obj["role"] = "assistant" chunk = { "id": resp_id, "object": "chat.completion.chunk", "created": created_ts, "model": req_model, "choices": [{ "index": 0, "delta": delta_obj, "finish_reason": None }] } self._write_chunk(f"data: {json.dumps(chunk)}\n\n".encode()) elif evt_type == "response.completed": finish_reason = "tool_calls" if has_tools else "stop" resp_obj = evt.get("response", {}) usage = resp_obj.get("usage") chunk = { "id": resp_id, "object": "chat.completion.chunk", "created": created_ts, "model": req_model, "choices": [{ "index": 0, "delta": {}, "finish_reason": finish_reason }], "usage": usage } self._write_chunk(f"data: {json.dumps(chunk)}\n\n".encode()) self._write_chunk(b"data: [DONE]\n\n") break self.wfile.write(b"0\r\n\r\n") self.wfile.flush() except (ConnectionResetError, BrokenPipeError): return def _stream_sse(self, response): """Forward the SSE stream, dropping nulls/billing and terminating on [DONE].""" try: while True: line = response.readline() if not line: break cleaned = clean_sse_line(line) if cleaned is not None: self._write_chunk(cleaned) if b"[DONE]" in line: break self.wfile.write(b"0\r\n\r\n") self.wfile.flush() except (ConnectionResetError, BrokenPipeError): return def _write_chunk(self, data: bytes): self.wfile.write(f"{len(data):X}\r\n".encode()) self.wfile.write(data) self.wfile.write(b"\r\n") self.wfile.flush() def log_message(self, fmt, *args): sys.stderr.write(f"[ar] {args[0]}\n") class ThreadedHTTPServer(http.server.HTTPServer): daemon_threads = True def process_request(self, request, client_address): thread = threading.Thread( target=self.process_request_thread, args=(request, client_address) ) thread.daemon = True thread.start() def process_request_thread(self, request, client_address): try: self.finish_request(request, client_address) except Exception: # noqa: BLE001 self.handle_error(request, client_address) finally: self.shutdown_request(request) if __name__ == "__main__": server = ThreadedHTTPServer(("127.0.0.1", PORT), ProxyHandler) print(f"[ar] Proxy actif sur ::{PORT} -> {UPSTREAM}") try: server.serve_forever() except KeyboardInterrupt: print("\n[ar] Proxy arrêté.") server.server_close()