Стриминг Claude API через SSE: TypeScript и Python

Если продукт на Claude кажется медленным, дело почти никогда не в общей пропускной способности. Тормозит время до первого токена (TTFT). Обычный ответ на 600 токенов, который приходит целиком через 8 секунд, ощущается как сбой. Те же 600 токенов в стриме выглядят мгновенными: первые слова появляются меньше чем за 400 мс, остальное набирается на глазах. Модель быстрее не становится. Зато воспринимаемая латентность падает примерно вдвое относительно реальной, а пользователи реагируют именно на неё.

Разберём, как Claude отдаёт ответ через Server-Sent Events (SSE), какие события Anthropic нужно обрабатывать и как выглядят рабочие паттерны на TypeScript (Next.js Edge runtime) и Python (httpx async). Во всех примерах используется Anthropic-совместимый шлюз codexkey, https://api.codexkey.ru/v1. Он ставится вместо api.anthropic.com/v1 без переделки клиентского кода.

TTFT и полная латентность

При замере вызова Claude записывайте две цифры:

• TTFT: миллисекунды до первой content-дельты.
• Total: миллисекунды до message_stop.

Для Sonnet, который генерирует около 800 выходных токенов, TTFT обычно укладывается в 300-600 мс, а Total занимает 4-8 секунд. Без стриминга пользователь смотрит на спиннер все 8 секунд. Со стримингом текст появляется через 400 мс, и его можно читать по ходу генерации. Скорость в токенах в секунду (TPS) в обоих случаях одинакова. Вы покупаете ощущение скорости, а не пропускную способность.

Как устроен формат SSE

Server-Sent Events: однонаправленный потоковый протокол поверх HTTP. Ответ приходит с Content-Type: text/event-stream, а тело состоит из записей, которые отделены друг от друга пустыми строками. Одна запись выглядит так:

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Привет"}}

Два правила, о которые спотыкаются все:

1. Запись заканчивается двумя переводами строки (\n\n), а не одним.
2. Одно логическое событие может состоять из нескольких строк data:. Потребитель склеивает их через \n и только потом разбирает JSON.

Большинство SDK прячут эти детали. Но когда вы разбираете SSE руками (в Edge Worker, на Go или при отладке), правила придётся соблюдать самому.

События стрима Anthropic

Стрим Claude отдаёт небольшой и чётко определённый набор событий. Обрабатывайте каждое явно и не рассчитывайте на порядок сверх того, что гарантирует спецификация.

• message_start: стартовый конверт сообщения с id, model и обнулённым блоком usage. Id сообщения удобно сохранить здесь, для логов.
• content_block_start: начался новый блок контента. Тип блока: text, tool_use, thinking или redacted_thinking. Если модель выдаёт несколько блоков, важен индекс.
• content_block_delta: инкрементальная нагрузка. У текста это text_delta, у вызовов инструментов input_json_delta (частичные фрагменты JSON, которые нужно склеить до разбора).
• content_block_stop: блок закончился.
• message_delta: обновления сообщения целиком, прежде всего stop_reason (end_turn, max_tokens, tool_use, stop_sequence) и финальные счётчики usage.
• message_stop: завершающее событие. Здесь читателя закрывают.
• ping: keep-alive примерно раз в 15 секунд. Содержимое игнорируйте, но поток не закрывайте. Пинги нужны затем, чтобы reverse proxy не оборвал простаивающее соединение.
• error: да, ошибка может прийти посреди стрима обычным SSE-событием (перегруженная модель, остановка по content policy, upstream timeout). Обработчик должен относиться к error-как-событию так же, как к брошенному исключению.

TypeScript: Next.js Edge Route Handler

Чище всего для Next.js получается Edge Route Handler, который проксирует стрим Claude прямо в браузер. Edge runtime даёт нативный ReadableStream и не берёт налог на холодный старт.

// app/api/chat/route.ts
import Anthropic from "@anthropic-ai/sdk";

export const runtime = "edge";

const client = new Anthropic({
  apiKey: process.env.CODEXKEY_API_KEY!,
  baseURL: "https://api.codexkey.ru/v1",
});

export async function POST(req: Request) {
  const { messages } = await req.json();
  const controller = new AbortController();
  req.signal.addEventListener("abort", () => controller.abort());

  const stream = await client.messages.stream(
    {
      model: "claude-sonnet-5",
      max_tokens: 1024,
      messages,
    },
    { signal: controller.signal },
  );

  const encoder = new TextEncoder();
  const body = new ReadableStream({
    async start(ctrl) {
      try {
        for await (const event of stream) {
          if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
            ctrl.enqueue(encoder.encode(event.delta.text));
          }
        }
      } catch (err) {
        ctrl.enqueue(encoder.encode(`\n[error] ${(err as Error).message}`));
      } finally {
        ctrl.close();
      }
    },
    cancel() {
      controller.abort();
    },
  });

  return new Response(body, {
    headers: {
      "Content-Type": "text/plain; charset=utf-8",
      "Cache-Control": "no-cache, no-transform",
      "X-Accel-Buffering": "no",
    },
  });
}

Три мелочи, на которых обжигаются в продакшне:

• X-Accel-Buffering: no выключает буферизацию ответа в nginx. Без него стрим собирается в один кусок, и TTFT откатывается к значению для нестримингового ответа.
• Cache-Control: no-transform запрещает посредникам сжимать ответ gzip и перерезать его на другие чанки.
• Привязка req.signal к AbortController позволяет отменить upstream-вызов Claude, когда вкладка браузера закрыта. Без неё вы продолжаете платить за токены, которые никто не прочтёт.

Python: async-стриминг на httpx

Для бэкенд-сервисов официальный SDK anthropic уже умеет стримить. Когда нужен контроль на уровень ниже (свой прокси, broadcast fan-out, инструментирование), спускайтесь к httpx.

import json
import httpx

URL = "https://api.codexkey.ru/v1/messages"

async def stream_claude(prompt: str, api_key: str):
    headers = {
        "x-api-key": api_key,
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
        "accept": "text/event-stream",
    }
    payload = {
        "model": "claude-sonnet-5",
        "max_tokens": 1024,
        "stream": True,
        "messages": [{"role": "user", "content": prompt}],
    }

    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream("POST", URL, headers=headers, json=payload) as resp:
            resp.raise_for_status()
            event_name = None
            async for line in resp.aiter_lines():
                if not line:
                    event_name = None
                    continue
                if line.startswith("event: "):
                    event_name = line[7:].strip()
                elif line.startswith("data: "):
                    data = json.loads(line[6:])
                    if event_name == "content_block_delta":
                        delta = data.get("delta", {})
                        if delta.get("type") == "text_delta":
                            yield delta["text"]
                    elif event_name == "error":
                        raise RuntimeError(data.get("error", {}).get("message", "stream error"))
                    elif event_name == "message_stop":
                        return

Что здесь важно:

• timeout=None у клиента обязателен. Read timeout по умолчанию в 5 секунд убьёт любой стрим длиннее пяти секунд.
• aiter_lines() сам разбирается с \r\n и \n.
• Проверка пустой строки сбрасывает event_name, и это не случайность: записи SSE разделены пустой строкой, а в следующей может не оказаться поля event: (тогда по умолчанию тип message).

Backpressure и отмена

Claude способен выдавать токены быстрее, чем нижестоящий потребитель успевает их записывать: в браузер, в базу, в другой сервис. Если backpressure не учитывать, рантайм копит байты в памяти, и под нагрузкой приходит OOM.

В Node ReadableStream со стандартной стратегией очереди byteLength даёт backpressure без усилий. В Python используйте async for и не собирайте результат в список. Для broadcast fan-out (один стрим Claude на много WebSocket-клиентов) заведите ограниченную asyncio.Queue на каждого клиента и отбрасывайте медленного клиента, но никогда не upstream.

Вторая половина задачи, отмена. Когда пользователь закрыл вкладку, upstream-запрос нужно прервать:

• TypeScript: передайте request.signal в опцию signal SDK.
• Python: оберните цикл по телу в try/finally и положитесь на контекстный менеджер httpx, который закроет соединение. Либо явно вызовите await resp.aclose(), когда upstream-клиент отключился.

Вы платите за каждый токен, который сгенерировала модель, в том числе за тот, который никто не прочитает. Поэтому отмена работает ещё и как контроль себестоимости, помимо удобства для интерфейса.

Частые ошибки

• Забыт flush. Express, Fastify и любой самописный Node-фреймворк по умолчанию буферизуют запись. Выставьте Content-Type: text/event-stream, Cache-Control: no-cache, X-Accel-Buffering: no и сразу вызовите res.flushHeaders().
• Нет пингов, и прокси обрывает соединение через 60 секунд. Cloudflare, nginx и большинство ALB закрывают простаивающие HTTP-соединения через 60 секунд. Пинги Claude держат соединение живым. Если вы их отфильтровали и модель долго думает перед началом текста, прокси вас отключит. Пробрасывайте пинги или отправляйте собственные keep-alive комментарии (: keepalive\n\n) каждые 15 секунд.
• JSON для tool use склеен неверно. Куски input_json_delta это частичные фрагменты JSON. Копите строку и вызывайте JSON.parse только после content_block_stop для этого блока.
• Ошибка посреди стрима принята за транспортную. Событие error это обычная SSE-запись, а не HTTP-сбой. Читатель сам не бросит исключение: тип события нужно проверить и поднять ошибку вручную.
• Дельты логируются синхронно. console.log на каждый text_delta роняет пропускную способность. Буферизуйте логи и выводите их на message_stop.

Что в итоге

Из всего, что можно изменить в интеграции с Claude, стриминг даёт самый большой выигрыш в UX на единицу усилий. Берите SDK там, где он подходит, и спускайтесь к голому SSE, когда без этого не обойтись. Запомните три вещи: обрабатывайте пинги, отменяйте запрос при дисконнекте и не доверяйте таймаутам прокси по умолчанию. Направьте base_url на https://api.codexkey.ru/v1, оставьте код на Anthropic SDK как есть и выкатывайте.