Wiring an LLM into a backend is mostly plumbing: request validation, transport, streaming, and failure handling. This guide builds a django rest framework llm api endpoint that proxies to an OpenAI-compatible gateway, supports token streaming, and fails over without crashing your service.
Step 1: Scaffold the Django project and DRF
Start with a clean virtualenv and install the minimal dependency set. You need Django, DRF for the API layer, and an HTTP client that speaks the OpenAI wire format. The official openai Python package works fine even if you never touch OpenAI’s servers, because the protocol is just JSON over HTTP.
python -m venv venv
source venv/bin/activate
pip install django djangorestframework openai
django-admin startproject llmproxy .
python manage.py startapp apiAdd rest_framework and api to INSTALLED_APPS in llmproxy/settings.py. Disable default browsable API in production by overriding DEFAULT_RENDERER_CLASSES to JSONRenderer only.
INSTALLED_APPS = [
"django.contrib.admin",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.messages",
"django.contrib.staticfiles",
"rest_framework",
"api",
]
REST_FRAMEWORK = {
"DEFAULT_RENDERER_CLASSES": ["rest_framework.renderers.JSONRenderer"],
}Step 2: Define the request contract
A django rest framework llm api endpoint must reject malformed prompts before they hit the network. Use a serializer to enforce types and defaults. Keep the surface small: model name, message list, temperature, and a stream flag.
from rest_framework import serializers
class ChatRequestSerializer(serializers.Serializer):
model = serializers.CharField(default="gpt-4o-mini")
messages = serializers.ListField(
child=serializers.DictField(child=serializers.CharField())
)
temperature = serializers.FloatField(default=0.7, min_value=0.0, max_value=2.0)
stream = serializers.BooleanField(default=False)
def validate_messages(self, value):
if not value:
raise serializers.ValidationError("messages must not be empty")
for m in value:
if "role" not in m or "content" not in m:
raise serializers.ValidationError("each message needs role and content")
return valueThe validate_messages hook catches the most common client bug: sending a string instead of a list of role/content pairs. Do not silently coerce; fail fast.
Step 3: Implement the non-streaming view
For batch callers, a plain JSON response is simplest. Instantiate the OpenAI client with a configurable base_url so you can swap providers without code changes. Pull the bearer token from the request header rather than hardcoding secrets.
from openai import OpenAI
from rest_framework.views import APIView
from rest_framework.response import Response
from .serializers import ChatRequestSerializer
class ChatView(APIView):
def post(self, request):
ser = ChatRequestSerializer(data=request.data)
ser.is_valid(raise_exception=True)
data = ser.validated_data
api_key = request.headers.get("Authorization", "").removeprefix("Bearer ")
client = OpenAI(base_url="https://api.openai.com/v1", api_key=api_key)
try:
resp = client.chat.completions.create(
model=data["model"],
messages=data["messages"],
temperature=data["temperature"],
stream=False,
)
except Exception as exc:
return Response({"error": str(exc)}, status=502)
return Response({
"id": resp.id,
"model": resp.model,
"usage": resp.usage.model_dump(),
"content": resp.choices[0].message.content,
})Map upstream failures to 502 instead of 500. A provider timeout is not your app’s bug; treat it as a bad gateway.
Step 4: Add streaming with Server-Sent Events
Token streaming keeps perceived latency low. Django’s StreamingHttpResponse pairs well with the OpenAI client’s iterator. Yield SSE frames and terminate with [DONE].
import json
from django.http import StreamingHttpResponse
from openai import OpenAI
from .serializers import ChatRequestSerializer
class ChatStreamView(APIView):
def post(self, request):
ser = ChatRequestSerializer(data=request.data)
ser.is_valid(raise_exception=True)
data = ser.validated_data
api_key = request.headers.get("Authorization", "").removeprefix("Bearer ")
client = OpenAI(base_url="https://api.openai.com/v1", api_key=api_key)
def event_stream():
try:
stream = client.chat.completions.create(
model=data["model"],
messages=data["messages"],
temperature=data["temperature"],
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield f"data: {json.dumps({'delta': delta})}\n\n"
yield "data: [DONE]\n\n"
except Exception as exc:
yield f"data: {json.dumps({'error': str(exc)})}\n\n"
return StreamingHttpResponse(event_stream(), content_type="text/event-stream")Run this behind a WSGI server with threaded workers. Streaming holds a connection open; gunicorn with --workers 4 --threads 8 handles moderate concurrency without async rewrite.
Step 5: Route models and survive provider outages
Hardcoding a single provider is fragile. If you point the client at a gateway such as n4n.ai, you get one OpenAI-compatible endpoint addressing 240+ models with automatic fallback when a provider is rate-limited or degraded. The gateway honors client routing directives and forwards provider cache-control hints, so you can pin a vendor or enable prompt caching via headers without branching your DRF code.
resp = client.chat.completions.create(
model=data["model"],
messages=data["messages"],
temperature=data["temperature"],
stream=data["stream"],
extra_headers={
"x-routing-prefer": "anthropic",
"cache-control": "max-age=3600",
},
)Per-token usage metering arrives in the response object regardless of which backend served the request, so your billing layer stays simple.
Step 6: Lock down the endpoint
Never expose an LLM proxy to the public internet without auth. Use DRF’s TokenAuthentication and throttle by user.
REST_FRAMEWORK = {
"DEFAULT_AUTHENTICATION_CLASSES": [
"rest_framework.authentication.TokenAuthentication"
],
"DEFAULT_PERMISSION_CLASSES": [
"rest_framework.permissions.IsAuthenticated"
],
"DEFAULT_THROTTLE_CLASSES": [
"rest_framework.throttling.UserRateThrottle"
],
"DEFAULT_THROTTLE_RATES": {"user": "60/min"},
}Generate tokens via python manage.py drf_create_token <username>. Reject requests missing the header at the Nginx layer to save Python workers.
Step 7: Verify end to end
Write a smoke test that asserts the non-streaming path returns usage and content.
import pytest
from rest_framework.test import APIClient
@pytest.mark.django_db
def test_chat_nonstream():
client = APIClient()
# create token and authenticate omitted for brevity
resp = client.post("/api/chat/", {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": "ping"}],
"stream": False,
}, format="json")
assert resp.status_code == 200
assert "content" in resp.json()
assert "usage" in resp.json()For streaming, use curl with -N:
curl -N -X POST http://localhost:8000/api/chat/stream/ \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}],"stream":true}'Success looks like repeated data: {"delta": "..."} lines ending with data: [DONE]. If you see a raw traceback, your serializer likely accepted a bad payload.
Step 8: Production deployment notes
Django’s sync view is adequate for streaming because the OpenAI client yields synchronously. Do not put the view behind asyncio unless you rewrite the client calls. Deploy with gunicorn and a reverse proxy that buffers correctly:
gunicorn llmproxy.wsgi:application --workers 4 --threads 8 --bind 0.0.0.0:8000Set CLIENT_BASE_URL via environment variable and read it in the view instead of the hardcoded string. Log the resp.usage tuple to your metrics pipeline; token counts are the only cost signal that matters.
A django rest framework llm api endpoint is not exotic. It is a validated serializer, a thin client, and disciplined error mapping. Build it once, point it at a gateway that handles model sprawl, and ship.