Simple AI Proxy Gateway（简易AI代理网关）

simple_ai_proxy_gateway.py

from fastapi import FastAPI, Request, Response
import httpx
import logging
import json

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()
client = httpx.AsyncClient()

@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy(request: Request, path: str):
    # 1. 可以在这里做鉴权或流量过滤
    # 示例，将请求转发到 小米mimo API，并替换 model 字段
    url = f"https://api.xiaomimimo.com/v1/chat/completions"
    
    # 记录请求信息
    logger.info(f"Request: {request.method} {url}")
    logger.info(f"Request Headers: {json.dumps(dict(request.headers), indent=2)}")
    
    body = await request.body()
    headers = dict(request.headers)
    
    # 移除不必要的头部，否则易403
    headers_to_remove = ["host", "connection", "content-length"]
    for header in headers_to_remove:
        headers.pop(header, None)
    
    if body:
        try:
            body_json = json.loads(body.decode())
            logger.info(f"Request Body (original): {json.dumps(body_json, indent=2, ensure_ascii=False)}")
            
            # 替换 model 字段
            if "model" in body_json:
                body_json["model"] = "mimo-v2-flash"
            
                body = json.dumps(body_json).encode()
                logger.info(f"Request Body (modified): {json.dumps(body_json, indent=2, ensure_ascii=False)}")

        except (json.JSONDecodeError, UnicodeDecodeError):
            logger.info(f"Request Body: {body}")

    # 2. 转发请求
    rp_resp = await client.request(
        method=request.method,
        url=url,
        headers=headers,
        content=body
    )

    # 记录响应信息
    logger.info(f"Response: {rp_resp.status_code}")
    logger.info(f"Response Headers: {json.dumps(dict(rp_resp.headers), indent=2)}")
    
    # 格式化响应体
    try:
        response_json = json.loads(rp_resp.content.decode())
        logger.info(f"Response Body: {json.dumps(response_json, indent=2, ensure_ascii=False)}")
    except (json.JSONDecodeError, UnicodeDecodeError):
        logger.info(f"Response Body: {rp_resp.content}")

    # 3. 返回响应
    # 过滤掉不应该转发的响应头
    response_headers = dict(rp_resp.headers)
    headers_to_exclude = ["content-encoding", "content-length", "transfer-encoding", "connection"]
    for header in headers_to_exclude:
        response_headers.pop(header, None)
    
    return Response(
        content=rp_resp.content,
        status_code=rp_resp.status_code,
        headers=response_headers
    )