#!/usr/bin/env python3 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import urllib.request import urllib.error import json # ============================================================ # Agnes 配置 # ============================================================ AGNES_URL = "https://apihub.agnes-ai.com/v1/responses" AGNES_KEY = "sk-***" # ============================================================ # Agnes 支持的 Tool 类型 # ============================================================ # SUPPORTED_TOOLS = { # "function", # "web_search_preview", # "code_interpreter", # "mcp", # } SUPPORTED_TOOLS = { "function", "web_search_preview", # "web_search", "code_interpreter", "file_search", "computer_use", "mcp", # "tool_search", "namespace", } # ============================================================ # HTTP Handler # ============================================================ class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def do_POST(self): # ---------------------------------------------------- # 只处理 /v1/responses # ---------------------------------------------------- if self.path != "/v1/responses": self.send_error(404) return # ---------------------------------------------------- # 读取请求 # ---------------------------------------------------- try: length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) data = json.loads(body) except Exception as e: print("\n===== REQUEST PARSE ERROR =====") print(repr(e)) self.send_response(400) self.send_header("Content-Type", "application/json") self.send_header("Connection", "close") self.end_headers() error = json.dumps({ "error": { "message": f"Invalid JSON request: {e}", "type": "invalid_request_error" } }).encode("utf-8") self.wfile.write(error) return # ==================================================== # 1. Model 映射 # ==================================================== original_model = data.get("model") if original_model == "gpt-5.5": data["model"] = "agnes-3.0-flash" print("\n===== MODEL =====") print("Codex :", original_model) print("Agnes :", data.get("model")) # ==================================================== # 2. Tool 过滤 # # Agnes 当前支持： # # function # web_search_preview # code_interpreter # mcp # # Codex 可能发送： # # custom # tool_search # 以及其他内部工具 # # 对不支持的类型直接删除 # ==================================================== if isinstance(data.get("tools"), list): old_tools = data["tools"] new_tools = [] print("\n===== TOOLS =====") for i, tool in enumerate(old_tools): tool_type = tool.get("type") if tool_type in SUPPORTED_TOOLS: new_tools.append(tool) print( f"[{i}] {tool_type:<25} KEEP" ) else: print( f"[{i}] {str(tool_type):<25} DROP" ) data["tools"] = new_tools # ------------------------------------------------ # 如果已经没有任何 tools # Agnes 不允许： # # "tool_choice": "auto" # # 在没有 tools 的情况下存在 # ------------------------------------------------ if not data["tools"]: data.pop("tools", None) data.pop("tool_choice", None) # ---------------------------------------------------- # 没有 tools 时，强制删除 tool_choice # ---------------------------------------------------- if "tools" not in data: data.pop("tool_choice", None) # ==================================================== # 打印最终 Tools # ==================================================== print("\n===== FINAL TOOLS =====") if "tools" in data: for i, tool in enumerate(data["tools"]): print( f"[{i}] {tool.get('type')}" ) else: print("No tools") # ==================================================== # 重新生成 JSON # ==================================================== body = json.dumps( data, ensure_ascii=False ).encode("utf-8") # ==================================================== # 打印最终请求 # ==================================================== print("\n===== REQUEST =====") try: print( json.dumps( data, ensure_ascii=False, indent=2 ) ) except Exception: print(body[:2000]) # ==================================================== # 请求 Agnes # ==================================================== req = urllib.request.Request( AGNES_URL, data=body, method="POST", headers={ "Authorization": f"Bearer {AGNES_KEY}", "Content-Type": "application/json", # Codex 一般会请求 SSE "Accept": self.headers.get( "Accept", "text/event-stream" ), }, ) # ==================================================== # 发送到 Agnes # ==================================================== try: resp = urllib.request.urlopen( req, timeout=600 ) print( "\nAgnes:", resp.status ) # ------------------------------------------------ # 返回 HTTP 状态 # ------------------------------------------------ self.send_response(resp.status) # ------------------------------------------------ # Content-Type # ------------------------------------------------ content_type = resp.headers.get( "Content-Type", "text/event-stream" ) self.send_header( "Content-Type", content_type ) # ------------------------------------------------ # SSE / HTTP 相关 Header # ------------------------------------------------ self.send_header( "Cache-Control", "no-cache" ) self.send_header( "Connection", "keep-alive" ) self.end_headers() # ================================================= # 流式转发 Agnes SSE # ================================================= while True: chunk = resp.read(4096) if not chunk: break self.wfile.write(chunk) self.wfile.flush() print("\n===== AGNES STREAM END =====") # ==================================================== # Agnes HTTP Error # ==================================================== except urllib.error.HTTPError as e: error_body = e.read() print( "\n===== AGNES ERROR =====" ) print( error_body.decode( "utf-8", errors="replace" ) ) try: self.send_response(e.code) self.send_header( "Content-Type", "application/json" ) self.send_header( "Connection", "close" ) self.end_headers() self.wfile.write( error_body ) except Exception as send_error: print( "ERROR SENDING ERROR RESPONSE:", repr(send_error) ) # ==================================================== # 其他异常 # ==================================================== except Exception as e: print( "\n===== PROXY ERROR =====" ) print( repr(e) ) try: self.send_response(500) self.send_header( "Content-Type", "application/json" ) self.send_header( "Connection", "close" ) self.end_headers() msg = json.dumps({ "error": { "message": str(e), "type": "proxy_error" } }).encode("utf-8") self.wfile.write(msg) except Exception: pass # ======================================================== # 禁止默认日志格式 # ======================================================== def log_message(self, fmt, *args): print( fmt % args ) # ============================================================ # 启动服务器 # ============================================================ server = ThreadingHTTPServer( ("0.0.0.0", 4000), Handler ) print( "Proxy listening on :4000" ) print( "→", AGNES_URL ) print( "Supported tools:", ", ".join(SUPPORTED_TOOLS) ) server.serve_forever() # #!/usr/bin/env python3 # from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # import urllib.request # import json # AGNES_URL = "https://apihub.agnes-ai.com/v1/responses" # AGNES_KEY = "sk-5bWELiz7grmKat0QTTQ6WJhEF4JWg8gMTPGOb4kcVOCMBrHH" # class Handler(BaseHTTPRequestHandler): # protocol_version = "HTTP/1.1" # def do_POST(self): # if self.path != "/v1/responses": # self.send_error(404) # return # length = int(self.headers.get("Content-Length", 0)) # body = self.rfile.read(length) # data = json.loads(body) # if data.get("model") == "gpt-5.5": # data["model"] = "agnes-3.0-flash" # # 2. 删除 Agnes 不支持的 custom tools # if "tools" in data: # tools = data["tools"] # if isinstance(tools, list): # data["tools"] = [ # tool for tool in tools # if tool.get("type") != "custom" # ] # # 如果已经没有 tools，tool_choice 也必须删除 # if not data["tools"]: # data.pop("tools", None) # data.pop("tool_choice", None) # # 3. 没有 tools 时不能存在 tool_choice # if "tools" not in data: # data.pop("tool_choice", None) # body = json.dumps( # data, # ensure_ascii=False # ).encode("utf-8") # print("\n===== REQUEST =====") # try: # print(json.dumps(json.loads(body), ensure_ascii=False, indent=2)) # except: # print(body[:1000]) # req = urllib.request.Request( # AGNES_URL, # data=body, # method="POST", # headers={ # "Authorization": f"Bearer {AGNES_KEY}", # "Content-Type": "application/json", # "Accept": self.headers.get( # "Accept", # "text/event-stream" # ), # }, # ) # try: # resp = urllib.request.urlopen(req, timeout=600) # print("Agnes:", resp.status) # self.send_response(resp.status) # content_type = resp.headers.get( # "Content-Type", # "text/event-stream" # ) # self.send_header("Content-Type", content_type) # self.send_header("Cache-Control", "no-cache") # self.send_header("Connection", "keep-alive") # self.end_headers() # while True: # chunk = resp.read(4096) # if not chunk: # break # self.wfile.write(chunk) # self.wfile.flush() # except urllib.error.HTTPError as e: # error_body = e.read() # print("\n===== AGNES ERROR =====") # print(error_body.decode("utf-8", errors="replace")) # self.send_response(e.code) # self.send_header("Content-Type", "application/json") # self.end_headers() # self.wfile.write(error_body) # except Exception as e: # print("ERROR:", repr(e)) # try: # self.send_response(500) # self.send_header("Content-Type", "application/json") # self.end_headers() # msg = json.dumps({ # "error": str(e) # }).encode() # self.wfile.write(msg) # except: # pass # def log_message(self, fmt, *args): # print(fmt % args) # server = ThreadingHTTPServer( # ("0.0.0.0", 9000), # Handler # ) # print("Proxy listening on :9000") # print("→", AGNES_URL) # server.serve_forever()