Ollama REST API: Build a Local LLM-Powered Backend with Python and FastAPI

Ollama gives you a fully local LLM inference server with a clean HTTP API. FastAPI gives you async Python routing with automatic request validation. Together, they form a production-grade AI backend that runs entirely on your own hardware — no cloud API keys, no per-token billing, no data leaving your network.

This tutorial walks through the full stack: calling the Ollama REST API from Python, wrapping it in a FastAPI service, adding streaming support, handling concurrent requests, and deploying the whole thing with Docker or systemd.

Contains affiliate links. Disclosure.

What you'll build

By the end of this guide you will have a running FastAPI service that:

• Accepts prompt requests over HTTP and forwards them to a local Ollama instance
• Streams token-by-token responses back to callers using Server-Sent Events (SSE)
• Validates and sanitises all inbound payloads with Pydantic
• Handles concurrent requests without blocking the event loop
• Exposes a /models endpoint to list and switch available Ollama models
• Applies a simple token-bucket rate limiter to prevent runaway inference queues
• Ships as a Docker container and as a systemd service

Prerequisites: Python 3.11+, Docker (optional), and Ollama installed locally. If you need a dedicated host, Liquid Web VPS plans start at specs that comfortably run 7B parameter models (32 GB RAM, modern CPU or GPU).

---

How the Ollama REST API works

Ollama exposes a local HTTP server on port 11434 by default. The core endpoints are:

Endpoint	Method	Purpose
/api/generate	POST	Single-turn text generation
/api/chat	POST	Multi-turn chat with message history
/api/embeddings	POST	Generate text embeddings
/api/tags	GET	List locally downloaded models
/api/pull	POST	Download a model by name
/api/delete	DELETE	Remove a model
/api/show	POST	Inspect model metadata

All responses are JSON. The /api/generate and /api/chat endpoints support streaming: when you set "stream": true, Ollama sends a sequence of newline-delimited JSON objects, one per token, instead of waiting until generation is complete.

Quick sanity check

Before wrapping anything in FastAPI, confirm Ollama is running and your chosen model is available:

# Start the Ollama daemon (if not already running as a service)
ollama serve

# Pull a model (llama3.2 is compact and fast)
ollama pull llama3.2

# Verify the REST API is live
curl http://localhost:11434/api/tags

A successful response looks like:

{
  "models": [
    {
      "name": "llama3.2:latest",
      "modified_at": "2026-03-10T09:14:22Z",
      "size": 2019393189
    }
  ]
}

---

Project setup

Create the project directory and install dependencies:

mkdir ollama-fastapi-backend && cd ollama-fastapi-backend
python -m venv .venv && source .venv/bin/activate

pip install fastapi uvicorn httpx pydantic python-dotenv

Package	Role
fastapi	Async web framework with Pydantic integration
uvicorn	ASGI server (Gunicorn-compatible)
httpx	Async HTTP client for upstream Ollama calls
pydantic	Request/response validation and serialisation
python-dotenv	Load .env config for Ollama host and other settings

Create a minimal .env file:

OLLAMA_HOST=http://localhost:11434
DEFAULT_MODEL=llama3.2
MAX_TOKENS=2048

---

Core implementation

1. Configuration and Ollama client

Create app/config.py:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    ollama_host: str = "http://localhost:11434"
    default_model: str = "llama3.2"
    max_tokens: int = 2048
    rate_limit_requests: int = 10   # requests per window
    rate_limit_window: int = 60     # window in seconds

    class Config:
        env_file = ".env"

settings = Settings()

Create app/ollama_client.py — a thin async wrapper around the Ollama HTTP API using httpx:

import httpx
import json
from typing import AsyncIterator
from app.config import settings

client = httpx.AsyncClient(base_url=settings.ollama_host, timeout=120.0)

