FastAPI AI Tutorial: Build LLM APIs with Ollama, Streaming, and Production Deployment for 2026

By Fadi Updated Jan 2, 2026 12 min read AI Backend

TL;DR — FastAPI AI tutorial: LLM integration, streaming, production. Ollama FastAPI Guide: "Ollama + FastAPI local LLM server with Docker production deployment. Install Ollama, pull model, create venv, FastAPI app with async endpoints." Ollama REST API FastAPI: "Semaphore for GPU concurrency — RTX 4090 handles 2-4 concurrent 7B streams. Enable parallelism or queue requests." FastAPI Ollama Demo: "Streaming, formatted, and complete JSON responses from AI models." Serve Local LLMs: "Ollama at localhost:11434, FastAPI with uvicorn, production deployment." Learn more with Ollama REST API, Ollama Docker, function calling, and Docker AI API.

Ollama FastAPI Guide introduces the setup: "Ollama + FastAPI Local LLM Server — Docker Production Guide. Install Ollama, pull model (llama3.2:3b is the lightest option), ollama serve. Create venv, install fastapi uvicorn httpx."

Ollama REST API FastAPI adds: "If the second request queues until the first completes, you need to either enable parallelism or add a semaphore in your FastAPI layer. 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."

FastAPI AI Architecture

