update

.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+.env

Dockerfile

@@ -0,0 +1,16 @@
+# Read the doc: https://huggingface.co/docs/hub/spaces-sdks-docker
+# you will also find guides on how best to write your Dockerfile
+
+FROM python:3.12
+
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+COPY --chown=user . /app
+CMD ["uvicorn", "proxyserver-fastapi:app", "--host", "0.0.0.0", "--port", "7860"]

agent_example.py

@@ -0,0 +1,103 @@
+import requests
+import json
+import os
+from dotenv import load_dotenv
+
+# 加载 .env 文件中的环境变量
+load_dotenv()
+
+# 从环境变量中获取配置
+# 代理服务器的地址
+# 确保 anyapi.py 正在运行在 127.0.0.1:7860
+BASE_URL = os.getenv("BASE_URL", "http://localhost:7860/v1/https/generativelanguage.googleapis.com/v1beta/openai")
+
+# 目标 OpenAI 兼容 API 的完整路径
+# 例如，对于智谱AI的GLM模型，其API路径可能是 https://open.bigmodel.cn/api/paas/v4/chat/completions
+# 代理服务器会将其解析为协议、域名和路径
+ENTER_POINT = os.getenv("ENTER_POINT", "/chat/completions")
+
+# 完整的代理URL
+API_URL = f"{BASE_URL}{ENTER_POINT}"
+
+# 您的 API 密钥
+# 注意：在实际应用中，API 密钥不应硬编码在代码中，应通过环境变量等方式安全管理。
+API_KEY = os.getenv("API_KEY", "YOUR_ACTUAL_API_KEY_HERE")
+MODEL_NAME = os.getenv("MODEL_NAME", "gemini-2.5-flash-preview-05-20")
+
+def call_openai_compatible_api(prompt_message: str):
+    print("正在调用 OpenAI 兼容 API...")
+    print('prompt_message',prompt_message)
+    """
+    调用 OpenAI 兼容 API 的示例函数。
+    """
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {API_KEY}",
+        "accept-encoding": "identity"
+    }
+
+    payload = {
+        "model": f"{MODEL_NAME}", # 替换为您希望使用的模型名称，例如 "gpt-3.5-turbo", "glm-4" 等
+        "messages": [
+            {"role": "system", "content": "你是一个有用的助手。"},
+            {"role": "user", "content": prompt_message}
+        ],
+        "stream": True # 如果需要流式响应，可以设置为 True
+    }
+
+    print(f"正在向代理服务器发送请求: {API_URL}")
+    print(f"请求头部: {headers}")
+    print(f"请求体: {json.dumps(payload, indent=2, ensure_ascii=False)}")
+
+    try:
+        response = requests.post(API_URL, headers=headers, json=payload, stream=payload.get("stream", False))
+        response.raise_for_status() # 如果请求返回了错误状态码，会抛出 HTTPError
+
+        print("\n代理服务器响应:")
+        if payload.get("stream"):
+            full_response_content = ""
+            for line in response.iter_lines():
+                if line:
+                    decoded_line = line.decode('utf-8')
+                    if decoded_line.startswith("data:"):
+                        json_data = decoded_line[len("data:"):].strip()
+                        if json_data == "[DONE]":
+                            break
+                        try:
+                            parsed_json = json.loads(json_data)
+                            # 打印或处理每个流式块
+                            # print(json.dumps(parsed_json, indent=2, ensure_ascii=False))
+                            # 假设我们只关心 content
+                            print(json.dumps(parsed_json, indent=2, ensure_ascii=False))
+                            full_response_content += json_data # 收集原始JSON数据
+                        except json.JSONDecodeError as e:
+                            print(f"JSON 解析错误: {e} - {json_data}")
+            print("\n--- 流式响应结束 ---")
+            return {"raw_stream_data": full_response_content} # 返回一个包含原始流数据的字典
+        else:
+            print(json.dumps(response.json(), indent=2, ensure_ascii=False))
+            return response.json()
+
+    except requests.exceptions.HTTPError as http_err:
+        print(f"HTTP 错误发生: {http_err}")
+        print(f"响应状态码: {response.status_code}")
+        print(f"响应体: {response.text}")
+    except requests.exceptions.ConnectionError as conn_err:
+        print(f"连接错误发生: {conn_err}")
+        print("请确保代理服务器 (anyapi.py) 正在运行。")
+    except requests.exceptions.Timeout as timeout_err:
+        print(f"请求超时: {timeout_err}")
+    except requests.exceptions.RequestException as req_err:
+        print(f"发生未知错误: {req_err}")
+
+if __name__ == "__main__":
+    # 在运行此脚本之前，请确保：
+    # 1. anyapi.py 代理服务器正在运行。
+    # 2. 您已将 API_KEY 替换为您的实际 API 密钥。
+    # 3. 您已安装 requests 库 (pip install requests)。
+
+    user_query = "你好！请介绍一下你自己。"
+    call_openai_compatible_api(user_query)
+
+    # 您可以尝试不同的查询
+    # call_openai_compatible_api("请给我讲一个关于人工智能的笑话。")