async def generate(prompt: str, model: str, stream: bool = False) -> dict | AsyncIterator[str]:
    payload = {
        "model": model,
        "prompt": prompt,
        "stream": stream,
        "options": {"num_predict": settings.max_tokens},
    }

    if not stream:
        response = await client.post("/api/generate", json=payload)
        response.raise_for_status()
        return response.json()

    # Streaming path: yield each decoded token string
    async def _stream() -> AsyncIterator[str]:
        async with client.stream("POST", "/api/generate", json=payload) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if line:
                    chunk = json.loads(line)
                    yield chunk.get("response", "")
                    if chunk.get("done"):
                        break

    return _stream()

async def list_models() -> list[dict]:
    response = await client.get("/api/tags")
    response.raise_for_status()
    return response.json().get("models", [])

async def pull_model(name: str) -> AsyncIterator[str]:
    async with client.stream("POST", "/api/pull", json={"name": name}) as resp:
        resp.raise_for_status()
        async for line in resp.aiter_lines():
            if line:
                yield line

2. Pydantic schemas

Create app/schemas.py:

from pydantic import BaseModel, Field, field_validator

class GenerateRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=8000)
    model: str | None = None      # falls back to DEFAULT_MODEL
    stream: bool = False

    @field_validator("prompt")
    @classmethod
    def strip_prompt(cls, v: str) -> str:
        return v.strip()

class GenerateResponse(BaseModel):
    model: str
    response: str
    prompt_tokens: int | None = None
    completion_tokens: int | None = None
    total_duration_ms: int | None = None

class ModelInfo(BaseModel):
    name: str
    size: int
    modified_at: str

class PullRequest(BaseModel):
    name: str = Field(..., pattern=r"^[a-zA-Z0-9._\-:/]+$")

3. Rate limiter middleware

A simple sliding-window rate limiter using Python's asyncio primitives — no Redis required for a single-instance deployment:

Create app/rate_limiter.py:

import asyncio
import time
from collections import defaultdict, deque

from fastapi import Request, HTTPException
from app.config import settings

_request_log: dict[str, deque] = defaultdict(deque)
_lock = asyncio.Lock()

async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host if request.client else "unknown"
    now = time.monotonic()
    window = settings.rate_limit_window
    limit = settings.rate_limit_requests

    async with _lock:
        log = _request_log[client_ip]
        # Drop timestamps outside the current window
        while log and log[0] < now - window:
            log.popleft()
        if len(log) >= limit:
            raise HTTPException(
                status_code=429,
                detail=f"Rate limit exceeded: {limit} requests per {window}s",
            )
        log.append(now)

    return await call_next(request)

4. FastAPI application

Create app/main.py:

from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.ollama_client import generate, list_models, pull_model
from app.rate_limiter import rate_limit_middleware
from app.schemas import GenerateRequest, GenerateResponse, ModelInfo, PullRequest

app = FastAPI(
    title="Ollama FastAPI Backend",
    version="1.0.0",
    description="Production REST API wrapper for local Ollama inference",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET", "POST", "DELETE"],
    allow_headers=["*"],
)

app.middleware("http")(rate_limit_middleware)


@app.get("/health")
async def health():
    return {"status": "ok", "ollama_host": settings.ollama_host}


@app.post("/generate", response_model=GenerateResponse)
async def generate_text(body: GenerateRequest):
    model = body.model or settings.default_model
    try:
        result = await generate(body.prompt, model, stream=False)
    except Exception as exc:
        raise HTTPException(status_code=502, detail=f"Ollama error: {exc}") from exc

    return GenerateResponse(
        model=result.get("model", model),
        response=result.get("response", ""),
        prompt_tokens=result.get("prompt_eval_count"),
        completion_tokens=result.get("eval_count"),
        total_duration_ms=result.get("total_duration", 0) // 1_000_000,
    )


@app.post("/generate/stream")
async def generate_stream(body: GenerateRequest):
    model = body.model or settings.default_model

    async def event_stream():
        try:
            token_iter = await generate(body.prompt, model, stream=True)
            async for token in token_iter:
                yield f"data: {token}\n\n"
            yield "data: [DONE]\n\n"
        except Exception as exc:
            yield f"event: error\ndata: {exc}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")