flowchart TD subgraph Client["Client"] User["User Request
POST /v1/chat/completions
messages, model, stream"] end subgraph FastAPI["FastAPI Application"] Models["Pydantic Models
ChatRequest
ChatResponse
Message
EmbeddingRequest"] Endpoints["Async Endpoints
POST /v1/chat/completions
POST /v1/embeddings
GET /health
GET /v1/models"] Semaphore["Concurrency Control
asyncio.Semaphore
RTX 4090: 2-4 concurrent
queue excess requests"] Stream["Streaming Response
StreamingResponse
async generator
token-by-token
SSE format"] end subgraph Ollama["Ollama LLM Server"] API["REST API
localhost:11434
POST /api/chat
POST /api/embeddings
stream=true"] Models2["Models
llama3.2:3b
llama3.2:7b
qwen2.5:7b
GPU inference"] end subgraph Deploy["Docker Deployment"] Compose["Docker Compose
FastAPI service
Ollama service
shared network
GPU support"] Uvicorn["uvicorn
main:app
host 0.0.0.0
port 8000
workers 4"] Nginx["Nginx Reverse Proxy
SSL termination
load balancing
rate limiting"] end User --> Endpoints Models --> Endpoints Endpoints --> Semaphore Semaphore --> Stream Semaphore --> API API --> Models2 Stream --> User Compose --> Uvicorn Compose --> API Uvicorn --> Nginx Nginx --> User style FastAPI fill:#4169E1,color:#fff style Ollama fill:#39FF14,color:#000 style Deploy fill:#2D1B69,color:#fff

Endpoint Reference

Endpoint	Method	Purpose	Request	Response
`/v1/chat/completions`	POST	Chat completion	messages, model, temp, stream	ChatResponse or stream
`/v1/embeddings`	POST	Text embeddings	model, input	EmbeddingResponse
`/v1/models`	GET	List models	—	Model list
`/health`	GET	Health check	—	Status

Implementation

from dataclasses import dataclass
from typing import Optional
from enum import Enum

class EndpointType(Enum):
    CHAT = "chat"
    EMBEDDING = "embedding"
    STREAMING = "streaming"
    HEALTH = "health"

@dataclass
class FastAPIAIGuide:
    """FastAPI AI tutorial implementation guide."""

    def get_app_setup(self) -> str:
        """FastAPI app setup with Pydantic models."""
        return (
            "# === FASTAPI APP SETUP ===\n"
            "# pip install fastapi uvicorn httpx\n"
            "\n"
            "from fastapi import FastAPI,\n"
            "    HTTPException\n"
            "from fastapi.responses import\n"
            "    StreamingResponse\n"
            "from pydantic import BaseModel,\n"
            "    Field\n"
            "from typing import Optional\n"
            "import httpx\n"
            "import asyncio\n"
            "import json\n"
            "\n"
            "app = FastAPI(\n"
            "    title='AI API',\n"
            "    version='1.0.0')\n"
            "\n"
            "OLLAMA_URL =\n"
            "    'http://localhost:11434'\n"
            "\n"
            "# Concurrency control\n"
            "# RTX 4090: 2-4 concurrent\n"
            "#   7B streams\n"
            "semaphore = asyncio.Semaphore(\n"
            "    4)\n"
            "\n"
            "# === PYDANTIC MODELS ===\n"
            "class Message(BaseModel):\n"
            "    role: str\n"
            "    content: str\n"
            "\n"
            "class ChatRequest(BaseModel):\n"
            "    model: str = 'llama3.2'\n"
            "    messages: list[Message]\n"
            "    temperature: float = 0.7\n"
            "    max_tokens: Optional[int]\n"
            "      = None\n"
            "    stream: bool = False\n"
            "\n"
            "class Choice(BaseModel):\n"
            "    message: Message\n"
            "    finish_reason: str\n"
            "\n"
            "class Usage(BaseModel):\n"
            "    prompt_tokens: int\n"
            "    completion_tokens: int\n"
            "    total_tokens: int\n"
            "\n"
            "class ChatResponse(BaseModel):\n"
            "    id: str\n"
            "    model: str\n"
            "    choices: list[Choice]\n"
            "    usage: Usage\n"
            "\n"
            "class EmbeddingRequest(BaseModel):\n"
            "    model: str = 'nomic-embed-text'\n"
            "    input: str\n"
            "\n"
            "class EmbeddingResponse(BaseModel):\n"
            "    embedding: list[float]\n"
            "    model: str"
        )

    def get_chat_endpoint(self) -> str:
        """Chat completion endpoint."""
        return (
            "# === CHAT ENDPOINT ===\n"
            "\n"
            "@app.post('/v1/chat/completions')\n"
            "async def chat_completions(\n"
            "    req: ChatRequest):\n"
            "    '''OpenAI-compatible chat\n"
            "    completion endpoint.''' \n"
            "    \n"
            "    if req.stream:\n"
            "        return StreamingResponse(\n"
            "            stream_chat(req),\n"
            "            media_type=\n"
            "              'text/event-stream')\n"
            "    \n"
            "    async with semaphore:\n"
            "        async with httpx.AsyncClient(\n"
            "            timeout=300.0\n"
            "        ) as client:\n"
            "            resp = await client.post(\n"
            "                f'{OLLAMA_URL}/api/chat',\n"
            "                json={\n"
            "                    'model': req.model,\n"
            "                    'messages': [\n"
            "                      {'role': m.role,\n"
            "                       'content': m.content}\n"
            "                      for m in\n"
            "                        req.messages],\n"
            "                    'options': {\n"
            "                      'temperature':\n"
            "                        req.temperature,\n"
            "                    },\n"
            "                    'stream': False,\n"
            "                })\n"
            "            \n"
            "            if resp.status_code != 200:\n"
            "                raise HTTPException(\n"
            "                    resp.status_code,\n"
            "                    resp.text)\n"
            "            \n"
            "            data = resp.json()\n"
            "            return ChatResponse(\n"
            "                id=data.get('id',''),\n"
            "                model=data['model'],\n"
            "                choices=[{\n"
            "                    'message': {\n"
            "                      'role':'assistant',\n"
            "                      'content': data[\n"
            "                        'message']['content']},\n"
            "                    'finish_reason':'stop'}],\n"
            "                usage={\n"
            "                  'prompt_tokens':\n"
            "                    data.get(\n"
            "                    'prompt_eval_count',0),\n"
            "                  'completion_tokens':\n"
            "                    data.get(\n"
            "                    'eval_count',0),\n"
            "                  'total_tokens':\n"
            "                    data.get(\n"
            "                    'prompt_eval_count',0)\n"
            "                    + data.get(\n"
            "                    'eval_count',0),\n"
            "                })"
        )

    def get_streaming(self) -> str:
        """Streaming response endpoint."""
        return (
            "# === STREAMING ===\n"
            "\n"
            "async def stream_chat(req):\n"
            "    '''Stream LLM tokens as\n"
            "    Server-Sent Events.''' \n"
            "    \n"
            "    async with semaphore:\n"
            "        async with httpx.AsyncClient(\n"
            "            timeout=300.0\n"
            "        ) as client:\n"
            "            async with client.stream(\n"
            "                'POST',\n"
            "                f'{OLLAMA_URL}/api/chat',\n"
            "                json={\n"
            "                    'model': req.model,\n"
            "                    'messages': [\n"
            "                      {'role': m.role,\n"
            "                       'content': m.content}\n"
            "                      for m in\n"
            "                        req.messages],\n"
            "                    'options': {\n"
            "                      'temperature':\n"
            "                        req.temperature,\n"
            "                    },\n"
            "                    'stream': True,\n"
            "                }\n"
            "            ) as response:\n"
            "                async for line in (\n"
            "                    response\n"
            "                    .aiter_lines()):\n"
            "                    if line:\n"
            "                        chunk = (\n"
            "                            json.loads(line))\n"
            "                        if chunk.get('done'):\n"
            "                            yield (\n"
            "                              'data: [DONE]')\n"
            "                              '\\n\\n')\n"
            "                            break\n"
            "                        content = (\n"
            "                            chunk\n"
            "                            .get('message',{})\n"
            "                            .get('content',''))\n"
            "                        if content:\n"
            "                            sse_data = (\n"
            "                              json.dumps({\n"
            "                              'choices': [{\n"
            "                                'delta': {\n"
            "                                  'content':\n"
            "                                    content}}]}))\n"
            "                            yield (\n"
            "                              f'data: {sse_data}'\n"
            "                              '\\n\\n')"
        )

    def get_embeddings(self) -> str:
        """Embeddings endpoint."""
        return (
            "# === EMBEDDINGS ===\n"
            "\n"
            "@app.post('/v1/embeddings')\n"
            "async def create_embedding(\n"
            "    req: EmbeddingRequest):\n"
            "    '''Generate text embeddings.''' \n"
            "    \n"
            "    async with httpx.AsyncClient(\n"
            "        timeout=60.0\n"
            "    ) as client:\n"
            "        resp = await client.post(\n"
            "            f'{OLLAMA_URL}/api/embeddings',\n"
            "            json={\n"
            "                'model': req.model,\n"
            "                'prompt': req.input,\n"
            "            })\n"
            "        \n"
            "        if resp.status_code != 200:\n"
            "            raise HTTPException(\n"
            "                resp.status_code,\n"
            "                resp.text)\n"
            "        \n"
            "        data = resp.json()\n"
            "        return EmbeddingResponse(\n"
            "            embedding=data['embedding'],\n"
            "            model=req.model)"
        )

    def get_health_models(self) -> str:
        """Health check and models endpoints."""
        return (
            "# === HEALTH & MODELS ===\n"
            "\n"
            "@app.get('/health')\n"
            "async def health():\n"
            "    '''Health check endpoint.''' \n"
            "    try:\n"
            "        async with httpx.AsyncClient(\n"
            "            timeout=5.0\n"
            "        ) as client:\n"
            "            resp = await client.get(\n"
            "                f'{OLLAMA_URL}/api/tags')\n"
            "            if resp.status_code == 200:\n"
            "                return {'status':'healthy',\n"
            "                    'ollama':'connected'}\n"
            "    except Exception:\n"
            "        pass\n"
            "    return {'status':'degraded',\n"
            "        'ollama':'disconnected'}\n"
            "\n"
            "@app.get('/v1/models')\n"
            "async def list_models():\n"
            "    '''List available models.''' \n"
            "    async with httpx.AsyncClient(\n"
            "        timeout=10.0\n"
            "    ) as client:\n"
            "        resp = await client.get(\n"
            "            f'{OLLAMA_URL}/api/tags')\n"
            "        data = resp.json()\n"
            "        return {\n"
            "            'models': [{\n"
            "                'id': m['name'],\n"
            "                'name': m['name'],\n"
            "            } for m in data.get(\n"
            "                'models', [])]\n"
            "        }"
        )

    def get_docker(self) -> str:
        """Docker deployment configuration."""
        return (
            "# === DOCKER DEPLOYMENT ===\n"
            "\n"
            "# Dockerfile\n"
            "FROM python:3.12-slim\n"
            "WORKDIR /app\n"
            "COPY requirements.txt .\n"
            "RUN pip install --no-cache-dir\n"
            "  -r requirements.txt\n"
            "COPY . .\n"
            "EXPOSE 8000\n"
            "CMD ['uvicorn', 'main:app',\n"
            "  '--host', '0.0.0.0',\n"
            "  '--port', '8000',\n"
            "  '--workers', '4']\n"
            "\n"
            "# requirements.txt\n"
            "fastapi>=0.115.0\n"
            "uvicorn[standard]>=0.34.0\n"
            "httpx>=0.28.0\n"
            "pydantic>=2.0\n"
            "\n"
            "# docker-compose.yml\n"
            "version: '3.8'\n"
            "services:\n"
            "  api:\n"
            "    build: .\n"
            "    ports: ['8000:8000']\n"
            "    environment:\n"
            "      - OLLAMA_URL=http://ollama:11434\n"
            "    depends_on:\n"
            "      - ollama\n"
            "    restart: unless-stopped\n"
            "  \n"
            "  ollama:\n"
            "    image: ollama/ollama:latest\n"
            "    ports: ['11434:11434']\n"
            "    volumes:\n"
            "      - ollama_data:/root/.ollama\n"
            "    deploy:\n"
            "      resources:\n"
            "        reservations:\n"
            "          devices:\n"
            "            - capabilities: [gpu]\n"
            "    restart: unless-stopped\n"
            "\n"
            "volumes:\n"
            "  ollama_data:\n"
            "\n"
            "# Run\n"
            "# docker-compose up -d\n"
            "# API: http://localhost:8000\n"
            "# Docs: http://localhost:8000/docs"
        )

    def get_concurrency_tips(self) -> dict:
        """Concurrency and production tips."""
        return {
            "semaphore": "asyncio.Semaphore(4) for RTX 4090 — limits concurrent LLM requests to GPU capacity",
            "ollama_parallel": "OLLAMA_NUM_PARALLEL env var for Ollama-side parallelism",
            "queue_handling": "Queue excess requests until slot available — return 503 when queue full",
            "connection_pool": "httpx.AsyncClient with limits for connection pooling — reuse connections",
            "timeout": "Set timeout=300 for long LLM generation, timeout=5 for health checks",
            "workers": "uvicorn --workers 4 for multi-process — each worker has own semaphore",
            "background_tasks": "BackgroundTasks for long-running inference — return job ID, poll for result",
            "gpu_monitoring": "Monitor GPU utilization to tune semaphore — nvidia-smi, prometheus",
            "rate_limiting": "SlowAPI or custom middleware for per-client rate limiting",
            "openai_compat": "OpenAI-compatible format for drop-in replacement — /v1/chat/completions",
        }

FastAPI AI Checklist

[ ] FastAPI: modern, fast web framework for building APIs with Python — easy to use, fast to code, production-ready
[ ] Install: pip install fastapi uvicorn httpx
[ ] App setup: FastAPI(title='AI API', version='1.0.0')
[ ] Ollama URL: http://localhost:11434 — REST API for LLM inference
[ ] Pydantic models: ChatRequest, ChatResponse, Message, EmbeddingRequest, EmbeddingResponse
[ ] ChatRequest: model, messages (list of role/content), temperature, max_tokens, stream
[ ] ChatResponse: id, model, choices (message/finish_reason), usage (prompt/completion/total tokens)
[ ] OpenAI-compatible format: /v1/chat/completions for drop-in replacement
[ ] Async endpoints: async def with httpx.AsyncClient for non-blocking Ollama calls
[ ] Streaming: StreamingResponse with async generator yielding SSE chunks
[ ] Streaming: Ollama stream=true for token-by-token generation
[ ] Streaming: SSE format data: {json} with data: [DONE] terminator
[ ] Streaming: lower latency, better UX for chat interfaces
[ ] Concurrency: asyncio.Semaphore(4) for RTX 4090 — 2-4 concurrent 7B streams
[ ] Concurrency: semaphore limits concurrent LLM requests to GPU capacity
[ ] Concurrency: OLLAMA_NUM_PARALLEL env var for Ollama-side parallelism
[ ] Concurrency: queue excess requests until slot available, return 503 when full
[ ] Connection pool: httpx.AsyncClient with limits for connection reuse
[ ] Timeout: 300s for LLM generation, 5s for health checks, 60s for embeddings
[ ] Health check: GET /health — check Ollama connectivity via /api/tags
[ ] Models list: GET /v1/models — list available Ollama models
[ ] Embeddings: POST /v1/embeddings — call Ollama /api/embeddings
[ ] Error handling: HTTPException with status code and message
[ ] Docker: FROM python:3.12-slim, install requirements, uvicorn main:app
[ ] Docker: uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
[ ] Docker Compose: FastAPI + Ollama services, shared network, GPU support
[ ] Docker Compose: Ollama volume for model persistence, GPU device reservation
[ ] Docker Compose: OLLAMA_URL=http://ollama:11434 for container-to-container
[ ] Nginx: reverse proxy for SSL termination, load balancing, rate limiting
[ ] Workers: --workers 4 for multi-process — each worker has own semaphore
[ ] Background tasks: BackgroundTasks for long-running inference — return job ID
[ ] Rate limiting: SlowAPI or custom middleware for per-client limits
[ ] GPU monitoring: nvidia-smi, Prometheus for utilization tracking
[ ] Auto docs: FastAPI auto-generates OpenAPI/Swagger at /docs
[ ] Field validators: Pydantic Field validators for parameter bounds
[ ] Read Ollama REST API for Ollama endpoints
[ ] Read Ollama Docker for Ollama deployment
[ ] Read function calling for structured output
[ ] Read Docker AI API for deployment patterns
[ ] Test: chat endpoint returns correct response from Ollama
[ ] Test: streaming yields tokens incrementally
[ ] Test: semaphore limits concurrency under load
[ ] Test: health check detects Ollama connectivity
[ ] Test: embeddings endpoint returns vector
[ ] Test: Docker Compose services communicate correctly
[ ] Test: error handling for invalid model or timeout
[ ] Test: OpenAI-compatible format works with existing clients
[ ] Document endpoint specs, Pydantic models, concurrency config, Docker setup

FAQ

How do you integrate FastAPI with Ollama for LLM serving?

Use httpx or aiohttp to call Ollama REST API from FastAPI async endpoints. Ollama FastAPI Guide: "Install Ollama, pull model (llama3.2:3b), ollama serve. Create venv, install fastapi uvicorn httpx. FastAPI app with async endpoints calling Ollama at localhost:11434." Ollama REST API FastAPI: "If second request queues until first completes, enable parallelism or add semaphore in FastAPI layer. Set semaphore bound to match GPU comfortable parallel batch size. Single NVIDIA RTX 4090 can handle 2-4 concurrent 7B-parameter streams." FastAPI Ollama Demo: "Streaming, formatted, and complete JSON responses from AI models." Integration: (1) Ollama at localhost:11434. (2) FastAPI async endpoints with httpx.AsyncClient. (3) Pydantic models for request/response. (4) Streaming via StreamingResponse. (5) Semaphore for GPU concurrency control. (6) OpenAI-compatible endpoint format.

How do you stream LLM responses with FastAPI?

Use StreamingResponse with async generator that yields chunks from Ollama streaming API. Ollama FastAPI Demo: "Streaming, formatted, and complete JSON responses from AI models. FastAPI with Ollama for building AI-driven applications." FastAPI Ollama Guide: "FastAPI is modern, fast web framework for building APIs with Python. Easy to use, fast to code, ready for production." Streaming: (1) async def generate(): async with httpx.AsyncClient() as client: async with client.stream('POST', url, json=payload) as response: async for chunk in response.aiter_text(): yield chunk. (2) return StreamingResponse(generate(), media_type='text/plain'). (3) Ollama stream=true for token-by-token. (4) SSE format: data: {chunk}. (5) Client receives tokens as they generate. (6) Lower latency, better UX for chat.

How do you handle concurrency in FastAPI with LLM endpoints?

Use asyncio.Semaphore to limit concurrent LLM requests to match GPU capacity. Ollama REST API FastAPI: "If second request queues until first completes, enable parallelism or add semaphore in FastAPI layer. Set semaphore bound to match GPU comfortable parallel batch size. Single NVIDIA RTX 4090 can handle 2-4 concurrent 7B-parameter streams." Concurrency: (1) semaphore = asyncio.Semaphore(4) for RTX 4090. (2) async with semaphore: result = await call_ollama(). (3) OLLAMA_NUM_PARALLEL env var for Ollama-side parallelism. (4) Queue excess requests until slot available. (5) Monitor GPU utilization to tune semaphore. (6) Return 503 Service Unavailable when queue full. (7) Background tasks for long-running inference. (8) Connection pool with httpx.AsyncClient limits.

How do you deploy FastAPI AI applications with Docker?

Use Docker with uvicorn workers, Ollama sidecar, and docker-compose for multi-service. Ollama FastAPI Guide: "Docker production deployment with Ollama and FastAPI. Dockerfile with Python slim, install requirements, uvicorn command. Docker Compose with Ollama and FastAPI services." Serve Local LLMs: "Ollama at localhost:11434, FastAPI with uvicorn, production deployment." Docker: (1) Dockerfile: FROM python:3.12-slim, pip install fastapi uvicorn httpx, COPY app, CMD uvicorn main:app. (2) docker-compose: FastAPI + Ollama services, shared network, volumes for Ollama models. (3) uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4. (4) Ollama service with GPU support. (5) Environment variables for configuration. (6) Health check endpoints. (7) Nginx reverse proxy for production.

What Pydantic models should you use for FastAPI LLM endpoints?

Define request and response models with Pydantic for type-safe LLM API. Pydantic models: (1) ChatRequest: model, messages (list of role/content), temperature, max_tokens, stream. (2) ChatResponse: id, model, choices (list of message/finish_reason), usage (prompt_tokens/completion_tokens/total_tokens). (3) Message: role (system/user/assistant), content. (4) OpenAI-compatible format for drop-in replacement. (5) EmbeddingRequest: model, input. (6) EmbeddingResponse: embedding, model. (7) ValidationError handling with 422 response. (8) Field validators for parameter bounds. (9) Optional fields with defaults. (10) JSON schema auto-generated for OpenAPI docs.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →