FastAPI: proxying Claude stream response via SSE

A model's answer to a long question takes tens of seconds to generate, and showing the user a blank screen all that time is a bad idea. The fix is streaming: FastAPI receives the response from the API in pieces and forwards them to the browser right away via Server-Sent Events. Your Zukko key stays on the server. Below is a working AsyncOpenAI backend, two frontend options, error handling, and disconnect handling.

Backend: streaming a response in FastAPI

Dependencies: pip install fastapi uvicorn openai. The key goes in the ZUKKO_API_KEY environment variable.

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,

Run it with uvicorn main:app --reload. What matters here:

• base_url must include /v1 — that's Zukko's OpenAI-compatible address.
• The if chunk.choices check is required: the server may send a final chunk with statistics and an empty choices array, and chunk.choices[0] would raise IndexError.
• Each event is a data: ... line followed by a blank line. We put only the text into the JSON, not the whole chunk.
• The X-Accel-Buffering: no header asks nginx not to buffer the response. Without it, behind a proxy the text arrives in one piece at the end.
• Two endpoints: GET is for EventSource (it only supports GET), POST is for fetch, when the message is long or shouldn't end up in URLs and logs.

Frontend: EventSource

The shortest browser option:

<textarea id="input"></textarea>
<button id="send">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]') {

Two things to note. Text is appended via textContent, not innerHTML: model output is untrusted data and must not be inserted as HTML. And when the connection drops, EventSource reconnects by itself, which means it re-sends the request to the model and you pay for it again. That's why we close the connection both on [DONE] and in onerror.

Frontend: fetch with POST

fetch lets you send the message in a POST body and add your own headers, such as your app's auth token:

<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 = 'Error ' + response.status;
      return;
    }

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

Network chunks don't line up with event boundaries, so data accumulates in buffer and is parsed only as whole events (separated by a blank line). An incomplete tail waits for the next read.

Disconnects and limits

When the user closes the tab, Starlette (which FastAPI is built on) cancels the generator: asyncio.CancelledError is raised inside generate. The async with stream block closes the connection to the API, so it doesn't hang until generation ends. Don't catch CancelledError without re-raising it, and don't yield in a finally block — cancellation must run to completion.

To cap how many streams hit the model at once, wrap the generator in a semaphore:

import asyncio

limiter = asyncio.Semaphore(20)


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

and pass limited_generate(...) to StreamingResponse. Other requests will wait for a free slot instead of getting a 429 from the API. You can also set daily and monthly spend limits on the key in the dashboard — a safeguard against an accidental loop in the frontend.

Logging tokens

To know your actual usage, ask the server to send statistics at the end of the stream with 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"

The statistics arrive in a separate final chunk with empty choices — one more reason to check the array. Cost follows the model's prices: claude-sonnet-5 is $0.3 per 1M input and $0.3 per 1M output tokens. An 800-token answer to a 200-token question costs (200 + 800) / 1,000,000 × 0.3 = $0.0003.

FAQ

Where should I keep the API key?

Only on the server: in an environment variable or a .env file that stays out of git. Never send the key to the browser — anyone can pull it from the page source in a minute. Create a key in the dashboard↗; a separate key with a daily limit for the web app is convenient.

Why does the text arrive in one piece instead of streaming?

Usually a proxy in front of the app is buffering the response. Check the X-Accel-Buffering: no header and nginx buffering settings, and whether response compression is enabled for text/event-stream. Locally, without a proxy, uvicorn sends the stream immediately.

Why does the browser show a connection error?

Check the server logs: API responses with codes 401, 402, 429, and 503 show up there. If the frontend and backend are on different domains, configure CORS with CORSMiddleware. If the request drops after exactly one minute, it's a proxy timeout — raise it for the streaming route.

How do I do the same without streaming?

Drop stream=True and return response.choices[0].message.content as regular JSON. Sync and async client examples are in the Python docs↗, and request parameters are in the Chat Completions reference↗.