note.md

@@ -0,0 +1,64 @@
+http://localhost:7860/v1/https/open.bigmodel.cn/api/paas/v4/chat/completions
+{
+    "model": "glm-4-flash",
+    "messages": [
+        {"role": "system", "content": "你是一个有用的助手"},
+        {"role": "user", "content": "你好！请介绍一下自己。"}
+    ],
+    "stream": true
+}
+
+
+
+http://127.0.0.1:7860/v1/https/generativelanguage.googleapis.com/v1beta/openai/chat/completions
+{
+    "model":"gemini-2.5-flash-preview-05-20",
+    "messages":[
+        {"role": "system", "content": "你是一个有用的助手"},
+        {"role": "user", "content": "你好！请介绍一下自己。"}
+    ],
+    "stream": true
+}
+
+
+curl -v --location 'http://127.0.0.1:7860/v1/https/generativelanguage.googleapis.com/v1beta/openai/chat/completions' \
+--header 'Authorization: Bearer xxx' \
+--header 'accept-encoding: identity' \
+--header 'User-Agent: PostmanRuntime/7.45.0' \
+--header 'Content-Type: application/json' \
+--data '{
+    "model":"gemini-2.5-flash-preview-05-20",
+    "messages":[
+        {"role": "system", "content": "你是一个有用的助手"},
+        {"role": "user", "content": "你好！请介绍一下自己。"}
+    ],
+    "stream": false
+}'
+
+
+
+转发程序为：@/proxyserver.py 
+
+假设数据为：
+{
+    "model":"gemini-2.5-flash-preview-05-20",
+    "messages":[
+        {"role": "system", "content": "你是一个有用的助手"},
+        {"role": "user", "content": "你好！请介绍一下自己。"}
+    ],
+    "stream": false
+}
+
+使用postman发送时没有问题
+
+使用curl 发送时，出现问题，数据无法返回，proxyserver 的log 为：
+```bash
+代理请求到 https://generativelanguage.googleapis.com/v1beta/openai/chat/completions
+/home/tanbushi/miniconda3/envs/any-api/lib/python3.11/site-packages/urllib3/connectionpool.py:1097: InsecureRequestWarning: Unverified HTTPS request is being made to host '127.0.0.1'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/latest/advanced-usage.html#tls-warnings
+  warnings.warn(
+目标API响应状态码: 200
+目标API响应体: {"choices":[{"finish_reason":"stop","index":0,"message":{"content":"你好！很高兴能与你交流。\n\n我是一个由 Google 训练的**大型语言模型**。我的主要目标是协助你，提供信息，并进行各种基于语言的任务。\n\n我可以做很多事情，比如：\n*   **回答你的问题**：只要是我的知识范围内，我会尽力给你准确、有用的答案。\n*   **撰写不同类型的内容**：无论是文章、邮件、诗歌、故事、代码，还是仅仅是帮你整理思绪，我都可以尝试。\n*   **进行翻译**：在不同的语言之间进行文本转换。\n*   **提供创意灵感**：当你需要一些新的想法或突破思维定势时，我可以提供帮助。\n*   **总结信息**：将长篇内容提炼成简明扼要的要点。\n*   **学习新知识**：帮助你理解复杂的概念或探索新的领域。\n*   **进行对话**：与你进行自然的交流，解答你的疑惑或只是聊聊天。\n\n我没有身体，也没有个人情感、意识或观点。我的存在是为了处理和生成文本，并以最有用的方式为你服务。\n\n如果你有任何问题或需要...
+127.0.0.1 - - [03/Aug/2025 12:01:11] "POST /v1/https/generativelanguage.googleapis.com/v1beta/openai/chat/completions HTTP/1.1" 200 -
+```
+
+

proxyserver-fastapi.py