@app.get("/models", response_model=list[ModelInfo])
async def get_models():
    try:
        models = await list_models()
    except Exception as exc:
        raise HTTPException(status_code=502, detail=f"Ollama error: {exc}") from exc
    return [
        ModelInfo(
            name=m["name"],
            size=m.get("size", 0),
            modified_at=m.get("modified_at", ""),
        )
        for m in models
    ]


@app.post("/models/pull")
async def pull_new_model(body: PullRequest):
    async def progress_stream():
        async for line in pull_model(body.name):
            yield f"data: {line}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(progress_stream(), media_type="text/event-stream")

Run it locally:

uvicorn app.main:app --reload --port 8000

Open http://localhost:8000/docs to see the automatically generated OpenAPI UI.

---

Streaming responses in depth

The /generate/stream endpoint uses Server-Sent Events (SSE). Each token arrives as a data: line; the connection closes with a data: [DONE] sentinel.

Consuming this from JavaScript:

const es = new EventSource("/generate/stream", { method: "POST" });
// Note: EventSource does not support POST natively.
// Use fetch + ReadableStream instead:

const response = await fetch("/generate/stream", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ prompt: "Explain async Python in 3 sentences." }),
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  const chunk = decoder.decode(value);
  const lines = chunk.split("\n\n").filter(Boolean);
  for (const line of lines) {
    const token = line.replace(/^data: /, "");
    if (token !== "[DONE]") process.stdout.write(token);
  }
}

Consuming from Python (e.g., another service calling your backend):

import httpx

async with httpx.AsyncClient() as client:
    async with client.stream(
        "POST",
        "http://localhost:8000/generate/stream",
        json={"prompt": "Explain ASGI in one paragraph."},
    ) as resp:
        async for line in resp.aiter_lines():
            if line.startswith("data: ") and not line.endswith("[DONE]"):
                print(line[6:], end="", flush=True)

---

Handling concurrent requests

FastAPI on ASGI (via uvicorn) is inherently concurrent: while one coroutine awaits an upstream Ollama response, the event loop services other incoming requests. However, there are two constraints to model correctly.

Ollama's default concurrency

Ollama 0.3+ supports --parallel inference. Check whether your running daemon has it enabled:

curl http://localhost:11434/api/generate \
  -d '{"model":"llama3.2","prompt":"ping","stream":false}' &
curl http://localhost:11434/api/generate \
  -d '{"model":"llama3.2","prompt":"pong","stream":false}' &
wait

If both return simultaneously, parallel inference is active. If the second request queues until the first completes, you need to either enable parallelism or add a semaphore in your FastAPI layer:

import asyncio

_inference_semaphore = asyncio.Semaphore(4)  # max 4 concurrent Ollama calls

@app.post("/generate", response_model=GenerateResponse)
async def generate_text(body: GenerateRequest):
    model = body.model or settings.default_model
    async with _inference_semaphore:
        try:
            result = await generate(body.prompt, model, stream=False)
        except Exception as exc:
            raise HTTPException(status_code=502, detail=f"Ollama error: {exc}") from exc
    ...

Set the semaphore bound to match your GPU's comfortable parallel batch size. A single NVIDIA RTX 4090 can typically handle 2–4 concurrent 7B-parameter streams before response latency degrades significantly.

Blocking CPU work

If you add synchronous preprocessing (tokenisation counts, embedding distance checks) before forwarding to Ollama, run it in a thread pool to avoid blocking the event loop:

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def preprocess(prompt: str) -> str:
    loop = asyncio.get_event_loop()
    # run_in_executor is non-blocking to the event loop
    return await loop.run_in_executor(executor, _sync_preprocess, prompt)

def _sync_preprocess(prompt: str) -> str:
    # Heavy CPU-bound sanitisation, embedding lookup, etc.
    return prompt.strip()

