Exposing Local LLMs via REST API with FastAPI
Running large language models locally has become remarkably accessible thanks to projects like llama.cpp, transformers, and ollama. But a model running in a Python script is only useful to that single process. To make your local LLM available to other applications — a chat UI, a Slack bot, a RAG pipeline, or a microservice — you need to wrap it behind an HTTP interface. FastAPI is the ideal tool for this: it is async-native, fast, generates automatic documentation, and integrates cleanly with Python's ML ecosystem.
This tutorial walks through building a production-grade REST API around a local LLM using FastAPI. We will cover model loading, request/response schemas, streaming, error handling, and deployment best practices.
Why Expose a Local LLM via REST?
- Decoupling: Separate the heavy inference process from lightweight consumer applications written in any language.
- Resource isolation: Run the GPU/CPU-bound model server on one machine and consume it from many clients.
- Standardization: A REST endpoint is trivially consumable by web frontends, notebooks, and CI pipelines.
- OpenAI compatibility: You can mimic the OpenAI Chat Completions API so existing tools work with your local model unchanged.
- Observability: Centralize logging, rate limiting, and metrics at the API layer rather than inside each consumer.
Project Setup
Create a new project directory and install the dependencies. We will use llama-cpp-python for efficient GGUF model inference and fastapi with uvicorn for the server.
mkdir local-llm-api && cd local-llm-api
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn pydantic llama-cpp-python
If you have a CUDA-capable GPU, install llama-cpp-python with GPU support to dramatically improve inference speed:
CMAKE_ARGS="-DGGML_CUDA=on" pip install llama-cpp-python --upgrade --force-reinstall --no-cache-dir
Download a quantized model in GGUF format. For this tutorial we will use a small instruct-tuned model, but the code works with any GGUF file:
curl -L https://huggingface.co/TheBloke/Mistral-7B-Instruct-v0.2-GGUF/resolve/main/mistral-7b-instruct-v0.2.Q4_K_M.gguf -o models/mistral.gguf
Defining the Data Models
FastAPI uses Pydantic for request and response validation. We will mirror the OpenAI chat completions schema so the API feels familiar. Create a file named main.py and start with the schemas:
from pydantic import BaseModel, Field
from typing import List, Optional, Literal
class ChatMessage(BaseModel):
role: Literal["system", "user", "assistant"]
content: str
class ChatCompletionRequest(BaseModel):
model: str = "local"
messages: List[ChatMessage]
temperature: float = Field(default=0.7, ge=0.0, le=2.0)
max_tokens: int = Field(default=512, ge=1, le=8192)
stream: bool = False
stop: Optional[List[str]] = None
class ChatCompletionResponse(BaseModel):
model: str
choices: List[dict]
usage: dict
These models give us automatic validation. If a client sends temperature: 5.0, FastAPI returns a 422 error before the model is ever touched.
Loading the Model at Startup
Loading a multi-gigabyte model on every request would be catastrophic for latency. Instead, load it once when the application starts and keep it in memory. FastAPI provides lifespan events for exactly this purpose.
from contextlib import asynccontextmanager
from llama_cpp import Llama
MODEL_PATH = "models/mistral.gguf"
llm: Optional[Llama] = None
@asynccontextmanager
async def lifespan(app):
global llm
print("Loading model...")
llm = Llama(
model_path=MODEL_PATH,
n_ctx=4096,
n_gpu_layers=-1, # offload all layers to GPU if available
verbose=False,
)
print("Model loaded.")
yield
print("Shutting down.")
app = FastAPI(title="Local LLM API", lifespan=lifespan)
The n_ctx parameter sets the context window size. Choose a value that balances memory usage against the longest conversation you expect to serve. The n_gpu_layers=-1 setting pushes the entire model to the GPU; on CPU-only machines, set this to 0.
Implementing the Chat Completion Endpoint
Now add the core endpoint. The llama-cpp-python library is synchronous and CPU/GPU-bound, so we must run inference in a thread pool to avoid blocking the async event loop. FastAPI's run_in_threadpool helper does this cleanly.
from fastapi import FastAPI, HTTPException
from fastapi.concurrency import run_in_threadpool
import time
@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
async def chat_completions(request: ChatCompletionRequest):
if llm is None:
raise HTTPException(status_code=503, detail="Model not loaded")
prompt = format_messages(request.messages)
start = time.time()
try:
result = await run_in_threadpool(
llm,
prompt=prompt,
max_tokens=request.max_tokens,
temperature=request.temperature,
stop=request.stop,
echo=False,
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
elapsed = time.time() - start
text = result["choices"][0]["text"]
return ChatCompletionResponse(
model=request.model,
choices=[{
"index": 0,
"message": {"role": "assistant", "content": text},
"finish_reason": result["choices"][0]["finish_reason"],
}],
usage={
"prompt_tokens": result["usage"]["prompt_tokens"],
"completion_tokens": result["usage"]["completion_tokens"],
"total_tokens": result["usage"]["total_tokens"],
"latency_seconds": round(elapsed, 3),
},
)
def format_messages(messages: List[ChatMessage]) -> str:
"""Convert chat messages into a single prompt string."""
parts = []
for msg in messages:
if msg.role == "system":
parts.append(f"<|system|>\n{msg.content}")
elif msg.role == "user":
parts.append(f"<|user|>\n{msg.content}")
elif msg.role == "assistant":
parts.append(f"<|assistant|>\n{msg.content}")
parts.append("<|assistant|>\n")
return "\n".join(parts)
The format_messages helper applies a chat template. Different models expect different templates — Mistral, Llama 3, and Phi all use distinct special tokens. Always check the model card for the correct format, or use the model's built-in apply_chat_template if available.
Adding Streaming Support
For chat interfaces, streaming tokens as they are generated dramatically improves perceived latency. FastAPI supports this through StreamingResponse with Server-Sent Events. Add a streaming path inside the same endpoint:
from fastapi.responses import StreamingResponse
import json
@app.post("/v1/chat/completions")
async def chat_completions(request: ChatCompletionRequest):
if llm is None:
raise HTTPException(status_code=503, detail="Model not loaded")
prompt = format_messages(request.messages)
if request.stream:
return StreamingResponse(
stream_response(prompt, request),
media_type="text/event-stream",
)
# ... non-streaming path from previous section ...
async def stream_response(prompt: str, request: ChatCompletionRequest):
def generate():
for chunk in llm(
prompt=prompt,
max_tokens=request.max_tokens,
temperature=request.temperature,
stop=request.stop,
stream=True,
):
token = chunk["choices"][0]["text"]
payload = {
"model": request.model,
"choices": [{"index": 0, "delta": {"content": token}}],
}
yield f"data: {json.dumps(payload)}\n\n"
yield "data: [DONE]\n\n"
# Run the blocking generator in a thread
import asyncio
loop = asyncio.get_event_loop()
queue: asyncio.Queue = asyncio.Queue()
async def producer():
for item in await loop.run_in_executor(None, lambda: list(generate())):
await queue.put(item)
await queue.put(None)
asyncio.create_task(producer())
while True:
item = await queue.get()
if item is None:
break
yield item
This pattern bridges the synchronous llama-cpp-python generator with FastAPI's async streaming. Each token is wrapped in the SSE data: format that the OpenAI client libraries already understand.
Adding a Health Check and Model Info Endpoint
Production services need health checks for orchestrators like Kubernetes or Docker Compose. Add lightweight endpoints that do not invoke the model:
@app.get("/health")
async def health():
return {"status": "ok", "model_loaded": llm is not None}
@app.get("/v1/models")
async def list_models():
return {
"object": "list",
"data": [
{
"id": "local",
"object": "model",
"owned_by": "local",
}
],
}
Running the Server
Launch the API with Uvicorn. Use multiple workers only if you have enough GPU memory to load the model multiple times — for most setups, a single worker is correct.
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1
Test it with curl:
curl -X POST http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{
"messages": [
{"role": "system", "content": "You are a concise coding assistant."},
{"role": "user", "content": "Explain async/await in one sentence."}
],
"temperature": 0.3,
"max_tokens": 128
}'
FastAPI also generates interactive documentation automatically. Open http://localhost:8000/docs in your browser to explore and test every endpoint.
Best Practices
Concurrency and Throughput
A single LLM instance processes one request at a time. If multiple requests arrive simultaneously, they queue behind the thread pool. For higher throughput, consider a continuous batching server like vLLM or llama.cpp's built-in server, and use FastAPI as a thin proxy in front of it for auth and schema translation.
Authentication
Even on a local network, add an API key check. A simple dependency works well:
from fastapi import Depends, Header, Security
from fastapi.security import APIKeyHeader
api_key_header = APIKeyHeader(name="Authorization", auto_error=False)
async def verify_api_key(authorization: str = Security(api_key_header)):
if authorization != f"Bearer {os.environ.get('API_KEY')}":
raise HTTPException(status_code=401, detail="Invalid API key")
return authorization
# Then add to protected routes:
@app.post("/v1/chat/completions", dependencies=[Depends(verify_api_key)])
Request Validation and Limits
Always cap max_tokens and validate input length. A malicious or buggy client could request 32,000 tokens and tie up the server for minutes. The Pydantic Field constraints we added earlier handle this, but also consider truncating very long prompts before they reach the model.
Graceful Shutdown
The lifespan context manager ensures the model is released on shutdown. When deploying behind Docker, send SIGTERM rather than SIGKILL so Uvicorn can finish in-flight requests. Configure this in your Dockerfile or Compose file with stop_grace_period.
Logging and Observability
Log token counts, latency, and errors for every request. Integrate with structlog or loguru for structured JSON logs, and expose Prometheus metrics via prometheus-fastapi-instrumentator to track request volume and inference time histograms.
Model Hot-Swapping
For multi-model setups, load models lazily into a dictionary keyed by model ID, and evict the least-recently-used model when memory is constrained. This lets a single API endpoint serve several models without restarting.
Conclusion
Wrapping a local LLM in a FastAPI service transforms a single-process experiment into a reusable, language-agnostic inference platform. By loading the model once at startup, offloading blocking inference to a thread pool, mirroring the OpenAI chat completions schema, and adding streaming, health checks, and authentication, you get a server that drops cleanly into existing toolchains and scales to serve real applications. Start with the single-endpoint version above, then layer in batching, metrics, and model management as your usage grows — the architecture stays the same, only the operational sophistication increases.