n4nAI
Open console →
API integrationTutorialFastAPI LLM Backend Integration

Building an LLM-powered FastAPI backend from scratch

A hands-on fastapi llm backend tutorial: scaffold an async Python service that proxies prompts to an OpenAI-compatible LLM API with streaming and error handling.

n4n Team2 min read525 words

Audio narration

Coming soon — every post will get a voice note here.

This fastapi llm backend tutorial walks through building an asynchronous Python service that accepts prompts and proxies them to an OpenAI-compatible inference endpoint. We’ll cover project setup, request validation, streaming responses, and graceful provider failures so you can ship a real backend rather than a toy script.

Prerequisites

  • Python 3.11 or newer
  • pip and a virtual environment
  • Working knowledge of FastAPI routes and Pydantic models
  • An API key for any OpenAI-compatible LLM service (OpenAI, or a gateway that exposes the same /v1 chat completions interface)

If you use a gateway, point BASE_URL at its endpoint instead of api.openai.com/v1. The client code does not change.

Project Setup

Create the project and install dependencies:

mkdir fastapi-llm && cd fastapi-llm
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn openai python-dotenv httpx

We use openai’s async client because it speaks the OpenAI HTTP contract, which most inference gateways mirror exactly. httpx is a transitive dep but useful if you later swap to raw requests.

Create a .env file:

OPENAI_API_KEY=sk-your-key
OPENAI_BASE_URL=https://api.openai.com/v1
MODEL_NAME=gpt-4o-mini
REQUEST_TIMEOUT=30

Configuration Module

Keep configuration isolated. Avoid scattering os.getenv calls across route handlers.

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("OPENAI_API_KEY", "")
BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")
TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "30"))

Async Client Singleton

Instantiate the client once at import time. The AsyncOpenAI client manages its own connection pool.

# client.py
from openai import AsyncOpenAI
from config import API_KEY, BASE_URL, TIMEOUT

client = AsyncOpenAI(
    api_key=API_KEY,
    base_url=BASE_URL,
    timeout=TIMEOUT,
)

Request and Response Models

Validate input at the edge. Pydantic saves you from malformed payloads reaching the upstream model.

# models.py
from pydantic import BaseModel, Field

class ChatRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=4000)
    max_tokens: int = Field(512, ge=1, le=4096)
    temperature: float = Field(0.7, ge=0.0, le=2.0)

class ChatResponse(BaseModel):
    text: str
    model: str
    finish_reason: str | None
    prompt_tokens: int | None = None
    completion_tokens: int | None = None

Exposing token counts helps callers estimate cost and latency.

Basic Non-Streaming Endpoint

Wire the pieces together in main.py.

# main.py
from fastapi import FastAPI, HTTPException
from client import client
from config import MODEL_NAME
from models import ChatRequest, ChatResponse

app = FastAPI(title="LLM Proxy")

@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    try:
        resp = await client.chat.completions.create(
            model=MODEL_NAME,
            messages=[{"role": "user", "content": req.prompt}],
            max_tokens=req.max_tokens,
            temperature=req.temperature,
        )
    except Exception as e:
        # Map upstream failures to a 502; adjust if you distinguish 429/5xx
        raise HTTPException(status_code=502, detail=f"upstream error: {e}")

    choice = resp.choices[0]
    usage = resp.usage
    return ChatResponse(
        text=choice.message.content or "",
        model=resp.model,
        finish_reason=choice.finish_reason,
        prompt_tokens=usage.prompt_tokens if usage else None,
        completion_tokens=usage.completion_tokens if usage else None,
    )

Run it:

uvicorn main:app --reload --port 8000

Checkpoint — call the endpoint:

curl -s -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"prompt":"Return a JSON object with a hello key"}'

Expected output:

{
  "text": "{\"hello\": \"world\"}",
  "model": "gpt-4o-mini",
  "finish_reason": "stop",
  "prompt_tokens": 14,
  "completion_tokens": 12
}

Streaming Endpoint

For chat UIs, token streaming is non-negotiable. FastAPI’s StreamingResponse works cleanly with async generators.

from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    async def token_gen():
        try:
            stream = await client.chat.completions.create(
                model=MODEL_NAME,
                messages=[{"role": "user", "content": req.prompt}],
                max_tokens=req.max_tokens,
                temperature=req.temperature,
                stream=True,
            )
            async for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    yield delta
        except Exception as e:
            yield f"[upstream error: {e}]"

    return StreamingResponse(token_gen(), media_type="text/plain")

Test with curl’s -N flag to disable buffering:

curl -N -X POST http://localhost:8000/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"prompt":"Count to three slowly"}'

You should see tokens arrive incrementally, not as one blob.

Handling Provider Degradation

Networks fail. Providers rate-limit. A production fastapi llm backend tutorial must address fallback. The minimal pattern: try primary, then a secondary client if configured.

# fallback.py
from client import client as primary
from openai import AsyncOpenAI
from config import API_KEY, TIMEOUT

secondary = None
if os.getenv("FALLBACK_BASE_URL"):
    secondary = AsyncOpenAI(
        api_key=API_KEY,
        base_url=os.getenv("FALLBACK_BASE_URL"),
        timeout=TIMEOUT,
    )

async def complete(req):
    try:
        return await primary.chat.completions.create(
            model=MODEL_NAME,
            messages=[{"role": "user", "content": req.prompt}],
            max_tokens=req.max_tokens,
            temperature=req.temperature,
        )
    except Exception:
        if secondary:
            return await secondary.chat.completions.create(
                model=os.getenv("FALLBACK_MODEL", MODEL_NAME),
                messages=[{"role": "user", "content": req.prompt}],
                max_tokens=req.max_tokens,
                temperature=req.temperature,
            )
        raise

If you’d rather not hand-roll this, an OpenAI-compatible gateway such as n4n.ai performs automatic fallback when a provider is rate-limited or degraded, and forwards provider cache-control hints, while keeping the exact same openai client code shown above.

Dependency Injection for Testability

Hard-coding the client makes unit tests hit the network. Use FastAPI’s Depends:

from typing import Callable
from fastapi import Depends

def get_client():
    return client

@app.post("/chat/di")
async def chat_di(req: ChatRequest, cli=Depends(get_client)):
    resp = await cli.chat.completions.create(
        model=MODEL_NAME,
        messages=[{"role": "user", "content": req.prompt}],
        max_tokens=req.max_tokens,
    )
    return {"text": resp.choices[0].message.content}

In tests, override get_client with a stub that returns a fake completion.

Testing the Service

A minimal pytest using TestClient:

# test_main.py
from fastapi.testclient import TestClient
from main import app

def test_chat_validation():
    with TestClient(app) as c:
        r = c.post("/chat", json={"prompt": ""})
        assert r.status_code == 422  # pydantic rejects empty

def test_chat_ok(monkeypatch):
    class FakeMsg:
        content = "ok"
    class FakeChoice:
        message = FakeMsg()
    class FakeUsage:
        prompt_tokens = 1
        completion_tokens = 1
    class FakeResp:
        model = "test"
        choices = [FakeChoice()]
        usage = FakeUsage()
    async def fake_create(*args, **kwargs):
        return FakeResp()
    monkeypatch.setattr("client.client.chat.completions.create", fake_create)
    with TestClient(app) as c:
        r = c.post("/chat", json={"prompt":"hi"})
        assert r.status_code == 200
        assert r.json()["text"] == "ok"

Run pytest -q. Green means your validation and serialization are correct.

Production Notes

  • Timeouts: The openai client timeout applies per request. Set it lower (e.g., 10s) for interactive endpoints.
  • Concurrency: Uvicorn with --workers 4 plus async I/O handles hundreds of concurrent streams on a small box.
  • Rate limiting: Put a reverse proxy (Caddy/Nginx) or a middleware in front to avoid blowing your provider quota.
  • Logging: Capture resp.usage and resp.model on every call. If you use a gateway that provides per-token usage metering, forward those fields to your metrics pipeline.
  • Caching: For repeated prompts, use HTTP Cache-Control semantics or a KV store. Some gateways honor provider cache-control hints; otherwise implement your own keyed by prompt hash.

Wrapping Up

You now have a runnable FastAPI service that validates input, calls an OpenAI-compatible LLM, streams tokens, and degrades reasonably when the primary provider fails. The same code works against OpenAI, a self-hosted vLLM instance, or a multi-provider gateway. From here, add auth, request queues, and structured logging before exposing it to real traffic.

Tagsfastapipythonllm-apibackend

Written by

n4n Team

The team building n4n — a single OpenAI-compatible API in front of 240+ models, with automatic fallback, load balancing and pay-per-token metering.

More from n4n Team →
API integrationStreaming LLM responses from FastAPI with StreamingResponseStep-by-step guide to building a FastAPI StreamingResponse LLM endpoint in Python: stream tokens from OpenAI-compatible APIs, handle disconnects, verify with curl.API integrationFastAPI background tasks for async LLM callsBuild resilient FastAPI background tasks for async LLM calls with this end-to-end how-to covering request patterns, status tracking, retries, and observability.API integrationFastAPI error handling for upstream LLM provider failuresA practical guide to fastapi error handling llm provider failures: structured exceptions, retries, fallbacks, and response shaping for robust backends.API integrationFastAPI vs Flask for building LLM-backed APIsA pragmatic head-to-head: FastAPI vs Flask for LLM APIs. Compare async streaming, validation, latency, and ecosystem to pick the right backend.API integrationRate limiting an LLM API in FastAPI with slowapiPractical step-by-step tutorial for fastapi rate limiting slowapi llm APIs: enforce per-client token limits, proxy to providers, and verify with curl tests.API integrationDeploying a FastAPI LLM backend with uvicorn and gunicornPractical guide to fastapi deploy uvicorn gunicorn llm services: build async inference routes, tune workers, run with gunicorn, and verify in production.

All fastapi llm backend integration posts →