@@ -0,0 +1,89 @@
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import StreamingResponse, Response
+import httpx
+import uvicorn
+import asyncio
+
+app = FastAPI()
+
+@app.api_route("/v1/{url_path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"])
+async def proxy_request(url_path: str, request: Request):
+    print(f"接收到的 url_path: {url_path}")
+
+    # url_path 示例: https/open.bigmodel.cn/api/paas/v4/chat/completions
+    # 找到第一个 '/' 的位置，分隔协议和域名
+    first_slash_idx = url_path.find('/')
+    if first_slash_idx == -1:
+        raise HTTPException(status_code=400, detail="无效的URL路径格式。期望协议/域名/路径。")
+    
+    protocol = url_path[:first_slash_idx] # 'https'
+    print(f"解析出的协议: {protocol}")
+
+    # 找到第二个 '/' 的位置，分隔域名和实际的路径
+    second_slash_idx = url_path.find('/', first_slash_idx + 1)
+
+    if second_slash_idx == -1:
+        domain = url_path[first_slash_idx + 1:]
+        remaining_path = ''
+    else:
+        domain = url_path[first_slash_idx + 1:second_slash_idx]
+        remaining_path = url_path[second_slash_idx:]
+
+    target_url = f"{protocol}://{domain}{remaining_path}"
+    print(f"\n\n\n代理请求到 {target_url}")
+
+    # 转发原始请求的头部，排除 'Host' 头部以避免冲突
+    # FastAPI 的 request.headers 是不可变的，需要转换为字典
+    headers = {key: value for key, value in request.headers.items() if key.lower() != 'host'}
+    
+    # 获取请求体
+    request_body = await request.body()
+    
+    # 获取查询参数
+    query_params = request.query_params
+
+    async with httpx.AsyncClient(verify=True, follow_redirects=False) as client:
+        try:
+            # 使用 httpx 库向目标 URL 发送请求
+            resp = await client.request(
+                method=request.method,
+                url=target_url,
+                headers=headers,
+                content=request_body, # 使用 content 传递请求体
+                params=query_params,
+                timeout=30.0, # 设置超时时间为30秒
+            )
+
+            # 打印目标 API 返回的实际状态码和响应体，用于调试
+            print(f"目标API响应状态码: {resp.status_code}")
+            print(f"目标API响应体: {resp.text[:500]}...") # 打印前500个字符，避免过长
+
+            # 构建响应头部
+            excluded_headers = ['content-encoding'] # 保持与 Flask 版本一致
+            response_headers = {
+                name: value for name, value in resp.headers.items()
+                if name.lower() not in excluded_headers
+            }
+            
+            # 返回流式响应内容
+            # httpx 的 .aiter_bytes() 返回异步迭代器
+            async def generate_response():
+                async for chunk in resp.aiter_bytes(chunk_size=8192):
+                    yield chunk
+
+            return StreamingResponse(generate_response(), status_code=resp.status_code, headers=response_headers)
+
+        except httpx.RequestError as e:
+            error_detail = f"代理请求到 {target_url} 失败: {type(e).__name__} - {e}"
+            print(f"代理请求失败: {error_detail}")
+            if e.request:
+                print(f"请求信息: {e.request.method} {e.request.url}")
+            if hasattr(e, 'response') and e.response:
+                print(f"响应信息: {e.response.status_code} {e.response.text[:200]}...")
+            raise HTTPException(status_code=500, detail=error_detail)
+
+# if __name__ == '__main__':
+#     # 提示：请确保您已激活 conda 环境 'any-api' (conda activate any-api)
+#     # 提示：请确保已安装 FastAPI, Uvicorn 和 httpx 库 (pip install fastapi uvicorn httpx)
+#     print(f"代理服务器正在 0.0.0.0:7860 上启动")
+#     uvicorn.run(app, host="0.0.0.0", port=7860)

proxyserver-flask.py

