FastAPI for AI services.

Why Wrap It in an API

Minimal App Anatomy

pip install fastapi uvicorn[standard]

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
def health() -> dict:
    return {"status": "ok"}

# run with: uvicorn main:app --reload
# visit http://localhost:8000/docs for auto-generated interactive API docs

Request/Response Models

Exactly Topic 4's Pydantic models, now describing the HTTP contract instead of the LLM's output shape — often the same models, reused directly.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ExtractRequest(BaseModel):
    text: str

class TaskResponse(BaseModel):
    title: str
    priority: str
    due_date: str | None
    confidence: float

@app.post("/extract-task", response_model=TaskResponse)
def extract(request: ExtractRequest) -> TaskResponse:
    task = extract_task(request.text)  # Topic 4's capstone function
    return TaskResponse(
        title=task.title, priority=task.priority.value,
        due_date=task.due_date, confidence=task.confidence,
    )

Streaming Endpoints (SSE)

Topic 2 Section 3 streamed tokens into a terminal with print(). Streaming them to an HTTP client uses the same idea over a persistent response — a StreamingResponse that yields chunks as they arrive from the provider.

from fastapi.responses import StreamingResponse
from openai import OpenAI

client = OpenAI()

@app.post("/chat/stream")
def chat_stream(request: ExtractRequest):
    def token_generator():
        stream = client.chat.completions.create(
            model="gpt-4.1-mini",
            messages=[{"role": "user", "content": request.text}],
            stream=True,
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f"data: {delta}\n\n"  # the SSE wire format from Topic 2 Section 3

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

Async Endpoints

Topic 1 Section 7 flagged this: sync streaming is fine for a script, but a service handling concurrent requests needs the async client, or one slow LLM call blocks every other request on the same worker.

from openai import OpenAI
client = OpenAI()

@app.post("/extract-task")
def extract(request: ExtractRequest):
    task = extract_task(request.text)  # sync call — blocks this worker
    return task

from openai import AsyncOpenAI
client = AsyncOpenAI()

@app.post("/extract-task")
async def extract(request: ExtractRequest):
    task = await extract_task_async(request.text)  # frees the worker while waiting
    return task

Dependency Injection

FastAPI's Depends() is conceptually close to Hilt/Dagger's constructor injection — declare what a route needs, and the framework provides it, rather than constructing it inline every time.

from fastapi import Depends
from functools import lru_cache

@lru_cache
def get_openai_client() -> AsyncOpenAI:
    return AsyncOpenAI()  # created once, reused across requests

@app.post("/extract-task")
async def extract(request: ExtractRequest,
                   client: AsyncOpenAI = Depends(get_openai_client)):
    ...

Errors & Status Codes

Topic 2 Section 7's retry logic still lives inside the route — the new piece is translating internal failures into the right HTTP status, so the client can react correctly instead of treating every failure the same way.

from fastapi import HTTPException

@app.post("/extract-task", response_model=TaskResponse)
async def extract(request: ExtractRequest,
                   client: AsyncOpenAI = Depends(get_openai_client)):
    if not request.text.strip():
        raise HTTPException(status_code=400, detail="text must not be empty")
    try:
        task = await extract_task_async(request.text, client)
    except RateLimitError:
        raise HTTPException(status_code=503, detail="upstream provider is rate limiting us")
    except Exception:
        raise HTTPException(status_code=502, detail="extraction failed")
    return task

CORS & Calling from Android

Android clients aren't affected by CORS — that's a browser-only restriction, so a Kotlin/Retrofit client just needs the URL and, typically, an auth header your server checks. A web client calling this same API from a browser does need CORS configured.

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourapp.com"],  # never "*" once real users hit this
    allow_methods=["POST"],
    allow_headers=["*"],
)

// Retrofit interface on the Android side — the client's view of Section 2's contract
interface AiService {
    @POST("extract-task")
    suspend fun extractTask(@Body request: ExtractRequest): TaskResponse
}

data class ExtractRequest(val text: String)
data class TaskResponse(val title: String, val priority: String,
                         val dueDate: String?, val confidence: Double)

Capstone: A Real Task Extraction Service

Combine every section: async endpoint, DI, proper error handling, and a streaming variant, wrapping Topic 4's extractor.

from functools import lru_cache
from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
from pydantic import BaseModel

app = FastAPI()

class ExtractRequest(BaseModel):
    text: str

class TaskResponse(BaseModel):
    title: str
    priority: str
    due_date: str | None
    confidence: float

@lru_cache
def get_client() -> AsyncOpenAI:
    return AsyncOpenAI()

@app.get("/health")
def health() -> dict:
    return {"status": "ok"}

@app.post("/extract-task", response_model=TaskResponse)
async def extract(request: ExtractRequest,
                   client: AsyncOpenAI = Depends(get_client)) -> TaskResponse:
    if not request.text.strip():
        raise HTTPException(status_code=400, detail="text must not be empty")
    try:
        completion = await client.beta.chat.completions.parse(
            model="gpt-4.1-mini",
            messages=[{"role": "user", "content": request.text}],
            response_format=Task,  # Topic 4's Pydantic model
            temperature=0,
        )
    except Exception:
        raise HTTPException(status_code=502, detail="extraction failed")

    task = completion.choices[0].message.parsed
    if task is None:
        raise HTTPException(status_code=502, detail="model returned no result")
    return TaskResponse(title=task.title, priority=task.priority.value,
                         due_date=task.due_date, confidence=task.confidence)

@app.post("/chat/stream")
async def chat_stream(request: ExtractRequest,
                       client: AsyncOpenAI = Depends(get_client)):
    async def token_generator():
        stream = await client.chat.completions.create(
            model="gpt-4.1-mini",
            messages=[{"role": "user", "content": request.text}],
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f"data: {delta}\n\n"

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

# run with: uvicorn main:app --reload