---

Ollama API vs OpenAI API compatibility

Ollama ships an OpenAI-compatible endpoint at /v1/chat/completions. This means any library that accepts an openai.OpenAI(base_url=..., api_key="ollama") configuration can use it as a drop-in replacement:

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama",          # Ollama ignores this but the library requires it
)

response = await client.chat.completions.create(
    model="llama3.2",
    messages=[{"role": "user", "content": "What is ASGI?"}],
    stream=False,
)
print(response.choices[0].message.content)

Feature	Ollama native API	Ollama /v1 OpenAI-compat
Single-turn generation	/api/generate	/v1/completions
Multi-turn chat	/api/chat	/v1/chat/completions
Embeddings	/api/embeddings	/v1/embeddings
Streaming	✅ native JSON lines	✅ SSE data: format
Function calling	❌ not supported	Limited (model-dependent)
Logprobs	❌	❌
Authentication	None (local only)	API key field (ignored)

The native API gives you slightly more metadata (duration, token counts, model digest) and direct control. Use the OpenAI-compatible layer when you want to reuse existing SDK integrations or swap in a remote OpenAI endpoint without code changes.

---

Adding Apify data as LLM context

A common production pattern is injecting fresh web-scraped content into the prompt at request time — a lightweight form of Retrieval-Augmented Generation. Apify provides ready-made scrapers for hundreds of sources (Google, Reddit, LinkedIn, news sites) that your FastAPI service can call before forwarding the enriched prompt to Ollama.

import httpx
from app.config import settings

APIFY_API_TOKEN = settings.apify_token  # add to your .env

async def fetch_web_context(query: str) -> str:
    """Run an Apify Google Search scraper and return the top 3 snippets."""
    async with httpx.AsyncClient() as client:
        run_resp = await client.post(
            "https://api.apify.com/v2/acts/apify~google-search-scraper/run-sync-get-dataset-items?fpr=use-apify",
            params={"token": APIFY_API_TOKEN},
            json={"queries": query, "maxPagesPerQuery": 1, "resultsPerPage": 3},
            timeout=30,
        )
        run_resp.raise_for_status()
        items = run_resp.json()
        snippets = [item.get("snippet", "") for item in items if item.get("snippet")]
        return "\n\n".join(snippets)

@app.post("/generate/grounded", response_model=GenerateResponse)
async def generate_grounded(body: GenerateRequest):
    """Generate a response grounded with live web context."""
    context = await fetch_web_context(body.prompt)
    enriched_prompt = (
        f"Use the following web search results as context:\n\n"
        f"{context}\n\n"
        f"Now answer the following question:\n{body.prompt}"
    )
    model = body.model or settings.default_model
    result = await generate(enriched_prompt, model, stream=False)
    return GenerateResponse(
        model=result.get("model", model),
        response=result.get("response", ""),
    )

For a full data ingestion architecture, the Scrape Websites for LLMs guide covers document chunking, cleaning, and vector storage in depth.

---

Deploying with Docker

Create Dockerfile:

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app/ ./app/
COPY .env .

EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

Create requirements.txt:

fastapi==0.115.0
uvicorn[standard]==0.30.6
httpx==0.27.2
pydantic==2.8.2
pydantic-settings==2.4.0
python-dotenv==1.0.1

Create docker-compose.yml to run both Ollama and the FastAPI service:

services:
  ollama:
    image: ollama/ollama:latest
    volumes:
      - ollama_data:/root/.ollama
    ports:
      - "11434:11434"
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: all
              capabilities: [gpu]

  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OLLAMA_HOST=http://ollama:11434
      - DEFAULT_MODEL=llama3.2
    depends_on:
      - ollama
    restart: unless-stopped

volumes:
  ollama_data:

Start the stack:

# Pull the model first (one-time)
docker compose run --rm ollama ollama pull llama3.2

# Start services
docker compose up -d