@@ -0,0 +1,97 @@
+import threading
+import requests
+from flask import Flask, request, jsonify, abort, Response
+
+class ProxyServer:
+    def __init__(self, host='127.0.0.1', port=7860):
+        self.app = Flask(__name__)
+        self.host = host
+        self.port = port
+        self._setup_routes()
+
+    def _setup_routes(self):
+        # 定义一个通用的代理路由，捕获所有以 /v1/ 开头的请求路径
+        # 例如：/v1/https/open.bigmodel.cn/api/paas/v4/chat/completions
+        @self.app.route('/v1/<path:url_path>', methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'OPTIONS'])
+        def proxy_request(url_path):
+            print(f"接收到的 url_path: {url_path}")
+
+            # url_path 示例: https/open.bigmodel.cn/api/paas/v4/chat/completions
+            # 找到第一个 '/' 的位置，分隔协议和域名
+            first_slash_idx = url_path.find('/')
+            if first_slash_idx == -1:
+                abort(400, description="无效的URL路径格式。期望协议/域名/路径。")
+            
+            protocol = url_path[:first_slash_idx] # 'https'
+            print(f"解析出的协议: {protocol}")
+
+            # 找到第二个 '/' 的位置，分隔域名和实际的路径
+            # 从 first_slash_idx + 1 开始查找，即从 'open.bigmodel.cn/api/paas/v4/chat/completions' 开始
+            second_slash_idx = url_path.find('/', first_slash_idx + 1)
+
+            if second_slash_idx == -1:
+                # 如果没有第二个斜杠，说明只有协议和域名，没有后续路径
+                domain = url_path[first_slash_idx + 1:] # 'open.bigmodel.cn'
+                remaining_path = ''
+            else:
+                domain = url_path[first_slash_idx + 1:second_slash_idx] # 'open.bigmodel.cn'
+                remaining_path = url_path[second_slash_idx:] # '/api/paas/v4/chat/completions'
+
+            target_url = f"{protocol}://{domain}{remaining_path}"
+            print(f"\n\n\n代理请求到 {target_url}")
+
+            # 转发原始请求的头部，排除 'Host' 头部以避免冲突
+            headers = {key: value for key, value in request.headers if key.lower() != 'host'}
+            
+            try:
+                # 使用 requests 库向目标 URL 发送请求，并转发原始请求的方法、头部、数据和查询参数
+                # stream=True 用于流式传输响应，这对于代理大文件或保持连接非常有用
+                resp = requests.request(
+                    method=request.method,
+                    url=target_url,
+                    headers=headers,
+                    data=request.get_data(),
+                    params=request.args,
+                    allow_redirects=False, # 不允许 requests 自动处理重定向
+                    verify=True # 禁用 SSL 证书验证，仅用于调试
+                )
+
+                # 打印目标 API 返回的实际状态码和响应体，用于调试
+                print(f"目标API响应状态码: {resp.status_code}")
+                print(f"目标API响应体: {resp.text[:500]}...") # 打印前500个字符，避免过长
+
+                # 构建响应头部，并确保 Content-Length 被移除，因为我们将使用流式传输
+                # 如果原始响应是分块编码，requests 会自动处理，但我们通过流式传输来确保 Flask 也以流式方式发送
+                excluded_headers = ['content-encoding']
+                response_headers = [
+                    (name, value) for name, value in resp.headers.items()
+                    if name.lower() not in excluded_headers
+                ]
+                
+                # 返回流式响应内容
+                # 使用 generate_response 函数来迭代响应内容，实现流式传输
+                def generate_response():
+                    for chunk in resp.iter_content(chunk_size=8192):
+                        yield chunk
+
+                response = Response(generate_response(), status=resp.status_code, headers=response_headers)
+                return response
+
+            except requests.exceptions.RequestException as e:
+                print(f"代理请求失败: {e}")
+                abort(500, description=f"代理请求到 {target_url} 失败: {e}")
+
+    def run(self):
+        print(f"代理服务器正在 {self.host}:{self.port} 上启动")
+        # Flask 默认在开发模式下会为每个请求创建一个新的线程，实现并发处理。
+        # 在生产环境中，建议使用 Gunicorn 或 uWSGI 等 WSGI 服务器来管理多线程/多进程。
+        self.app.run(host=self.host, port=self.port, debug=True) # 将 debug 模式设置为 True
+
+if __name__ == '__main__':
+    # 提示：请确保您已激活 conda 环境 'any-api' (conda activate any-api)
+    # 提示：请确保已安装 Flask 和 requests 库 (pip install Flask requests)
+    proxy = ProxyServer(
+        host='0.0.0.0',
+        port=7860
+    )
+    proxy.run()

push.sh

@@ -0,0 +1,3 @@
+git add .
+git commit -m "update"
+git push

readme.md

