FastAPI: проксирование потока ответа Claude через SSE

Ответ модели на длинный вопрос генерируется десятки секунд, и показывать пользователю пустой экран всё это время — плохая идея. Решение — стриминг: FastAPI получает ответ от API по частям и сразу пересылает их в браузер через Server-Sent Events. Ключ Zukko при этом остаётся на сервере. Ниже — рабочий бэкенд на AsyncOpenAI, два варианта фронтенда, обработка ошибок и разрыва соединения.

Бэкенд: стриминг ответа в FastAPI

Зависимости: pip install fastapi uvicorn openai. Ключ — в переменной окружения ZUKKO_API_KEY.

import json
import os

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI, APIConnectionError, APIStatusError
from pydantic import BaseModel

app = FastAPI()

client = AsyncOpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key=os.environ["ZUKKO_API_KEY"],
    timeout=120.0,

Запуск: uvicorn main:app --reload. Что здесь важно:

• base_url обязательно с /v1 — это OpenAI-совместимый адрес Zukko.
• Проверка if chunk.choices нужна: сервер может прислать последний чанк со статистикой и пустым массивом choices, и обращение к chunk.choices[0] упадёт с IndexError.
• Каждое событие — строка data: ... и пустая строка после неё. В JSON кладём только текст, а не весь чанк.
• Заголовок X-Accel-Buffering: no просит nginx не копить ответ в буфере. Без него за прокси текст придёт одним куском в конце.
• Два эндпоинта: GET нужен для EventSource (он умеет только GET), POST — для fetch, когда сообщение длинное или не должно попадать в URL и логи.

Фронтенд: EventSource

Самый короткий вариант для браузера:

<textarea id="input"></textarea>
<button id="send">Отправить</button>
<pre id="output"></pre>

<script>
  document.getElementById('send').addEventListener('click', () => {
    const message = document.getElementById('input').value;
    const output = document.getElementById('output');
    output.textContent = '';

    const source = new EventSource('/chat/stream?message=' + encodeURIComponent(message));

    source.onmessage = (event) => {
      if (event.data === '[DONE]') {

Два момента. Текст добавляется через textContent, а не innerHTML: ответ модели — это недоверенные данные, и вставлять их как HTML нельзя. И EventSource при обрыве соединения сам переподключается, то есть повторно отправит запрос к модели и вы заплатите за него ещё раз. Поэтому соединение закрываем и по [DONE], и в onerror.

Фронтенд: fetch с POST

fetch позволяет отправить сообщение в теле POST-запроса и добавить свои заголовки, например токен вашего приложения:

<script>
  async function sendMessage(message, output) {
    const response = await fetch('/chat/stream', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ message }),
    });
    if (!response.ok) {
      output.textContent = 'Ошибка ' + response.status;
      return;
    }

    const reader = response.body.getReader();
    const decoder = new TextDecoder();

Сетевые куски не совпадают с границами событий, поэтому данные копятся в buffer и разбираются только целыми событиями (они разделены пустой строкой). Неполный хвост остаётся до следующего чтения.

Разрыв соединения и лимиты

Когда пользователь закрывает вкладку, Starlette (на нём построен FastAPI) отменяет генератор: внутри generate возникает asyncio.CancelledError. Блок async with stream закрывает соединение с API, и оно не висит до конца генерации. Не ловите CancelledError без повторного raise и не делайте yield в finally — отмена должна дойти до конца.

Если нужно ограничить, сколько потоков одновременно идёт к модели, оберните тело генератора в семафор:

import asyncio

limiter = asyncio.Semaphore(20)


async def limited_generate(message: str):
    async with limiter:
        async for event in generate(message):
            yield event

и передавайте в StreamingResponse уже limited_generate(...). Остальные запросы подождут свободного слота, а не получат 429 от API. Дневной и месячный лимит расходов на ключ можно задать в кабинете — это защита от случайного цикла на фронтенде.

Логирование токенов

Чтобы знать реальный расход, попросите сервер прислать статистику в конце потока параметром stream_options:

import logging

logger = logging.getLogger("chat")


async def generate_with_usage(message: str):
    stream = await client.chat.completions.create(
        model="claude-sonnet-5",
        messages=[{"role": "user", "content": message}],
        stream=True,
        stream_options={"include_usage": True},
    )
    usage = None
    async with stream:
        async for chunk in stream:
            if chunk.usage:
                usage = chunk.usage
            if chunk.choices and chunk.choices[0].delta.content:
                yield sse({"text": chunk.choices[0].delta.content})
    if usage:
        logger.info("tokens: in=%s out=%s", usage.prompt_tokens, usage.completion_tokens)
    yield "data: [DONE]\n\n"

Статистика приходит в отдельном последнем чанке с пустым choices — ещё одна причина проверять массив. Стоимость считается по ценам модели: у claude-sonnet-5 $0.3 за 1M входных и $0.3 за 1M выходных токенов. Ответ на 800 токенов при вопросе на 200 токенов стоит (200 + 800) / 1 000 000 × 0,3 = $0.0003.

Частые вопросы

Где хранить API-ключ?

Только на сервере: в переменной окружения или в .env, который не попадает в git. В браузер ключ не передавайте никогда — его можно достать из кода страницы за минуту. Создать ключ можно в кабинете↗; для веб-приложения удобно сделать отдельный ключ с дневным лимитом.

Почему текст приходит одним куском, а не по частям?

Чаще всего ответ буферизует прокси перед приложением. Проверьте заголовок X-Accel-Buffering: no и настройки буферизации nginx, а также не включено ли сжатие ответа для text/event-stream. Локально, без прокси, uvicorn отдаёт поток сразу.

Почему в браузере «Ошибка соединения»?

Смотрите логи сервера: там видны ответы API с кодами 401, 402, 429 и 503. Если фронтенд и бэкенд на разных доменах, настройте CORS через CORSMiddleware. Если запрос обрывается ровно через минуту, это таймаут прокси — увеличьте его для маршрута стриминга.

Как сделать то же самое без стриминга?

Уберите stream=True и верните response.choices[0].message.content обычным JSON. Примеры синхронного и асинхронного клиента — в документации по Python↗, параметры запроса — в описании Chat Completions↗.