agnes-proxy/proxy.py

#!/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()