An HTTP endpoint with FastAPI: your function, callable by any program

Why an endpoint

A Gradio page is for a person. When another program needs your function — a website's backend, a phone app, a colleague's script, a scheduled job on another machine — it needs an HTTP endpoint: a URL that accepts JSON and returns JSON. Module 6 taught you to call such endpoints. This is the other side.

FastAPI (free) is the framework that fits what you know, because its request and response models are pydantic — module 6's validation, now at the door of your own service.

The smallest service

# app.py
from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(title="Notes assistant")

class AskRequest(BaseModel):
    question: str = Field(min_length=1, max_length=2000)
    k: int = Field(default=5, ge=1, le=20)

class AskResponse(BaseModel):
    answer: str
    sources: list[str]

@app.post("/ask", response_model=AskResponse)
def ask(req: AskRequest) -> AskResponse:
    answer, hits = answer_from_notes(req.question, req.k)
    return AskResponse(answer=answer, sources=[c["doc"] for _, c in hits])

pip install fastapi uvicorn
uvicorn app:app --reload --port 8000

uvicorn is the server; app:app means the app object in app.py; --reload restarts on file changes during development. Now:

curl -X POST http://127.0.0.1:8000/ask \
     -H "content-type: application/json" \
     -d '{"question": "how do I rotate the key?"}'

returns JSON. And http://127.0.0.1:8000/docs shows an interactive page listing every route with its request and response schema, generated from the pydantic models — a form you can submit from the browser with no curl.

What the decorator does

@app.post("/ask") is module 7's decorator: it registers ask as the handler for POST /ask. The parameter annotation req: AskRequest tells FastAPI to read the request body as JSON, validate it against AskRequest, and pass the resulting object in. A body that fails validation never reaches your function; the client receives a 422 with a JSON list of errors naming each bad field — the same ValidationError from module 6, formatted for the caller. response_model validates what you return, so a bug that puts a None in sources becomes a 500 in your logs rather than malformed JSON in someone else's program.

Path parameters come from the URL, query parameters from ?k=3:

@app.get("/notes/{doc_id}")
def get_note(doc_id: str, full: bool = False):
    ...

doc_id is filled from the path; full from the query string, converted to bool. Type hints are doing real work throughout, which is what the type-checker lesson promised they could.

Errors you raise

from fastapi import HTTPException

if not hits:
    raise HTTPException(status_code=404, detail="nothing in the notes matches")

Choose codes as module 6 read them: 400 for a malformed request the validator did not catch, 404 for a thing not found, 429 if you rate-limit, 503 if the model behind you is down. A RuntimeError you do not catch becomes a 500 with no detail — correct, since the caller should not see your traceback, but log it with exc_info=True so you can.

Sync, async, and the mistake

FastAPI accepts both def and async def routes, and the difference is module 6's:

• A plain def route runs on a thread from a pool. Blocking calls inside — requests.post, sqlite, a local model — are fine; other requests proceed on other threads.
• An async def route runs on the event loop. Only awaitable calls belong inside — httpx.AsyncClient, AsyncOpenAI. A blocking requests.post inside an async def stalls the loop, and every request waits for it. Ten callers, four seconds each, forty seconds for everyone.

When in doubt, write def. Switch a route to async def only when everything it does is async, and the gain is a server that handles thousands of slow model calls concurrently on one process.

Startup, and the things that load once

The index matrix, the embedding model, the provider — load them once, not per request:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    app.state.index = load_index("index/")
    app.state.provider = OllamaProvider()
    yield
    # shutdown: close connections here

app = FastAPI(lifespan=lifespan)

Module 7's context manager, wrapping the whole life of the server. Routes reach app.state.index through request.app.state, or through FastAPI's dependency injection, which is the framework's own composition mechanism.

Streaming out

from fastapi.responses import StreamingResponse

@app.post("/ask/stream")
def ask_stream(req: AskRequest):
    return StreamingResponse(stream_answer(req.question), media_type="text/plain")

Return a generator and the client receives it as it is produced — module 6's stream, from the serving side. A browser or a requests call with stream=True reads it line by line.

Where the key goes, and who may call

The service holds the model API key, so the caller never needs one — that is often the point. The service is now the thing to protect. At minimum, require a token in a header and compare it in a dependency; bind to 127.0.0.1 behind a reverse proxy rather than to the world. Deploying, TLS, rate limiting and the rest are the shipping course. The Python you write does not change when it moves.

Try this now

Build /ask over your notes search, run it, and submit a question from /docs. Send a body with k: 50 and read the 422. Then declare the route async def while it still calls a synchronous provider, hit it from a thread pool of ten, and time it — then put def back.