# Tail logs
docker compose logs -f api

GPU note: The deploy.resources block in docker-compose.yml requires the NVIDIA Container Toolkit. Remove that block for CPU-only deployments.

---

Deploying as a systemd service

For long-running bare-metal deployments on Linux VPS (recommended for production inference workloads — see Liquid Web dedicated servers), systemd provides automatic restarts, log management, and boot persistence.

Create /etc/systemd/system/ollama-api.service:

[Unit]
Description=Ollama FastAPI Backend
After=network.target ollama.service
Requires=ollama.service

[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/ollama-fastapi-backend
EnvironmentFile=/opt/ollama-fastapi-backend/.env
ExecStart=/opt/ollama-fastapi-backend/.venv/bin/uvicorn \
    app.main:app \
    --host 0.0.0.0 \
    --port 8000 \
    --workers 2 \
    --log-level info
Restart=on-failure
RestartSec=5s
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

Enable and start:

sudo systemctl daemon-reload
sudo systemctl enable ollama-api
sudo systemctl start ollama-api
sudo journalctl -u ollama-api -f   # stream logs

Place Nginx in front as a reverse proxy to handle TLS termination and add X-Forwarded-For headers:

server {
    listen 443 ssl;
    server_name api.yourdomain.com;

    ssl_certificate     /etc/letsencrypt/live/api.yourdomain.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/api.yourdomain.com/privkey.pem;

    location / {
        proxy_pass         http://127.0.0.1:8000;
        proxy_set_header   Host $host;
        proxy_set_header   X-Real-IP $remote_addr;
        proxy_set_header   X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_buffering    off;          # required for SSE streaming
        proxy_read_timeout 300s;         # long enough for slow inference
    }
}

proxy_buffering off is critical — without it, Nginx buffers the entire SSE response before forwarding, which breaks streaming.

---

Full project structure

ollama-fastapi-backend/
├── app/
│   ├── __init__.py
│   ├── config.py          # Settings via pydantic-settings
│   ├── main.py            # FastAPI routes
│   ├── ollama_client.py   # Async Ollama HTTP wrapper
│   ├── rate_limiter.py    # Sliding-window rate limiter
│   └── schemas.py         # Pydantic request/response models
├── .env
├── docker-compose.yml
├── Dockerfile
└── requirements.txt

---

FAQ

Frequently Asked Questions

Send HTTP POST requests to http://localhost:11434/api/generate with a JSON body containing 'model' and 'prompt' fields. Set 'stream': true to receive token-by-token responses as newline-delimited JSON. Use the /api/tags endpoint to list your downloaded models. No API key is required — Ollama runs entirely on your local machine.

Yes. Ollama exposes an OpenAI-compatible layer at http://localhost:11434/v1. You can use the official openai Python library by setting base_url='http://localhost:11434/v1' and api_key='ollama'. The /v1/chat/completions and /v1/embeddings endpoints are supported. Function calling is model-dependent and not universally available in open-weight models.

Use httpx.AsyncClient with client.stream() and iterate over resp.aiter_lines(). Each line is a JSON object with a 'response' key containing the next token. The stream closes when the object has 'done': true. In FastAPI, return a StreamingResponse with a generator that yields 'data: <token>\n\n' lines for SSE compatibility.

Ollama 0.3+ supports parallel inference via the --parallel flag. Without it, requests are queued. In FastAPI, use asyncio.Semaphore to cap concurrent upstream calls at a value your hardware can sustain. A single 24 GB GPU typically handles 2–4 concurrent 7B-parameter streams before latency degrades noticeably.

The two main options are Docker Compose (portable, good for multi-service stacks) and systemd (lower overhead, better suited for dedicated inference servers). In both cases, put Nginx or Caddy in front for TLS termination and set proxy_buffering off to preserve SSE streaming. For serious production workloads, a dedicated GPU VPS — such as a Liquid Web bare-metal server — gives you predictable latency without noisy-neighbour effects.