support@serv.host
Как поднять свой API-шлюз и использовать его под свои проекты
Как поднять свой API-шлюз и использовать его под свои проекты
В 26-м году почти каждый, кто пилит ботов, сервисы или пет-проекты на ИИ, упирается в одну и ту же проблему: если нужны модели разных провайдеров, то нужно, чтобы софт поддерживал несколько API-ключей. Еще и большинство провайдеров нормально не работают из России.
Поднять свой API-шлюз (агрегатор) — самый простой способ решить эту проблему. Логика элементарная: вы выдаёте своим скриптам и проектам один нормальный эндпоинт и проксируете запросы на любые внешние API.
Вуаля. Теперь ваш софт может использовать модели разных провайдеров одновременно, имея один API-ключ.
На чём тут зарабатывать?
Схема монетизации простая: маржа на курсовой разнице. $1 в стоит условно 75 рублей. Ты продаёшь клиентам внутренние токены из расчёта 78 рублей за $1 эквивалента.
Однако: продажа доступа и коммерческий агрегатор запускаются исключительно на ваш свой страх и риск — из-за блокировок аккаунтов со стороны провайдеров, чарджбэков и банковских ограничений. Основной сценарий использования этой системы — личное использование и инфраструктура своих проектов.
И да, тут абсолютно не нужны сервера с GPU. Все это спокойно крутится на базовом VPS за 400 рублей, потому что на сервере нет весов — через него тупо летает JSON.
Все возможности библиотеки OpenAI
Перед тем как писать шлюз, разберём возможности официального SDK ==openai==, на котором строится взаимодействие:
- Произвольный ==base_url== и ключи: Можно перенаправить вызовы на любой совместимый сервис (OpenCode Zen, DeepSeek, OpenAI).
- Асинхронность (==AsyncOpenAI==): Высокая скорость обработки без блокировки потока (Event Loop).
- Стриминг (==stream=True==): Возврат ответа по токенам через Server-Sent Events (SSE).
- Управление уровнями рассуждений (Reasoning Effort): Передача параметров мышления (==low==, ==medium==, ==high==) для reasoning-моделей.
- Динамический список моделей (==client.models.list()==): Получение доступных моделей в стандарте ==/v1/models==.
Зависимости проекта (==requirements.txt==)
fastapi>=0.110.0
uvicorn[standard]>=0.28.0
openai>=1.14.0
httpx>=0.27.0
python-dotenv>=1.0.1
aiosqlite>=0.20.0
tiktoken>=0.6.0
Конфигурация окружения (==.env==)
# Настройки сервера
PORT=2091
# Провайдер OpenCode Zen (бесплатный провайдер с публичным ключом)
OPENCODE_ZEN_BASE_URL=https://api.opencode.zen/v1
OPENCODE_ZEN_API_KEY=public
# Дополнительные провайдеры
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxxxxxx
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1
DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxxxxxx
Конфиг провайдеров (==providers.json==)
Модели описаны объектом, а не массивом: у каждой указано, поддерживает ли она reasoning, и какие уровни рассуждения допустимы. Шлюз сам вырезает ==reasoning_effort== / ==thinking== для моделей, которые их не понимают, — клиентские приложения не получают 400 от апстрима.
{
"providers": {
"opencode": {
"base_url": "https://opencode.ai/zen/v1",
"api_key": "public"
}
}
"models": {
"mimo-v2.5": {
"provider": "opencode",
"upstream": "mimo-v2.5-free",
"supports_reasoning": false
},
"deepseek-v4": {
"provider": "opencode",
"upstream": "deepseek-v4-flash-free",
"supports_reasoning": true,
"allowed_efforts": ["high", "max"]
}
}
}
Конфигурация Nginx (==nginx.conf==)
Nginx принимает внешние SSL-запросы на ==https://domain.com/v1/== и проксирует их на ==localhost:2091==, отключая буферизацию для бесшовной работы SSE-стриминга.
server {
listen 80;
server_name domain.com;
return 301 https://$host$request_uri;
}
server {
listen 443 ssl http2;
server_name domain.com;
ssl_certificate /etc/letsencrypt/live/domain.com/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/domain.com/privkey.pem;
proxy_connect_timeout 600s;
proxy_send_timeout 600s;
proxy_read_timeout 600s;
location /v1/ {
proxy_pass http://127.0.0.1:2091/v1/;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_buffering off;
proxy_cache off;
chunked_transfer_encoding on;
}
}
Модуль базы данных (==db.py==)
База данных SQLite содержит ровно две колонки: ==api_key== и ==used_tokens==.
import aiosqlite
DB_PATH = "gateway.db"
async def init_db():
async with aiosqlite.connect(DB_PATH) as db:
await db.execute("""
CREATE TABLE IF NOT EXISTS keys (
api_key TEXT PRIMARY KEY,
used_tokens INTEGER DEFAULT 0
)
""")
await db.commit()
async def verify_key(api_key: str) -> bool:
async with aiosqlite.connect(DB_PATH) as db:
async with db.execute("SELECT 1 FROM keys WHERE api_key = ?", (api_key,)) as cursor:
row = await cursor.fetchone()
return row is not None
async def add_key(api_key: str) -> bool:
try:
async with aiosqlite.connect(DB_PATH) as db:
await db.execute("INSERT INTO keys (api_key, used_tokens) VALUES (?, 0)", (api_key,))
await db.commit()
return True
except aiosqlite.IntegrityError:
return False
async def update_token_usage(api_key: str, tokens_count: int):
async with aiosqlite.connect(DB_PATH) as db:
await db.execute(
"UPDATE keys SET used_tokens = used_tokens + ? WHERE api_key = ?",
(tokens_count, api_key)
)
await db.commit()
async def get_all_keys():
async with aiosqlite.connect(DB_PATH) as db:
async with db.execute("SELECT api_key, used_tokens FROM keys") as cursor:
return await cursor.fetchall()
async def delete_key(api_key: str):
async with aiosqlite.connect(DB_PATH) as db:
await db.execute("DELETE FROM keys WHERE api_key = ?", (api_key,))
await db.commit()
Панель управления ключами в консоли (==admin_cli.py==)
Скрипт генерирует новые ключи, удаляет существующие и показывает баланс токенов из БД.
import asyncio
import secrets
import db
async def main():
await db.init_db()
while True:
print("\n= ПАНЕЛЬ УПРАВЛЕНИЯ КЛЮЧАМИ =")
print("1. Посмотреть список ключей и затраченные токены")
print("2. Создать новый API-ключ")
print("3. Удалить API-ключ")
print("0. Выход")
choice = input("\nВыберите действие: ").strip()
if choice == "1":
keys = await db.get_all_keys()
print("\n--- Ключи в базе (api_key | used_tokens) ---")
if not keys:
print("База пуста.")
for k, t in keys:
print(f"Ключ: {k} | Затрачено токенов: {t}")
elif choice == "2":
custom = input("Введите свой ключ (или Enter для генерации): ").strip()
new_k = custom if custom else f"sk-gw-{secrets.token_hex(12)}"
if await db.add_key(new_k):
print(f"\nУспешно создан ключ: {new_k}")
else:
print("\nОшибка: Такой ключ уже существует.")
elif choice == "3":
del_k = input("Введите ключ для удаления: ").strip()
await db.delete_key(del_k)
print(f"\nКлюч {del_k} удален.")
elif choice == "0":
break
else:
print("Неверная команда.")
if __name__ == "__main__":
asyncio.run(main())
Главный файл прокси-сервера (main.py)
Загружает провайдеров из providers.json, реализует эндпоинты /v1/models и /v1/chat/completions, обрабатывает уровни рассуждения (reasoning_effort) только для моделей, которые их поддерживают, и ведет подсчет токенов при стриминге.
import os
import json
from contextlib import asynccontextmanager
from typing import AsyncGenerator, Dict, Any
from fastapi import FastAPI, Request, HTTPException, Depends, Header
from fastapi.responses import StreamingResponse, JSONResponse
from openai import AsyncOpenAI
import tiktoken
from dotenv import load_dotenv
import db
load_dotenv()
@asynccontextmanager
async def lifespan(app: FastAPI):
await db.init_db()
enc = tiktoken.get_encoding("cl100k_base")
app.state.enc = enc
yield
app = FastAPI(title="OpenAI-Compatible Gateway", lifespan=lifespan)
def load_config(file_path: str = "providers.json") -> Dict[str, Any]:
with open(file_path, "r", encoding="utf-8") as f:
raw = json.load(f)
providers = {}
for name, data in raw.get("providers", {}).items():
providers[name] = AsyncOpenAI(base_url=data["base_url"], api_key=data["api_key"])
models = raw.get("models", {})
return {"providers": providers, "models": models}
CONFIG = load_config()
async def authenticate(authorization: str = Header(None)) -> str:
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
token = authorization.replace("Bearer ", "").strip()
if not await db.verify_key(token):
raise HTTPException(status_code=401, detail="Invalid API Key")
return token
def count_tokens(text: str, enc=None) -> int:
if enc is None:
try:
enc = tiktoken.get_encoding("cl100k_base")
except Exception:
return len(text) // 4
try:
return len(enc.encode(text))
except Exception:
return len(text) // 4
def get_model_entry(model_name: str) -> Dict[str, Any]:
return CONFIG["models"].get(model_name, {})
def resolve_provider_and_upstream(model_name: str):
entry = get_model_entry(model_name)
if not entry:
return None, model_name
provider_name = entry.get("provider")
upstream = entry.get("upstream", model_name)
client = CONFIG["providers"].get(provider_name)
return client, upstream
@app.get("/health")
@app.get("/")
async def health():
return {"status": "ok", "service": "api-gateway"}
@app.get("/v1/models")
async def list_models(user_key: str = Depends(authenticate)):
data = []
for model_id, entry in CONFIG["models"].items():
data.append({
"id": model_id,
"object": "model",
"created": 1700000000,
"owned_by": entry.get("provider", "unknown"),
})
return {"object": "list", "data": data}
async def stream_proxy(response, user_key: str, prompt_tokens: int) -> AsyncGenerator[str, None]:
completion_text = ""
async for chunk in response:
chunk_json = chunk.model_dump_json()
yield f"data: {chunk_json}\n\n"
if chunk.choices and len(chunk.choices) > 0:
delta = chunk.choices[0].delta
if delta.content:
completion_text += delta.content
if hasattr(delta, "reasoning_content") and delta.reasoning_content:
completion_text += delta.reasoning_content
if hasattr(delta, "reasoning") and delta.reasoning:
completion_text += delta.reasoning
yield "data: [DONE]\n\n"
total_tokens = prompt_tokens + count_tokens(completion_text)
await db.update_token_usage(user_key, total_tokens)
@app.post("/v1/chat/completions")
async def chat_completions(request: Request, user_key: str = Depends(authenticate)):
body = await request.json()
client_model = body.get("model", "")
messages = body.get("messages", [])
stream = body.get("stream", False)
entry = get_model_entry(client_model)
if not entry:
raise HTTPException(status_code=400, detail=f"Unknown model: {client_model}")
upstream_model = entry.get("upstream", client_model)
client, _ = resolve_provider_and_upstream(client_model)
if not client:
raise HTTPException(status_code=500, detail=f"No provider for model: {client_model}")
reasoning_effort = body.get("reasoning_effort", None)
thinking = body.get("thinking", None)
kwargs: Dict[str, Any] = {
"model": upstream_model,
"messages": messages,
"stream": stream,
}
if entry.get("supports_reasoning"):
if reasoning_effort and reasoning_effort in entry.get("allowed_efforts", []):
kwargs["reasoning_effort"] = reasoning_effort
if thinking:
kwargs["extra_body"] = {"thinking": thinking}
prompt_str = "".join([m.get("content", "") for m in messages if isinstance(m.get("content"), str)])
enc = getattr(app.state, "enc", None)
prompt_tokens = count_tokens(prompt_str, enc)
try:
if stream:
response = await client.chat.completions.create(**kwargs)
return StreamingResponse(
stream_proxy(response, user_key, prompt_tokens),
media_type="text/event-stream",
)
else:
response = await client.chat.completions.create(**kwargs)
resp_dict = response.model_dump()
resp_dict["model"] = client_model
if response.usage:
total = response.usage.total_tokens
else:
comp_text = response.choices[0].message.content or ""
total = prompt_tokens + count_tokens(comp_text, enc)
await db.update_token_usage(user_key, total)
return JSONResponse(content=resp_dict)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Upstream error: {str(e)}")
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=int(os.getenv("PORT", 2091)))
Формат ответа JSON для /v1/models
Вот так выглядит структурированный JSON-ответ, который отдаёт сервер на запрос GET https://domain.com/v1/models для автоматического парсинга через openai.models.list():
{
"object": "list",
"data": [
{
"id": "mimo-v2.5-free",
"object": "model",
"created": 1700000000,
"owned_by": "opencode"
},
{
"id": "deepseek-v4-flash-free",
"object": "model",
"created": 1700000000,
"owned_by": "opencode"
}
}
Запуск и использование
- Запустите сервис FastAPI:
python main.py
- Откройте в соседнем окне панель управления:
python admin_cli.py
Создайте API-ключ (например, sk-mysecret).
- Подключите свой шлюз в Python-скриптах через библиотеку openai:
from openai import OpenAI
client = OpenAI(
base_url="https://domain.com/v1",
api_key="sk-mysecret"
)
# Получение списка моделей
print([m.id for m in client.models.list().data])
# Вызов с рассуждениями
response = client.chat.completions.create(
model="opencode-zen-free",
messages=[{"role": "user", "content": "Привет! Как дела?"}],
reasoning_effort="high"
)
print(response.choices[0].message.content)