AI-эндпоинт на FastAPI через AI Gateway: async и стриминг
Как добавить AI-эндпоинт в FastAPI: асинхронный клиент OpenAI SDK с base_url AI Gateway, обычный ответ и стриминг через StreamingResponse, ключ в переменных окружения.
Соберём в FastAPI пару эндпоинтов, которые обращаются к моделям из каталога AI Gateway: один отдаёт готовый ответ, другой стримит его по мере генерации. FastAPI асинхронный, поэтому берём AsyncOpenAI — он совместим с шлюзом на уровне base_url и не блокирует event loop.
Подготовка
Нужен ключ sk-aigate-… из кабинета и базовый URL https://ai-gatewey.ru/v1. Ключ держим в переменной окружения, а не в коде.
pip install fastapi uvicorn openaiРеализация
- Настройте асинхронный клиент
Один экземпляр
AsyncOpenAIна всё приложение — создаём при старте и переиспользуем во всех обработчиках:pythonpythonimport osfrom openai import AsyncOpenAIclient = AsyncOpenAI(base_url="https://ai-gatewey.ru/v1",api_key=os.environ["AIGATE_API_KEY"],) - Эндпоинт с обычным ответом
Принимаем текст в теле запроса, отдаём ответ модели одним JSON:
pythonpythonfrom fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI()class Prompt(BaseModel):message: str@app.post("/chat")async def chat(body: Prompt):resp = await client.chat.completions.create(model="deepseek/deepseek-v4-flash",messages=[{"role": "user", "content": body.message}],)return {"reply": resp.choices[0].message.content} - Стриминг через StreamingResponse
Для «печатающегося» ответа включаем
stream=Trueи отдаём чанки генератором вStreamingResponse:pythonpythonfrom fastapi.responses import StreamingResponse@app.post("/chat/stream")async def chat_stream(body: Prompt):async def gen():stream = await client.chat.completions.create(model="deepseek/deepseek-v4-flash",messages=[{"role": "user", "content": body.message}],stream=True,)async for chunk in stream:delta = chunk.choices[0].delta.contentif delta:yield deltareturn StreamingResponse(gen(), media_type="text/plain") - Запустите сервер
Поднимаем uvicorn и проверяем эндпоинты; расход по токенам виден в кабинете AI Gateway в реальном времени:
bashbashuvicorn main:app --reload
AIGATE_API_KEY), в секретах контейнера или менеджере секретов. Никогда не передавайте его на фронтенд: клиент должен ходить в ваш FastAPI, а тот уже в шлюз.Какие модели подойдут
Для бэкенд-эндпоинтов под нагрузкой обычно берут быструю недорогую модель, а флагман подключают точечно там, где нужен максимум качества:
Частые вопросы
Почему AsyncOpenAI, а не обычный OpenAI?
Обработчики FastAPI выполняются в общем event loop. Синхронный клиент блокировал бы его на время запроса к модели и снижал пропускную способность; AsyncOpenAI с await этого избегает.
Как отдать стрим на фронтенд Server-Sent Events?
Замените media_type на text/event-stream и оборачивайте каждый чанк в формат SSE (data: …\n\n). Клиент читает его через EventSource или fetch с ReadableStream.
Что вернётся при ошибке модели или лимите?
SDK бросит исключение (RateLimitError, APIStatusError) с кодом 429, 402 или 401. Ловите его в обработчике и возвращайте клиенту корректный HTTP-статус вместо 500.