@@ -0,0 +1,102 @@
+# AnyAPI 代理服务器
+
+本项目旨在实现一个多线程的 Python 代理服务器，用于中转 HTTP 请求到各种兼容 OpenAI 接口的外部 API。它提供了一个统一的访问入口，并能够处理协议转换和简化客户端配置。
+
+## 1. 项目简介
+
+当用户访问中转主机 URL `http://127.0.0.1:7860/v1/{protocol}/{domain}/{path}` 时，代理服务器会将其解析并转发到实际的目标 URL `protocol://domain/path`。例如，`http://127.0.0.1:7860/v1/https/open.bigmodel.cn/api/paas/v4/chat/completions` 会被代理到 `https://open.bigmodel.cn/api/paas/v4/chat/completions`。
+
+## 2. 项目结构
+
+*   `proxyserver-fastapi.py`: 基于 FastAPI 实现的代理服务器核心逻辑。它负责接收客户端请求，解析代理 URL，转发请求到目标 API，并将响应返回给客户端。支持多线程并发处理。
+*   `agent_example.py`: 一个 Python 客户端示例，演示如何通过代理服务器调用兼容 OpenAI 接口的 API。它从 `.env` 文件中读取配置信息。
+*   `requirements.txt`: 项目所需的 Python 依赖库列表。
+*   `.env`: 环境变量配置文件，用于 `agent_example.py` 的配置。
+
+## 3. 技术栈
+
+*   **Python 3.12**
+*   **FastAPI**: 用于构建高性能的异步 API。
+*   **httpx**: 异步 HTTP 客户端，用于向目标 API 发送请求。
+*   **uvicorn**: ASGI 服务器，用于运行 FastAPI 应用。
+*   **requests**: (在 `agent_example.py` 中使用) 同步 HTTP 客户端。
+*   **python-dotenv**: 用于从 `.env` 文件加载环境变量。
+*   **Conda**: 环境管理工具。
+
+## 4. 环境设置
+
+1.  **创建并激活 Conda 环境**:
+    ```bash
+    conda create -n any-api python=3.12
+    conda activate any-api
+    ```
+
+2.  **安装依赖**:
+    ```bash
+    pip install -r requirements.txt
+    ```
+
+## 5. 配置说明
+
+`agent_example.py` 通过 `.env` 文件读取配置。请确保 `.env` 文件存在于项目根目录，并包含以下变量：
+
+```dotenv
+# 代理服务器的基地址
+# 示例: http://localhost:7860/v1/https/generativelanguage.googleapis.com/v1beta/openai
+PROXY_BASE_URL="http://localhost:7860/v1/https/generativelanguage.googleapis.com/v1beta/openai"
+
+# 目标 API 的入口点（相对于目标域名的路径）
+# 示例: /chat/completions
+TARGET_API_PATH="/chat/completions"
+
+# 您的 API 密钥
+YOUR_API_KEY="YOUR_ACTUAL_API_KEY_HERE"
+
+# 使用的模型名称
+MODEL_NAME="gemini-2.5-flash-preview-05-20"
+```
+
+请根据您的实际需求修改这些值。
+
+## 6. 如何运行
+
+### 6.1 启动代理服务器
+
+在激活 `any-api` Conda 环境后，运行 `proxyserver-fastapi.py`：
+
+```bash
+uvicorn proxyserver-fastapi:app --host 0.0.0.0 --port 7860 --reload
+```
+或者直接运行文件（如果文件中包含 `if __name__ == '__main__':` 块）：
+```bash
+python proxyserver-fastapi.py
+```
+服务器将在 `http://0.0.0.0:7860` 上监听请求。
+
+### 6.2 运行客户端示例
+
+在代理服务器运行后，您可以在另一个终端中运行 `agent_example.py` 来测试代理功能：
+
+```bash
+python agent_example.py
+```
+
+这将向代理服务器发送一个示例请求，并通过代理转发到配置的目标 API。
+
+## 7. URL 格式说明
+
+代理服务器期望的 URL 格式为：`http://<proxy_host>:<proxy_port>/v1/{protocol}/{domain}/{path}`
+
+*   `<proxy_host>:<proxy_port>`: 代理服务器的地址和端口，例如 `127.0.0.1:7860`。
+*   `v1`: 版本标识符。
+*   `{protocol}`: 目标 API 的协议（例如 `http` 或 `https`）。
+*   `{domain}`: 目标 API 的域名（例如 `open.bigmodel.cn`）。
+*   `{path}`: 目标 API 的具体路径（例如 `/api/paas/v4/chat/completions`）。
+
+代理服务器会将此 URL 转换为 `protocol://domain/path` 并进行转发。
+
+## 8. 注意事项
+
+*   **API 密钥安全**: 永远不要将 API 密钥硬编码到代码中。本项目使用 `.env` 文件进行管理，请确保 `.env` 文件不被提交到版本控制系统（已在 `.gitignore` 中配置）。
+*   **错误处理**: 代理服务器包含了基本的错误处理机制，但生产环境中可能需要更健壮的日志记录和监控。
+*   **性能**: FastAPI 和 httpx 提供了异步能力，有助于提高并发性能。Python GIL 对 CPU 密集型任务有影响，但对于 I/O 密集型代理任务影响较小。

requirements.txt

@@ -0,0 +1,6 @@
+flask
+httpx
+fastapi
+uvicorn
+requests
+python-dotenv