LLM Streaming Responses with FastAPI and Anthropic SDK in Production

# pip install anthropic fastapi uvicorn sse-starlette
from anthropic import Anthropic
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse
import asyncio
import json
 
app = FastAPI()
client = Anthropic()
 
 
@app.get("/stream")
async def stream_response(prompt: str):
    """Stream Claude response as Server-Sent Events."""
 
    async def generate():
        with client.messages.stream(
            model="claude-sonnet-5",
            max_tokens=2000,
            messages=[{"role": "user", "content": prompt}]
        ) as stream:
            for text in stream.text_stream:
                # SSE format: data: {json}\n\n
                yield {
                    "event": "token",
                    "data": json.dumps({"text": text})
                }
 
        # Send completion event
        yield {
            "event": "done",
            "data": json.dumps({"status": "complete"})
        }
 
    return EventSourceResponse(generate())

import asyncio
import json
import logging
from anthropic import Anthropic, APIConnectionError, RateLimitError, APIStatusError
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import AsyncIterator
 
app = FastAPI()
client = Anthropic()
logger = logging.getLogger(__name__)
 
 
class ChatRequest(BaseModel):
    messages: list[dict]
    max_tokens: int = 2000
    system: str | None = None
 
 
async def stream_claude_response(request: ChatRequest) -> AsyncIterator[str]:
    """Yield SSE-formatted chunks from Claude streaming API."""
 
    try:
        kwargs = {
            "model": "claude-sonnet-5",
            "max_tokens": request.max_tokens,
            "messages": request.messages,
        }
        if request.system:
            kwargs["system"] = request.system
 
        with client.messages.stream(**kwargs) as stream:
            input_tokens = 0
            output_tokens = 0
 
            for event in stream:
                from anthropic.types import (
                    ContentBlockDeltaEvent,
                    MessageStartEvent,
                    MessageDeltaEvent,
                )
 
                if isinstance(event, MessageStartEvent):
                    input_tokens = event.message.usage.input_tokens
                    yield f"data: {json.dumps({'type': 'start', 'input_tokens': input_tokens})}\n\n"
 
                elif isinstance(event, ContentBlockDeltaEvent):
                    if event.delta.type == "text_delta":
                        yield f"data: {json.dumps({'type': 'token', 'text': event.delta.text})}\n\n"
 
                elif isinstance(event, MessageDeltaEvent):
                    output_tokens = event.usage.output_tokens
 
            # Send final usage stats
            yield f"data: {json.dumps({'type': 'done', 'output_tokens': output_tokens, 'total_tokens': input_tokens + output_tokens})}\n\n"
            yield "data: [DONE]\n\n"
 
    except RateLimitError as e:
        logger.error(f"Rate limit hit: {e}")
        error_data = json.dumps({"type": "error", "code": "rate_limit", "message": "Rate limit reached. Please try again shortly."})
        yield f"data: {error_data}\n\n"
 
    except APIConnectionError as e:
        logger.error(f"Connection error: {e}")
        error_data = json.dumps({"type": "error", "code": "connection", "message": "Connection interrupted. Please retry."})
        yield f"data: {error_data}\n\n"
 
    except APIStatusError as e:
        logger.error(f"API error {e.status_code}: {e.message}")
        error_data = json.dumps({"type": "error", "code": str(e.status_code), "message": str(e.message)})
        yield f"data: {error_data}\n\n"
 
 
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest, http_request: Request):
    """Stream chat response with client disconnect detection."""
 
    async def generate_with_disconnect_check():
        async for chunk in stream_claude_response(request):
            # Check if client disconnected
            if await http_request.is_disconnected():
                logger.info("Client disconnected, stopping stream")
                return
            yield chunk
 
    return StreamingResponse(
        generate_with_disconnect_check(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",    # Disable nginx buffering
        }
    )

// hooks/useStreamingChat.ts
import { useState, useCallback } from "react";
 
interface Message {
  role: "user" | "assistant";
  content: string;
}
 
export function useStreamingChat() {
  const [messages, setMessages] = useState<Message[]>([]);
  const [isStreaming, setIsStreaming] = useState(false);
  const [currentResponse, setCurrentResponse] = useState("");
 
  const sendMessage = useCallback(async (userMessage: string) => {
    const newMessages: Message[] = [
      ...messages,
      { role: "user", content: userMessage }
    ];
    setMessages(newMessages);
    setIsStreaming(true);
    setCurrentResponse("");
 
    const response = await fetch("/chat/stream", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ messages: newMessages }),
    });
 
    if (!response.body) return;
 
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let fullResponse = "";
 
    try {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
 
        const chunk = decoder.decode(value, { stream: true });
        const lines = chunk.split("\n");
 
        for (const line of lines) {
          if (!line.startsWith("data: ")) continue;
          const data = line.slice(6).trim();
          if (data === "[DONE]") continue;
 
          try {
            const parsed = JSON.parse(data);
            if (parsed.type === "token") {
              fullResponse += parsed.text;
              setCurrentResponse(fullResponse);
            }
          } catch {
            // Skip malformed chunks
          }
        }
      }
    } finally {
      setIsStreaming(false);
      setMessages(prev => [...prev, { role: "assistant", content: fullResponse }]);
      setCurrentResponse("");
    }
  }, [messages]);
 
  return { messages, isStreaming, currentResponse, sendMessage };
}

location /chat/stream {
    proxy_pass http://localhost:8000;
    proxy_buffering off;
    proxy_cache off;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding on;
}

# Increase read timeout for streaming
client = Anthropic(timeout=120.0)    # 2 minutes for long responses

# Count tokens without consuming API credits (use count_tokens)
token_count = client.messages.count_tokens(
    model="claude-sonnet-5",
    messages=request.messages
)
if token_count.input_tokens > 50000:
    raise HTTPException(400, "Request too large")