--- title: "Streaming Chat API" description: "Cost/latency tracking and rate-limiting are what interviewers and hiring managers actually probe for beyond 'can you call an LLM' — this project gives you..." source: "https://confidentprep.com/paths/ai-engineering-fundamentals/streaming-chat-api/" path: "AI Engineering Fundamentals" repository: "https://github.com/confidentprep14-web/ai-engineering-projects/tree/main/streaming-chat-api" token_estimate: 3016 --- # Streaming Chat API ## Overview Everything so far has been a CLI. This project moves it to a FastAPI server that streams LLM responses over SSE, tracks token usage and cost per request in SQLite, and rate-limits callers per IP. This is the shape every real LLM-backed product takes — cost and abuse control aren't optional extras, they're the difference between a side project and something you could actually run. ## What to do **Path:** 1 **Position:** 8 of 12 **Difficulty:** 🟢 Local — one API key, no Docker, no cloud **Estimated time:** 4–5 hours **AWS cost:** None --- ### Agent Pickup Instructions **This spec is self-contained. Build this project without reading any other spec file.** ```bash mkdir -p path-1/p1-08-streaming-chat-api cd path-1/p1-08-streaming-chat-api pip install -r requirements.txt pytest tests/ -v uvicorn src.main:app --reload & curl -N -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{"message": "Count to 5 slowly"}' \ -H "Accept: text/event-stream" curl http://localhost:8000/stats curl http://localhost:8000/health ``` **Done when:** - [ ] `pytest tests/ -v` → minimum 5 tests, all green - [ ] `/chat` streams tokens via SSE (visible with `curl -N`) - [ ] Token count and cost logged to SQLite on every request - [ ] `/stats` returns p50 and p95 latency from real request log - [ ] Rate limiter returns HTTP 429 with `Retry-After` header when limit exceeded - [ ] LLM provider error returns HTTP 503 with `Retry-After: 10` header - [ ] No file in `src/` exceeds 200 lines --- ### What this project is A production-shaped FastAPI server that serves LLM responses over Server-Sent Events (SSE), tracks every request's token count, cost, and latency in SQLite, and rate-limits callers per IP. This is the shape of every real LLM inference endpoint. The motivation is natural: you've built a persistent agent that people want to use — now expose it over HTTP. --- ### What the learner achieves "I built a streaming chat API with SSE, per-request cost and token tracking in SQLite, and a sliding-window rate limiter — and the /stats endpoint shows real p50/p95 latency from logged request data." --- ### Folder structure ``` p1-08-streaming-chat-api/ ├── README.md ├── GUIDE.md ├── .env.example ├── requirements.txt ├── src/ │ ├── main.py ← FastAPI app, routes │ ├── llm.py ← streaming LLM wrapper │ ├── database.py ← SQLite schema + request logger │ ├── rate_limiter.py ← sliding window rate limiter │ └── config.py ← reads all env vars └── tests/ └── test_api.py ``` --- ### .env.example ```bash LLM_PROVIDER=anthropic LLM_API_KEY= LLM_MODEL= # Rate limiting RATE_LIMIT_REQUESTS_PER_MINUTE=20 # Cost per 1M tokens (required; verify against the selected model's current vendor pricing) COST_PER_1M_INPUT_TOKENS= COST_PER_1M_OUTPUT_TOKENS= # SQLite database path DATABASE_PATH=./chat_requests.db ``` --- ### requirements.txt ``` anthropic==0.40.0 openai==1.58.0 ollama==0.4.4 fastapi==0.115.6 uvicorn==0.34.0 sse-starlette==2.2.1 tiktoken==0.8.0 python-dotenv==1.0.1 httpx==0.28.1 pytest==8.3.4 pytest-asyncio==0.25.2 ``` --- ### src/ — what to implement #### src/config.py ```python import os from dataclasses import dataclass from dotenv import load_dotenv load_dotenv() @dataclass class Config: llm_provider: str = os.getenv("LLM_PROVIDER", "anthropic") llm_model: str = os.environ["LLM_MODEL"] llm_api_key: str = os.getenv("LLM_API_KEY", "") rate_limit_rpm: int = int(os.getenv("RATE_LIMIT_REQUESTS_PER_MINUTE", "20")) cost_per_1m_input: float = float(os.getenv("COST_PER_1M_INPUT_TOKENS", "1.00")) cost_per_1m_output: float = float(os.getenv("COST_PER_1M_OUTPUT_TOKENS", "5.00")) database_path: str = os.getenv("DATABASE_PATH", "./chat_requests.db") config = Config() ``` #### src/database.py Schema (create on startup): ```sql CREATE TABLE IF NOT EXISTS requests ( id INTEGER PRIMARY KEY AUTOINCREMENT, request_id TEXT NOT NULL, client_ip TEXT, prompt_tokens INTEGER NOT NULL, completion_tokens INTEGER NOT NULL, cost_usd REAL NOT NULL, latency_ms INTEGER NOT NULL, model TEXT NOT NULL, status TEXT NOT NULL, -- 'success' | 'error' created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); ``` **`init_db(db_path: str) -> None`** — creates tables if needed **`log_request(db_path: str, request_id: str, client_ip: str, prompt_tokens: int, completion_tokens: int, cost_usd: float, latency_ms: int, model: str, status: str) -> None`** **`get_stats(db_path: str) -> dict`** - Returns: `{"total_requests": N, "total_cost_usd": X, "p50_latency_ms": N, "p95_latency_ms": N, "error_rate": 0.05}` - Compute p50/p95 by loading all latencies and using `statistics.median()` and the 95th percentile - Error rate: `error_count / total_count` #### src/rate_limiter.py **`RateLimiter` class — sliding window per IP:** `__init__(self, requests_per_minute: int)` — `self.window: dict[str, list[float]] = {}` `is_allowed(self, client_ip: str) -> tuple[bool, int]` - Returns `(allowed: bool, retry_after_seconds: int)` - Get current time. Remove timestamps from `self.window[client_ip]` older than 60 seconds. - If `len(timestamps) >= requests_per_minute`: return `(False, 60 - (now - oldest_timestamp))` - Else: append `now`, return `(True, 0)` #### src/llm.py **`stream_completion(prompt: str, system: str = "") -> Iterator[tuple[str, int, int]]`** - Yields `(token_text, prompt_tokens, completion_tokens)` tuples - `prompt_tokens` and `completion_tokens` are only non-zero on the LAST yield (use tiktoken for prompt, accumulate completion) - `anthropic`: use streaming context manager; yield text deltas; on final event yield `("", prompt_tokens, completion_tokens)` with usage from response - `openai`: iterate `stream=True` chunks; yield delta content; track usage from final chunk - `ollama`: iterate streaming response; yield content chunks; count tokens via tiktoken on completion #### src/main.py FastAPI app with three routes: **`GET /health`** ```json {"status": "ok", "model": "", "provider": "anthropic"} ``` **`POST /chat`** — body: `{"message": str, "system": str (optional)}` - Get `client_ip` from `request.client.host` - Check rate limiter: if not allowed, return `HTTP 429` with `Retry-After: {seconds}` header and body `{"error": "Rate limit exceeded", "retry_after": N}` - Generate `request_id = str(uuid.uuid4())` - Record `t_start = time.perf_counter()` - Return `EventSourceResponse` (from sse_starlette) that: - Streams tokens via `stream_completion()` - On each token: yield `{"data": token_text}` - On final yield: record `latency_ms`, calculate cost, call `log_request()` - On LLM error: yield `{"data": "[ERROR]"}`, log request with status="error", return - If LLM provider raises an error: return `HTTP 503` with `Retry-After: 10` header **`GET /stats`** ```json { "total_requests": 42, "total_cost_usd": 0.0031, "p50_latency_ms": 1200, "p95_latency_ms": 3400, "error_rate": 0.02 } ``` --- ### tests/ — what to test **File:** `tests/test_api.py` — use `httpx.AsyncClient` with FastAPI's `app` directly; mock LLM calls. **Test 1 — /health returns 200:** `GET /health` → assert status 200, body contains "status": "ok". **Test 2 — rate limiter returns 429 on excess:** Mock rate limiter to return `(False, 30)`. `POST /chat` → assert status 429, `Retry-After` header present. **Test 3 — stats returns correct structure:** Pre-populate DB with 3 test requests. `GET /stats` → assert response contains "total_requests", "p50_latency_ms", "p95_latency_ms". **Test 4 — rate limiter allows under limit:** Create `RateLimiter(requests_per_minute=5)`. Call `is_allowed("127.0.0.1")` 5 times. Assert all return `(True, 0)`. Call 6th time. Assert returns `(False, ...)`. **Test 5 — get_stats calculates p50 correctly:** Insert 10 requests with latencies [100, 200, 300, 400, 500, 600, 700, 800, 900, 1000]. Call `get_stats()`. Assert `p50_latency_ms == 550` (median of sorted list). --- ### README.md content ```markdown # Streaming Chat API A FastAPI server that streams LLM responses over SSE, tracks token usage and cost per request in SQLite, and rate-limits callers per IP. ## Setup ```bash cd p1-08-streaming-chat-api cp .env.example .env pip install -r requirements.txt ``` ## Run ```bash uvicorn src.main:app --reload ``` Test with curl: ```bash ## Stream a response curl -N -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{"message": "Tell me about black holes"}' \ -H "Accept: text/event-stream" ## Check stats curl http://localhost:8000/stats ``` ## Tests ```bash pytest tests/ -v ``` ## What to try next - Add an API key header for authentication - Add the persistent agent from p1-07 as the chat backend - Deploy this with Docker (see p1-10-dockerize) ``` --- ### GUIDE.md content ```markdown # Build guide: Streaming Chat API ## What you're building and why it matters Every production LLM application eventually needs an HTTP API. SSE (Server-Sent Events) is the standard transport for streaming text: the client opens one long-lived HTTP connection and the server pushes data as it arrives. This is how ChatGPT, Claude.ai, and every other chat product delivers streaming responses. Rate limiting and cost tracking are not optional in production: without them, one busy user can exhaust your API quota and your budget. ## The decision that matters in this build **SSE vs WebSockets for streaming.** WebSockets are bidirectional and stateful; SSE is one-directional (server to client) and stateless. For streaming LLM output, SSE is the right choice: the client sends one HTTP request and the server streams back. No need for the complexity of a WebSocket connection for this use case. sse-starlette makes this a three-line addition to any FastAPI route. ## What will break **`curl` without `-N` buffers SSE output.** When testing with curl, always use `-N` (disable buffering). Without it, you'll see the entire response appear at once when the stream closes, not token by token. **p95 latency requires enough data.** The /stats endpoint computes p95 from all stored requests. With fewer than 20 requests, p95 is not statistically meaningful. Make 30+ requests before trusting the p95 number. ## How to talk about this in an interview "I built a streaming chat API where every request's token count, cost, and latency is logged to SQLite. The /stats endpoint computes p50/p95 latency from real data. I also implemented a sliding-window rate limiter that returns a Retry-After header — the same mechanism used by real API rate limiters." ``` --- ### Failure modes to handle | Failure | Where | How to handle | |---------|-------|---------------| | Rate limit exceeded | `main.py /chat` | Return HTTP 429 with `Retry-After` header | | LLM provider error | `main.py /chat` | Return HTTP 503 with `Retry-After: 10`, log as error | | DB write fails | `database.log_request()` | Log warning to stderr, don't fail the request | | Empty message body | `main.py /chat` | Return HTTP 422 (FastAPI validates automatically via Pydantic) | | Unknown LLM provider | `llm.py` | Return HTTP 500 with provider name in error body | --- ### The metric this project measures **p50 and p95 latency** — computed from SQLite log by `/stats` endpoint. **Total cost (USD)** — accumulated across all requests, shown in `/stats`. **Error rate** — fraction of failed requests. ### Model source currency - Anthropic model list: https://docs.anthropic.com/en/docs/about-claude/models/overview - OpenAI model list: https://platform.openai.com/docs/models - verifiedOn: 2026-08-07 - `LLM_MODEL` is required and has no implicit default; choose a supported model for the configured provider at setup time. ## Source code A reference implementation of Streaming Chat API — README, guide, source and tests — is at https://github.com/confidentprep14-web/ai-engineering-projects/tree/main/streaming-chat-api.